


"""simple docstring"""
from typing import List, Optional, Tuple, Union

import torch
from torch import nn
from torch.nn import CrossEntropyLoss

from ... import AutoBackbone
from ...modeling_outputs import SemanticSegmenterOutput
from ...modeling_utils import PreTrainedModel
from ...utils import add_start_docstrings, add_start_docstrings_to_model_forward, replace_return_docstrings
from ...utils.backbone_utils import BackboneMixin
from .configuration_upernet import UperNetConfig


__SCREAMING_SNAKE_CASE							:			Optional[int]						=			[
    "openmmlab/upernet-convnext-tiny",
    # See all UperNet models at https://huggingface.co/models?filter=upernet
]

# General docstring
__SCREAMING_SNAKE_CASE							:			int						=			"UperNetConfig"





class    __A   (nn.Module):





							'''simple docstring'''

							def __init__( self							:					Tuple						,				UpperCAmelCase_							:					int						,				UpperCAmelCase_							:					int						,				UpperCAmelCase_							:					Union[int, Tuple[int, int]]						,				UpperCAmelCase_							:					Union[int, Tuple[int, int], str] = 0						,				UpperCAmelCase_							:					bool = False						,				UpperCAmelCase_							:					Union[int, Tuple[int, int]] = 1						,				)    ->None:



														"""simple docstring"""



														super().__init__()
														snake_case_							=						nn.Convad(
														    in_channels=UpperCAmelCase_						,				out_channels=UpperCAmelCase_						,				kernel_size=UpperCAmelCase_						,				padding=UpperCAmelCase_						,				bias=UpperCAmelCase_						,				dilation=UpperCAmelCase_						,				)
														snake_case_							=						nn.BatchNormad(UpperCAmelCase_				)
														snake_case_							=						nn.ReLU()



							def 						lowerCAmelCase				( self							:					Tuple						,				UpperCAmelCase_							:					torch.Tensor				)    ->torch.Tensor:



														"""simple docstring"""



														snake_case_							=						self.conv(UpperCAmelCase_				)
														snake_case_							=						self.batch_norm(UpperCAmelCase_				)
														snake_case_							=						self.activation(UpperCAmelCase_				)

														return output





class    __A   (nn.Module):





							'''simple docstring'''

							def __init__( self							:					List[str]						,				UpperCAmelCase_							:					int						,				UpperCAmelCase_							:					int						,				UpperCAmelCase_							:					int				)    ->None:



														"""simple docstring"""



														super().__init__()
														snake_case_							=						[
														    nn.AdaptiveAvgPoolad(UpperCAmelCase_				),
														    UperNetConvModule(UpperCAmelCase_						,				UpperCAmelCase_						,				kernel_size=1				),
														]
														for i, layer in enumerate(self.layers				):
																					self.add_module(str(UpperCAmelCase_				)						,				UpperCAmelCase_				)



							def 						lowerCAmelCase				( self							:					List[str]						,				UpperCAmelCase_							:					torch.Tensor				)    ->torch.Tensor:



														"""simple docstring"""



														snake_case_							=						input
														for layer in self.layers:
																					snake_case_							=						layer(UpperCAmelCase_				)
														return hidden_state





class    __A   (nn.Module):





							'''simple docstring'''

							def __init__( self							:					Optional[int]						,				UpperCAmelCase_							:					Tuple[int, ...]						,				UpperCAmelCase_							:					int						,				UpperCAmelCase_							:					int						,				UpperCAmelCase_							:					bool				)    ->None:



														"""simple docstring"""



														super().__init__()
														snake_case_							=						pool_scales
														snake_case_							=						align_corners
														snake_case_							=						in_channels
														snake_case_							=						channels
														snake_case_							=						[]
														for i, pool_scale in enumerate(UpperCAmelCase_				):
																					snake_case_							=						UperNetPyramidPoolingBlock(pool_scale=UpperCAmelCase_						,				in_channels=UpperCAmelCase_						,				channels=UpperCAmelCase_				)
																					self.blocks.append(UpperCAmelCase_				)
																					self.add_module(str(UpperCAmelCase_				)						,				UpperCAmelCase_				)



							def 						lowerCAmelCase				( self							:					Optional[Any]						,				UpperCAmelCase_							:					torch.Tensor				)    ->List[torch.Tensor]:



														"""simple docstring"""



														snake_case_							=						[]
														for ppm in self.blocks:
																					snake_case_							=						ppm(UpperCAmelCase_				)
																					snake_case_							=						nn.functional.interpolate(
																					    UpperCAmelCase_						,				size=x.size()[2:]						,				mode="""bilinear"""						,				align_corners=self.align_corners				)
																					ppm_outs.append(UpperCAmelCase_				)
														return ppm_outs





class    __A   (nn.Module):





							'''simple docstring'''

							def __init__( self							:					int						,				UpperCAmelCase_							:					List[str]						,				UpperCAmelCase_							:					List[Any]				)    ->Dict:



														"""simple docstring"""



														super().__init__()

														snake_case_							=						config
														snake_case_							=						config.pool_scales  # e.g. (1, 2, 3, 6)
														snake_case_							=						in_channels
														snake_case_							=						config.hidden_size
														snake_case_							=						False
														snake_case_							=						nn.Convad(self.channels						,				config.num_labels						,				kernel_size=1				)

														# PSP Module
														snake_case_							=						UperNetPyramidPoolingModule(
														    self.pool_scales						,				self.in_channels[-1]						,				self.channels						,				align_corners=self.align_corners						,				)
														snake_case_							=						UperNetConvModule(
														    self.in_channels[-1] + len(self.pool_scales				) * self.channels						,				self.channels						,				kernel_size=3						,				padding=1						,				)
														# FPN Module
														snake_case_							=						nn.ModuleList()
														snake_case_							=						nn.ModuleList()
														for in_channels in self.in_channels[:-1]:  # skip the top layer
																					snake_case_							=						UperNetConvModule(UpperCAmelCase_						,				self.channels						,				kernel_size=1				)
																					snake_case_							=						UperNetConvModule(self.channels						,				self.channels						,				kernel_size=3						,				padding=1				)
																					self.lateral_convs.append(UpperCAmelCase_				)
																					self.fpn_convs.append(UpperCAmelCase_				)

														snake_case_							=						UperNetConvModule(
														    len(self.in_channels				) * self.channels						,				self.channels						,				kernel_size=3						,				padding=1						,				)



							def 						lowerCAmelCase				( self							:					List[Any]				)    ->Any:



														"""simple docstring"""



														self.apply(self._init_weights				)



							def 						lowerCAmelCase				( self							:					Optional[int]						,				UpperCAmelCase_							:					str				)    ->List[Any]:



														"""simple docstring"""



														if isinstance(UpperCAmelCase_						,				nn.Convad				):
																					module.weight.data.normal_(mean=0.0						,				std=self.config.initializer_range				)
																					if module.bias is not None:
																												module.bias.data.zero_()



							def 						lowerCAmelCase				( self							:					List[str]						,				UpperCAmelCase_							:					Optional[Any]				)    ->Any:



														"""simple docstring"""



														snake_case_							=						inputs[-1]
														snake_case_							=						[x]
														psp_outs.extend(self.psp_modules(UpperCAmelCase_				)				)
														snake_case_							=						torch.cat(UpperCAmelCase_						,				dim=1				)
														snake_case_							=						self.bottleneck(UpperCAmelCase_				)

														return output



							def 						lowerCAmelCase				( self							:					Dict						,				UpperCAmelCase_							:					torch.Tensor				)    ->torch.Tensor:



														"""simple docstring"""



														snake_case_							=						[lateral_conv(encoder_hidden_states[i]				) for i, lateral_conv in enumerate(self.lateral_convs				)]

														laterals.append(self.psp_forward(UpperCAmelCase_				)				)

														# build top-down path
														snake_case_							=						len(UpperCAmelCase_				)
														for i in range(used_backbone_levels - 1						,				0						,				-1				):
																					snake_case_							=						laterals[i - 1].shape[2:]
																					snake_case_							=						laterals[i - 1] + nn.functional.interpolate(
																					    laterals[i]						,				size=UpperCAmelCase_						,				mode="""bilinear"""						,				align_corners=self.align_corners				)

														# build outputs
														snake_case_							=						[self.fpn_convs[i](laterals[i]				) for i in range(used_backbone_levels - 1				)]
														# append psp feature
														fpn_outs.append(laterals[-1]				)

														for i in range(used_backbone_levels - 1						,				0						,				-1				):
																					snake_case_							=						nn.functional.interpolate(
																					    fpn_outs[i]						,				size=fpn_outs[0].shape[2:]						,				mode="""bilinear"""						,				align_corners=self.align_corners				)
														snake_case_							=						torch.cat(UpperCAmelCase_						,				dim=1				)
														snake_case_							=						self.fpn_bottleneck(UpperCAmelCase_				)
														snake_case_							=						self.classifier(UpperCAmelCase_				)

														return output





class    __A   (nn.Module):





							'''simple docstring'''

							def __init__( self							:					List[str]						,				UpperCAmelCase_							:					Dict						,				UpperCAmelCase_							:					int = 2						,				UpperCAmelCase_							:					int = 3						,				UpperCAmelCase_							:					Union[int, Tuple[int, int]] = 1				)    ->None:



														"""simple docstring"""



														super().__init__()

														snake_case_							=						config
														snake_case_							=						config.auxiliary_in_channels
														snake_case_							=						config.auxiliary_channels
														snake_case_							=						config.auxiliary_num_convs
														snake_case_							=						config.auxiliary_concat_input
														snake_case_							=						in_index

														snake_case_							=						(kernel_size // 2) * dilation
														snake_case_							=						[]
														convs.append(
														    UperNetConvModule(
														        self.in_channels						,				self.channels						,				kernel_size=UpperCAmelCase_						,				padding=UpperCAmelCase_						,				dilation=UpperCAmelCase_				)				)
														for i in range(self.num_convs - 1				):
																					convs.append(
																					    UperNetConvModule(
																					        self.channels						,				self.channels						,				kernel_size=UpperCAmelCase_						,				padding=UpperCAmelCase_						,				dilation=UpperCAmelCase_				)				)
														if self.num_convs == 0:
																					snake_case_							=						nn.Identity()
														else:
																					snake_case_							=						nn.Sequential(*UpperCAmelCase_				)
														if self.concat_input:
																					snake_case_							=						UperNetConvModule(
																					    self.in_channels + self.channels						,				self.channels						,				kernel_size=UpperCAmelCase_						,				padding=kernel_size // 2				)

														snake_case_							=						nn.Convad(self.channels						,				config.num_labels						,				kernel_size=1				)



							def 						lowerCAmelCase				( self							:					int				)    ->List[Any]:



														"""simple docstring"""



														self.apply(self._init_weights				)



							def 						lowerCAmelCase				( self							:					Union[str, Any]						,				UpperCAmelCase_							:					Optional[Any]				)    ->Optional[Any]:



														"""simple docstring"""



														if isinstance(UpperCAmelCase_						,				nn.Convad				):
																					module.weight.data.normal_(mean=0.0						,				std=self.config.initializer_range				)
																					if module.bias is not None:
																												module.bias.data.zero_()



							def 						lowerCAmelCase				( self							:					Any						,				UpperCAmelCase_							:					torch.Tensor				)    ->torch.Tensor:



														"""simple docstring"""



														snake_case_							=						encoder_hidden_states[self.in_index]
														snake_case_							=						self.convs(UpperCAmelCase_				)
														if self.concat_input:
																					snake_case_							=						self.conv_cat(torch.cat([hidden_states, output]						,				dim=1				)				)
														snake_case_							=						self.classifier(UpperCAmelCase_				)
														return output





class    __A   (_SCREAMING_SNAKE_CASE):





							'''simple docstring'''

							__lowercase:						List[Any]												=			UperNetConfig
							__lowercase:						Optional[int]												=			"""pixel_values"""
							__lowercase:						Any												=			True



							def 						lowerCAmelCase				( self							:					List[str]						,				UpperCAmelCase_							:					Dict				)    ->Optional[Any]:



														"""simple docstring"""



														if isinstance(UpperCAmelCase_						,				UpperCAmelCase_				):
																					module.backbone.init_weights()
																					module.decode_head.init_weights()
																					module.auxiliary_head.init_weights()



							def 						lowerCAmelCase				( self							:					Optional[int]				)    ->int:



														"""simple docstring"""



														self.backbone.init_weights()
														self.decode_head.init_weights()
														self.auxiliary_head.init_weights()



							def 						lowerCAmelCase				( self							:					Union[str, Any]						,				UpperCAmelCase_							:					List[str]						,				UpperCAmelCase_							:					Any=False				)    ->Optional[int]:



														"""simple docstring"""



														if isinstance(UpperCAmelCase_						,				UpperCAmelCase_				):
																					snake_case_							=						value


__SCREAMING_SNAKE_CASE							:			Union[str, Any]						=			R"\n    Parameters:\n    This model is a PyTorch [torch.nn.Module](https://pytorch.org/docs/stable/nn.html#torch.nn.Module) sub-class. Use\n    it as a regular PyTorch Module and refer to the PyTorch documentation for all matter related to general usage and\n    behavior.\n        config ([`UperNetConfig`]): Model configuration class with all the parameters of the model.\n            Initializing with a config file does not load the weights associated with the model, only the\n            configuration. Check out the [`~PreTrainedModel.from_pretrained`] method to load the model weights.\n"

__SCREAMING_SNAKE_CASE							:			str						=			R"\n    Args:\n        pixel_values (`torch.FloatTensor` of shape `(batch_size, num_channels, height, width)`):\n            Pixel values. Padding will be ignored by default should you provide it. Pixel values can be obtained using\n            [`AutoImageProcessor`]. See [`SegformerImageProcessor.__call__`] for details.\n        output_attentions (`bool`, *optional*):\n            Whether or not to return the attentions tensors of all attention layers in case the backbone has them. See\n            `attentions` under returned tensors for more detail.\n        output_hidden_states (`bool`, *optional*):\n            Whether or not to return the hidden states of all layers of the backbone. See `hidden_states` under\n            returned tensors for more detail.\n        return_dict (`bool`, *optional*):\n            Whether or not to return a [`~utils.ModelOutput`] instead of a plain tuple.\n"





@add_start_docstrings(
    """UperNet framework leveraging any vision backbone e.g. for ADE20k, CityScapes."""  ,						_SCREAMING_SNAKE_CASE  ,						)
class    __A   (_SCREAMING_SNAKE_CASE):





							'''simple docstring'''

							def __init__( self							:					Any						,				UpperCAmelCase_							:					List[Any]				)    ->Union[str, Any]:



														"""simple docstring"""



														super().__init__(UpperCAmelCase_				)

														snake_case_							=						AutoBackbone.from_config(config.backbone_config				)

														# Semantic segmentation head(s)
														snake_case_							=						UperNetHead(UpperCAmelCase_						,				in_channels=self.backbone.channels				)
														snake_case_							=						UperNetFCNHead(UpperCAmelCase_				) if config.use_auxiliary_head else None

														# Initialize weights and apply final processing
														self.post_init()



							@add_start_docstrings_to_model_forward(UPERNET_INPUTS_DOCSTRING.format("""batch_size, sequence_length"""				)				)
							@replace_return_docstrings(output_type=UpperCAmelCase_						,				config_class=_CONFIG_FOR_DOC				)
							def 						lowerCAmelCase				( self							:					Any						,				UpperCAmelCase_							:					Optional[torch.Tensor] = None						,				UpperCAmelCase_							:					Optional[bool] = None						,				UpperCAmelCase_							:					Optional[bool] = None						,				UpperCAmelCase_							:					Optional[torch.Tensor] = None						,				UpperCAmelCase_							:					Optional[bool] = None						,				)    ->Union[tuple, SemanticSegmenterOutput]:



														"""simple docstring"""



														snake_case_							=						return_dict if return_dict is not None else self.config.use_return_dict
														snake_case_							=						(
														    output_hidden_states if output_hidden_states is not None else self.config.output_hidden_states
														)
														snake_case_							=						output_attentions if output_attentions is not None else self.config.output_attentions

														snake_case_							=						self.backbone.forward_with_filtered_kwargs(
														    UpperCAmelCase_						,				output_hidden_states=UpperCAmelCase_						,				output_attentions=UpperCAmelCase_				)
														snake_case_							=						outputs.feature_maps

														snake_case_							=						self.decode_head(UpperCAmelCase_				)
														snake_case_							=						nn.functional.interpolate(UpperCAmelCase_						,				size=pixel_values.shape[2:]						,				mode="""bilinear"""						,				align_corners=UpperCAmelCase_				)

														snake_case_							=						None
														if self.auxiliary_head is not None:
																					snake_case_							=						self.auxiliary_head(UpperCAmelCase_				)
																					snake_case_							=						nn.functional.interpolate(
																					    UpperCAmelCase_						,				size=pixel_values.shape[2:]						,				mode="""bilinear"""						,				align_corners=UpperCAmelCase_				)

														snake_case_							=						None
														if labels is not None:
																					if self.config.num_labels == 1:
																												raise ValueError("""The number of labels should be greater than one"""				)
																					else:
																												# compute weighted loss
																												snake_case_							=						CrossEntropyLoss(ignore_index=self.config.loss_ignore_index				)
																												snake_case_							=						loss_fct(UpperCAmelCase_						,				UpperCAmelCase_				)
																												snake_case_							=						loss_fct(UpperCAmelCase_						,				UpperCAmelCase_				)
																												snake_case_							=						main_loss + self.config.auxiliary_loss_weight * auxiliary_loss

														if not return_dict:
																					if output_hidden_states:
																												snake_case_							=						(logits,) + outputs[1:]
																					else:
																												snake_case_							=						(logits,) + outputs[2:]
																					return ((loss,) + output) if loss is not None else output

														return SemanticSegmenterOutput(
														    loss=UpperCAmelCase_						,				logits=UpperCAmelCase_						,				hidden_states=outputs.hidden_states						,				attentions=outputs.attentions						,				)







'''simple docstring'''






import os
import posixpath
import uuid
from dataclasses import dataclass
from typing import TYPE_CHECKING, Iterable, List, Optional, Tuple, Union

import numpy as np
import pyarrow as pa

import datasets
from datasets.arrow_writer import ArrowWriter, ParquetWriter
from datasets.config import MAX_SHARD_SIZE
from datasets.filesystems import (
    is_remote_filesystem,
    rename,
)
from datasets.iterable_dataset import _BaseExamplesIterable
from datasets.utils.py_utils import convert_file_size_to_int


__A					:  Optional[Any]											=      datasets.utils.logging.get_logger(__name__)

if TYPE_CHECKING:
  import pyspark






@dataclass
class    __snake_case       (   datasets.BuilderConfig):







    """simple docstring"""



    lowercase    =					None







def 				UpperCamelCase_			(    A__					:   "pyspark.sql.DataFrame"							,    A__					:   List[int]							,    ):
  '''simple docstring'''





  import pyspark

  def generate_fn():
    lowerCAmelCase_			:      Union[str, Any]			=   df.select("""*"""							,    pyspark.sql.functions.spark_partition_id().alias("""part_id"""					)					)
    for partition_id in partition_order:
      lowerCAmelCase_			:      int			=   df_with_partition_id.select("""*"""					).where(f'part_id = {partition_id}'					).drop("""part_id"""					)
      lowerCAmelCase_			:      Union[str, Any]			=   partition_df.collect()
      lowerCAmelCase_			:      str			=   0
      for row in rows:
        yield f'{partition_id}_{row_id}', row.asDict()
        row_id += 1

  return generate_fn






class    __snake_case       (   _BaseExamplesIterable):







    """simple docstring"""



    def __init__(   self    :  Optional[int]    ,						lowerCamelCase    :  "pyspark.sql.DataFrame"    ,						lowerCamelCase    :  List[Any]=None    ,						)   ->  Optional[Any]:
      lowerCAmelCase_			:      str			=   df
      lowerCAmelCase_			:      List[Any]			=   partition_order or range(self.df.rdd.getNumPartitions()		)
      lowerCAmelCase_			:      List[str]			=   _generate_iterable_examples(self.df    ,						self.partition_order		)

    def __iter__(   self    :  Dict		)   ->  int:
      yield from self.generate_examples_fn()

    def __lowercase   (   self    :  Any    ,						lowerCamelCase    :  np.random.Generator		)   ->  "SparkExamplesIterable":
      lowerCAmelCase_			:      Optional[int]			=   list(range(self.df.rdd.getNumPartitions()		)		)
      generator.shuffle(lowerCamelCase		)
      return SparkExamplesIterable(self.df    ,						partition_order=lowerCamelCase		)

    def __lowercase   (   self    :  Optional[Any]    ,						lowerCamelCase    :  int    ,						lowerCamelCase    :  int		)   ->  "SparkExamplesIterable":
      lowerCAmelCase_			:      Tuple			=   self.split_shard_indices_by_worker(lowerCamelCase    ,						lowerCamelCase		)
      return SparkExamplesIterable(self.df    ,						partition_order=lowerCamelCase		)





    @property
    def __lowercase   (   self    :  str		)   ->  int:
      return len(self.partition_order		)







class    __snake_case       (   datasets.DatasetBuilder):







    """simple docstring"""



    lowercase    =					SparkConfig

    def __init__(   self    :  Union[str, Any]    ,						lowerCamelCase    :  "pyspark.sql.DataFrame"    ,						lowerCamelCase    :  str = None    ,						lowerCamelCase    :  str = None    ,						**lowerCamelCase    :  int    ,						)   ->  Tuple:
      import pyspark

      lowerCAmelCase_			:      Dict			=   pyspark.sql.SparkSession.builder.getOrCreate()
      lowerCAmelCase_			:      List[str]			=   df
      lowerCAmelCase_			:      List[Any]			=   working_dir

      super().__init__(
          cache_dir=lowerCamelCase    ,						config_name=str(self.df.semanticHash()		)    ,						**lowerCamelCase    ,						)

    def __lowercase   (   self    :  Tuple		)   ->  Union[str, Any]:
      # Returns the path of the created file.
      def create_cache_and_write_probe(lowerCamelCase    :  int		):
        # makedirs with exist_ok will recursively create the directory. It will not throw an error if directories
        # already exist.
        os.makedirs(self._cache_dir    ,						exist_ok=lowerCamelCase		)
        lowerCAmelCase_			:      List[str]			=   os.path.join(self._cache_dir    ,						"""fs_test""" + uuid.uuida().hex		)
        # Opening the file in append mode will create a new file unless it already exists, in which case it will not
        # change the file contents.
        open(lowerCamelCase    ,						"""a"""		)
        return [probe_file]

      if self._spark.conf.get("""spark.master"""    ,						""""""		).startswith("""local"""		):
        return

      # If the cluster is multi-node, make sure that the user provided a cache_dir and that it is on an NFS
      # accessible to the driver.
      # TODO: Stream batches to the driver using ArrowCollectSerializer instead of throwing an error.
      if self._cache_dir:
        lowerCAmelCase_			:      str			=   (
            self._spark.sparkContext.parallelize(range(1		)    ,						1		).mapPartitions(lowerCamelCase		).collect()
        )
        if os.path.isfile(probe[0]		):
          return

      raise ValueError(
          """When using Dataset.from_spark on a multi-node cluster, the driver and all workers should be able to access cache_dir"""		)

    def __lowercase   (   self    :  Any		)   ->  List[Any]:
      return datasets.DatasetInfo(features=self.config.features		)

    def __lowercase   (   self    :  str    ,						lowerCamelCase    :  datasets.download.download_manager.DownloadManager		)   ->  Optional[int]:
      return [datasets.SplitGenerator(name=datasets.Split.TRAIN		)]

    def __lowercase   (   self    :  Any    ,						lowerCamelCase    :  Tuple		)   ->  Optional[Any]:
      import pyspark

      def get_arrow_batch_size(lowerCamelCase    :  Any		):
        for batch in it:
          yield pa.RecordBatch.from_pydict({"""batch_bytes""": [batch.nbytes]}		)

      lowerCAmelCase_			:      str			=   self.df.count()
      lowerCAmelCase_			:      Any			=   df_num_rows if df_num_rows <= 1_00 else 1_00
      # Approximate the size of each row (in Arrow format) by averaging over a max-100-row sample.
      lowerCAmelCase_			:      str			=   (
          self.df.limit(lowerCamelCase		)
          .repartition(1		)
          .mapInArrow(lowerCamelCase    ,						"""batch_bytes: long"""		)
          .agg(pyspark.sql.functions.sum("""batch_bytes"""		).alias("""sample_bytes"""		)		)
          .collect()[0]
          .sample_bytes
          / sample_num_rows
      )
      lowerCAmelCase_			:      List[Any]			=   approx_bytes_per_row * df_num_rows
      if approx_total_size > max_shard_size:
        # Make sure there is at least one row per partition.
        lowerCAmelCase_			:      Tuple			=   min(lowerCamelCase    ,						int(approx_total_size / max_shard_size		)		)
        lowerCAmelCase_			:      Dict			=   self.df.repartition(lowerCamelCase		)

    def __lowercase   (   self    :  Optional[int]    ,						lowerCamelCase    :  str    ,						lowerCamelCase    :  str    ,						lowerCamelCase    :  int    ,						)   ->  Iterable[Tuple[int, bool, Union[int, tuple]]]:
      import pyspark

      lowerCAmelCase_			:      List[str]			=   ParquetWriter if file_format == """parquet""" else ArrowWriter
      lowerCAmelCase_			:      int			=   os.path.join(self._working_dir    ,						os.path.basename(lowerCamelCase		)		) if self._working_dir else fpath
      lowerCAmelCase_			:      Tuple			=   file_format == """parquet"""

      # Define these so that we don't reference self in write_arrow, which will result in a pickling error due to
      # pickling the SparkContext.
      lowerCAmelCase_			:      str			=   self.config.features
      lowerCAmelCase_			:      Tuple			=   self._writer_batch_size
      lowerCAmelCase_			:      Any			=   self._fs.storage_options

      def write_arrow(lowerCamelCase    :  Optional[int]		):
        # Within the same SparkContext, no two task attempts will share the same attempt ID.
        lowerCAmelCase_			:      Union[str, Any]			=   pyspark.TaskContext().taskAttemptId()
        lowerCAmelCase_			:      Union[str, Any]			=   next(lowerCamelCase    ,						lowerCamelCase		)
        if first_batch is None:
          # Some partitions might not receive any data.
          return pa.RecordBatch.from_arrays(
              [[task_id], [0], [0]]    ,						names=["""task_id""", """num_examples""", """num_bytes"""]    ,						)
        lowerCAmelCase_			:      List[Any]			=   0
        lowerCAmelCase_			:      Dict			=   writer_class(
            features=lowerCamelCase    ,						path=working_fpath.replace("""SSSSS"""    ,						F'{shard_id:05d}'		).replace("""TTTTT"""    ,						F'{task_id:05d}'		)    ,						writer_batch_size=lowerCamelCase    ,						storage_options=lowerCamelCase    ,						embed_local_files=lowerCamelCase    ,						)
        lowerCAmelCase_			:      str			=   pa.Table.from_batches([first_batch]		)
        writer.write_table(lowerCamelCase		)
        for batch in it:
          if max_shard_size is not None and writer._num_bytes >= max_shard_size:
            lowerCAmelCase_,							lowerCAmelCase_			:      Any			=   writer.finalize()
            writer.close()
            yield pa.RecordBatch.from_arrays(
                [[task_id], [num_examples], [num_bytes]]    ,						names=["""task_id""", """num_examples""", """num_bytes"""]    ,						)
            shard_id += 1
            lowerCAmelCase_			:      Optional[int]			=   writer_class(
                features=writer._features    ,						path=working_fpath.replace("""SSSSS"""    ,						F'{shard_id:05d}'		).replace("""TTTTT"""    ,						F'{task_id:05d}'		)    ,						writer_batch_size=lowerCamelCase    ,						storage_options=lowerCamelCase    ,						embed_local_files=lowerCamelCase    ,						)
          lowerCAmelCase_			:      Tuple			=   pa.Table.from_batches([batch]		)
          writer.write_table(lowerCamelCase		)

        if writer._num_bytes > 0:
          lowerCAmelCase_,							lowerCAmelCase_			:      List[str]			=   writer.finalize()
          writer.close()
          yield pa.RecordBatch.from_arrays(
              [[task_id], [num_examples], [num_bytes]]    ,						names=["""task_id""", """num_examples""", """num_bytes"""]    ,						)

        if working_fpath != fpath:
          for file in os.listdir(os.path.dirname(lowerCamelCase		)		):
            lowerCAmelCase_			:      str			=   os.path.join(os.path.dirname(lowerCamelCase		)    ,						os.path.basename(lowerCamelCase		)		)
            shutil.move(lowerCamelCase    ,						lowerCamelCase		)

      lowerCAmelCase_			:      Dict			=   (
          self.df.mapInArrow(lowerCamelCase    ,						"""task_id: long, num_examples: long, num_bytes: long"""		)
          .groupBy("""task_id"""		)
          .agg(
              pyspark.sql.functions.sum("""num_examples"""		).alias("""total_num_examples"""		)    ,						pyspark.sql.functions.sum("""num_bytes"""		).alias("""total_num_bytes"""		)    ,						pyspark.sql.functions.count("""num_bytes"""		).alias("""num_shards"""		)    ,						pyspark.sql.functions.collect_list("""num_examples"""		).alias("""shard_lengths"""		)    ,						)
          .collect()
      )
      for row in stats:
        yield row.task_id, (row.total_num_examples, row.total_num_bytes, row.num_shards, row.shard_lengths)

    def __lowercase   (   self    :  Any    ,						lowerCamelCase    :  "datasets.SplitGenerator"    ,						lowerCamelCase    :  str = "arrow"    ,						lowerCamelCase    :  Optional[Union[str, int]] = None    ,						lowerCamelCase    :  Optional[int] = None    ,						**lowerCamelCase    :  List[str]    ,						)   ->  Optional[int]:
      self._validate_cache_dir()

      lowerCAmelCase_			:      List[str]			=   convert_file_size_to_int(max_shard_size or MAX_SHARD_SIZE		)
      self._repartition_df_if_needed(lowerCamelCase		)
      lowerCAmelCase_			:      Optional[int]			=   not is_remote_filesystem(self._fs		)
      lowerCAmelCase_			:      Dict			=   os.path.join if is_local else posixpath.join

      lowerCAmelCase_			:      int			=   """-TTTTT-SSSSS-of-NNNNN"""
      lowerCAmelCase_			:      Any			=   F'{self.name}-{split_generator.name}{SUFFIX}.{file_format}'
      lowerCAmelCase_			:      List[str]			=   path_join(self._output_dir    ,						lowerCamelCase		)

      lowerCAmelCase_			:      Dict			=   0
      lowerCAmelCase_			:      int			=   0
      lowerCAmelCase_			:      Dict			=   0
      lowerCAmelCase_			:      Any			=   []
      lowerCAmelCase_			:      Dict			=   []

      for task_id, content in self._prepare_split_single(lowerCamelCase    ,						lowerCamelCase    ,						lowerCamelCase		):
        (
            (
            lowerCAmelCase_
        ),							(
            lowerCAmelCase_
        ),							(
            lowerCAmelCase_
        ),							(
            lowerCAmelCase_
        ),							
        )			:      str			=   content
        if num_bytes > 0:
          total_num_examples += num_examples
          total_num_bytes += num_bytes
          total_shards += num_shards
          task_id_and_num_shards.append((task_id, num_shards)		)
          all_shard_lengths.extend(lowerCamelCase		)

      lowerCAmelCase_			:      Union[str, Any]			=   total_num_examples
      lowerCAmelCase_			:      int			=   total_num_bytes

      # should rename everything at the end
      logger.debug(F'Renaming {total_shards} shards.'		)
      if total_shards > 1:
        lowerCAmelCase_			:      Optional[int]			=   all_shard_lengths

        # Define fs outside of _rename_shard so that we don't reference self in the function, which will result in a
        # pickling error due to pickling the SparkContext.
        lowerCAmelCase_			:      List[Any]			=   self._fs

        # use the -SSSSS-of-NNNNN pattern
        def _rename_shard(
            lowerCamelCase    :  int    ,						lowerCamelCase    :  int    ,						lowerCamelCase    :  int    ,						):
          rename(
              lowerCamelCase    ,						fpath.replace("""SSSSS"""    ,						F'{shard_id:05d}'		).replace("""TTTTT"""    ,						F'{task_id:05d}'		)    ,						fpath.replace("""TTTTT-SSSSS"""    ,						F'{global_shard_id:05d}'		).replace("""NNNNN"""    ,						F'{total_shards:05d}'		)    ,						)

        lowerCAmelCase_			:      Any			=   []
        lowerCAmelCase_			:      str			=   0
        for i in range(len(lowerCamelCase		)		):
          lowerCAmelCase_,							lowerCAmelCase_			:      Tuple			=   task_id_and_num_shards[i]
          for shard_id in range(lowerCamelCase		):
            args.append([task_id, shard_id, global_shard_id]		)
            global_shard_id += 1
        self._spark.sparkContext.parallelize(lowerCamelCase    ,						len(lowerCamelCase		)		).map(lambda lowerCamelCase		: _rename_shard(*lowerCamelCase		)		).collect()
      else:
        # don't use any pattern
        lowerCAmelCase_			:      List[Any]			=   0
        lowerCAmelCase_			:      str			=   task_id_and_num_shards[0][0]
        self._rename(
            fpath.replace("""SSSSS"""    ,						F'{shard_id:05d}'		).replace("""TTTTT"""    ,						F'{task_id:05d}'		)    ,						fpath.replace(lowerCamelCase    ,						""""""		)    ,						)





    def __lowercase   (   self    :  Dict    ,						lowerCamelCase    :  "datasets.SplitGenerator"    ,						)   ->  SparkExamplesIterable:
      return SparkExamplesIterable(self.df		)



