


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


snake_case    :   Union[str, Any]	    =					datasets.utils.logging.get_logger(__name__)

if TYPE_CHECKING:
     import pyspark







@dataclass
class 						_snake_case  (				datasets.BuilderConfig							):
 SCREAMING_SNAKE_CASE__				=	None







def     __lowerCamelCase						(			UpperCAmelCase_						:		"pyspark.sql.DataFrame"			,  UpperCAmelCase_						:		List[int]			,  ):




      """simple docstring"""





      import pyspark

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

      return generate_fn







class 						_snake_case  (				_BaseExamplesIterable							):




 def __init__(		self ,  _lowerCamelCase ,  _lowerCamelCase=None ,  ):
       a						:List[str]										=	df
       a						:int										=	partition_order or range(self.df.rdd.getNumPartitions()   )
       a						:Optional[int]										=	_generate_iterable_examples(self.df ,  self.partition_order   )




 def __iter__(		self   ):
       yield from self.generate_examples_fn()




 def 	SCREAMING_SNAKE_CASE__    (		self ,  _lowerCamelCase   ):
       a						:Optional[Any]										=	list(range(self.df.rdd.getNumPartitions()   )   )
       generator.shuffle(__lowerCAmelCase   )
       return SparkExamplesIterable(self.df ,  partition_order=__lowerCAmelCase   )




 def 	SCREAMING_SNAKE_CASE__    (		self ,  _lowerCamelCase ,  _lowerCamelCase   ):
       a						:str										=	self.split_shard_indices_by_worker(__lowerCAmelCase ,  __lowerCAmelCase   )
       return SparkExamplesIterable(self.df ,  partition_order=__lowerCAmelCase   )




 @property
 def 	SCREAMING_SNAKE_CASE__    (		self   ):
       return len(self.partition_order   )







class 						_snake_case  (				datasets.DatasetBuilder							):
 SCREAMING_SNAKE_CASE__				=	SparkConfig




 def __init__(		self ,  _lowerCamelCase ,  _lowerCamelCase = None ,  _lowerCamelCase = None ,  **_lowerCamelCase ,  ):
       import pyspark

       a						:int										=	pyspark.sql.SparkSession.builder.getOrCreate()
       a						:int										=	df
       a						:Dict										=	working_dir

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




 def 	SCREAMING_SNAKE_CASE__    (		self   ):
       # Returns the path of the created file.
       def create_cache_and_write_probe(_lowerCamelCase   ):
             # 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   )
             a						:Optional[int]										=	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:
             a						: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 	SCREAMING_SNAKE_CASE__    (		self   ):
       return datasets.DatasetInfo(features=self.config.features   )




 def 	SCREAMING_SNAKE_CASE__    (		self ,  _lowerCamelCase   ):
       return [datasets.SplitGenerator(name=datasets.Split.TRAIN   )]




 def 	SCREAMING_SNAKE_CASE__    (		self ,  _lowerCamelCase   ):
       import pyspark

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

       a						:Optional[Any]										=	self.df.count()
       a						:Tuple										=	df_num_rows if df_num_rows <= 100 else 100
       # Approximate the size of each row (in Arrow format) by averaging over a max-100-row sample.
       a						:Union[str, Any]										=	(
           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
       )
       a						:Dict										=	approx_bytes_per_row * df_num_rows
       if approx_total_size > max_shard_size:
             # Make sure there is at least one row per partition.
             a						:Optional[int]										=	min(__lowerCAmelCase ,  int(approx_total_size / max_shard_size   )   )
             a						:List[str]										=	self.df.repartition(__lowerCAmelCase   )




 def 	SCREAMING_SNAKE_CASE__    (		self ,  _lowerCamelCase ,  _lowerCamelCase ,  _lowerCamelCase ,  ):
       import pyspark

       a						:Any										=	ParquetWriter if file_format == '''parquet''' else ArrowWriter
       a						:List[Any]										=	os.path.join(self._working_dir ,  os.path.basename(__lowerCAmelCase   )   ) if self._working_dir else fpath
       a						: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.
       a						:int										=	self.config.features
       a						:Any										=	self._writer_batch_size
       a						:Union[str, Any]										=	self._fs.storage_options

       def write_arrow(_lowerCamelCase   ):
             # Within the same SparkContext, no two task attempts will share the same attempt ID.
             a						:Optional[Any]										=	pyspark.TaskContext().taskAttemptId()
             a						:Optional[int]										=	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'''] ,  )
             a						:Tuple										=	0
             a						:List[Any]										=	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 ,  )
             a						:Dict										=	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:
                         a							,    a						:Dict										=	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
                         a						:List[Any]										=	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 ,  )
                   a						:Any										=	pa.Table.from_batches([batch]   )
                   writer.write_table(__lowerCAmelCase   )

             if writer._num_bytes > 0:
                   a							,    a						: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   )   ):
                         a						:Optional[int]										=	os.path.join(os.path.dirname(__lowerCAmelCase   ) ,  os.path.basename(__lowerCAmelCase   )   )
                         shutil.move(__lowerCAmelCase ,  __lowerCAmelCase   )

       a						: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 	SCREAMING_SNAKE_CASE__    (		self ,  _lowerCamelCase ,  _lowerCamelCase = "arrow" ,  _lowerCamelCase = None ,  _lowerCamelCase = None ,  **_lowerCamelCase ,  ):
       self._validate_cache_dir()

       a						:Dict										=	convert_file_size_to_int(max_shard_size or MAX_SHARD_SIZE   )
       self._repartition_df_if_needed(__lowerCAmelCase   )
       a						:Tuple										=	not is_remote_filesystem(self._fs   )
       a						:Tuple										=	os.path.join if is_local else posixpath.join

       a						:Dict										=	'''-TTTTT-SSSSS-of-NNNNN'''
       a						:Union[str, Any]										=	F'''{self.name}-{split_generator.name}{SUFFIX}.{file_format}'''
       a						:List[str]										=	path_join(self._output_dir ,  __lowerCAmelCase   )

       a						:Tuple										=	0
       a						:List[Any]										=	0
       a						:List[Any]										=	0
       a						:Dict										=	[]
       a						:str										=	[]

       for task_id, content in self._prepare_split_single(__lowerCAmelCase ,  __lowerCAmelCase ,  __lowerCAmelCase   ):
             (
                 (
                 a       
             )							,    (
                 a       
             )							,    (
                 a       
             )							,    (
                 a       
             )							,           
             )						:int										=	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   )

       a						:Any										=	total_num_examples
       a						:int										=	total_num_bytes

       # should rename everything at the end
       logger.debug(F'''Renaming {total_shards} shards.'''   )
       if total_shards > 1:
             a						:Dict										=	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.
             a						:int										=	self._fs

             # use the -SSSSS-of-NNNNN pattern
             def _rename_shard(
                 _lowerCamelCase ,  _lowerCamelCase ,  _lowerCamelCase ,  ):
                   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}'''   ) ,  )

             a						:str										=	[]
             a						:Tuple										=	0
             for i in range(len(__lowerCAmelCase   )   ):
                   a							,    a						:Union[str, Any]										=	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
             a						:List[Any]										=	0
             a						:Tuple										=	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 	SCREAMING_SNAKE_CASE__    (		self ,  _lowerCamelCase ,  ):
       return SparkExamplesIterable(self.df   )









from collections import OrderedDict
from typing import List, Mapping

from packaging import version

from ...configuration_utils import PretrainedConfig
from ...onnx import OnnxConfig
from ...utils import logging


snake_case    :   Union[str, Any]	    =					logging.get_logger(__name__)

snake_case    :   List[str]	    =					{
    '''google/efficientnet-b7''': '''https://huggingface.co/google/efficientnet-b7/resolve/main/config.json''',
}







class 		_snake_case  (				_snake_case							):
 SCREAMING_SNAKE_CASE__				=	'efficientnet'




 def __init__(		self ,  _lowerCamelCase = 3 ,  _lowerCamelCase = 600 ,  _lowerCamelCase = 2.0 ,  _lowerCamelCase = 3.1 ,  _lowerCamelCase = 8 ,  _lowerCamelCase = [3, 3, 5, 3, 5, 5, 3] ,  _lowerCamelCase = [32, 16, 24, 40, 80, 112, 192] ,  _lowerCamelCase = [16, 24, 40, 80, 112, 192, 320] ,  _lowerCamelCase = [] ,  _lowerCamelCase = [1, 2, 2, 2, 1, 2, 1] ,  _lowerCamelCase = [1, 2, 2, 3, 3, 4, 1] ,  _lowerCamelCase = [1, 6, 6, 6, 6, 6, 6] ,  _lowerCamelCase = 0.25 ,  _lowerCamelCase = "swish" ,  _lowerCamelCase = 2560 ,  _lowerCamelCase = "mean" ,  _lowerCamelCase = 0.02 ,  _lowerCamelCase = 0.001 ,  _lowerCamelCase = 0.99 ,  _lowerCamelCase = 0.5 ,  _lowerCamelCase = 0.2 ,  **_lowerCamelCase ,  ):
       super().__init__(**_lowerCamelCase   )

       a						:Optional[int]										=	num_channels
       a						:List[str]										=	image_size
       a						:int										=	width_coefficient
       a						:Optional[Any]										=	depth_coefficient
       a						:Any										=	depth_divisor
       a						:Any										=	kernel_sizes
       a						:Tuple										=	in_channels
       a						:Union[str, Any]										=	out_channels
       a						:Any										=	depthwise_padding
       a						:Any										=	strides
       a						:Optional[Any]										=	num_block_repeats
       a						:Tuple										=	expand_ratios
       a						:Dict										=	squeeze_expansion_ratio
       a						:int										=	hidden_act
       a						:Dict										=	hidden_dim
       a						:Tuple										=	pooling_type
       a						:Any										=	initializer_range
       a						:Tuple										=	batch_norm_eps
       a						:Optional[int]										=	batch_norm_momentum
       a						:List[Any]										=	dropout_rate
       a						:Optional[int]										=	drop_connect_rate
       a						:Tuple										=	sum(_lowerCamelCase   ) * 4







class 		_snake_case  (				_snake_case							):
 SCREAMING_SNAKE_CASE__				=	version.parse('1.11'							)




 @property
 def 	SCREAMING_SNAKE_CASE__    (		self   ):
       return OrderedDict(
           [
               ('''pixel_values''', {0: '''batch''', 1: '''num_channels''', 2: '''height''', 3: '''width'''}),
           ]   )




 @property
 def 	SCREAMING_SNAKE_CASE__    (		self   ):
       return 1e-5





