





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


UpperCamelCase__     =   datasets.utils.logging.get_logger(__name__)

if TYPE_CHECKING:
					import pyspark

@dataclass
class     __SCREAMING_SNAKE_CASE   (       datasets.BuilderConfig    ):
							snake_case       : Optional[datasets.Features]         =    None






def       _UpperCamelCase       (a__				:"pyspark.sql.DataFrame"				,     a__				:List[int]				,     ):






							"""simple docstring"""





							import pyspark

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

							return generate_fn

class     __SCREAMING_SNAKE_CASE   (       _BaseExamplesIterable    ):






							def __init__(  self					,	__lowerCAmelCase					,	__lowerCAmelCase=None					,	):
														UpperCamelCase__			       =     df
														UpperCamelCase__			       =     partition_order or range(self.df.rdd.getNumPartitions()					)
														UpperCamelCase__			       =     _generate_iterable_examples(self.df					,	self.partition_order					)






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






							def 			_lowerCamelCase						(  self					,	__lowerCAmelCase					):
														UpperCamelCase__			       =     list(range(self.df.rdd.getNumPartitions()					)					)
														generator.shuffle(__lowerCAmelCase					)
														return SparkExamplesIterable(self.df					,	partition_order=__lowerCAmelCase					)






							def 			_lowerCamelCase						(  self					,	__lowerCAmelCase					,	__lowerCAmelCase					):
														UpperCamelCase__			       =     self.split_shard_indices_by_worker(__lowerCAmelCase					,	__lowerCAmelCase					)
														return SparkExamplesIterable(self.df					,	partition_order=__lowerCAmelCase					)







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






class     __SCREAMING_SNAKE_CASE   (       datasets.DatasetBuilder    ):
							snake_case       : Any         =    SparkConfig






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

														UpperCamelCase__			       =     pyspark.sql.SparkSession.builder.getOrCreate()
														UpperCamelCase__			       =     df
														UpperCamelCase__			       =     working_dir

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






							def 			_lowerCamelCase						(  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					)
																					UpperCamelCase__			       =     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:
																					UpperCamelCase__			       =     (
																					    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 			_lowerCamelCase						(  self					):
														return datasets.DatasetInfo(features=self.config.features					)






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






							def 			_lowerCamelCase						(  self					,	__lowerCAmelCase					):
														import pyspark

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

														UpperCamelCase__			       =     self.df.count()
														UpperCamelCase__			       =     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.
														UpperCamelCase__			       =     (
														    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
														)
														UpperCamelCase__			       =     approx_bytes_per_row * df_num_rows
														if approx_total_size > max_shard_size:
																					# Make sure there is at least one row per partition.
																					UpperCamelCase__			       =     min(__lowerCAmelCase					,	int(approx_total_size / max_shard_size					)					)
																					UpperCamelCase__			       =     self.df.repartition(__lowerCAmelCase					)






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

														UpperCamelCase__			       =     ParquetWriter if file_format == """parquet""" else ArrowWriter
														UpperCamelCase__			       =     os.path.join(self._working_dir					,	os.path.basename(__lowerCAmelCase					)					) if self._working_dir else fpath
														UpperCamelCase__			       =     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.
														UpperCamelCase__			       =     self.config.features
														UpperCamelCase__			       =     self._writer_batch_size
														UpperCamelCase__			       =     self._fs.storage_options

														def write_arrow(__lowerCAmelCase					):
																					# Within the same SparkContext, no two task attempts will share the same attempt ID.
																					UpperCamelCase__			       =     pyspark.TaskContext().taskAttemptId()
																					UpperCamelCase__			       =     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"""]					,	)
																					UpperCamelCase__			       =     0
																					UpperCamelCase__			       =     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					,	)
																					UpperCamelCase__			       =     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:
																																			UpperCamelCase__       ,       UpperCamelCase__			       =     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
																																			UpperCamelCase__			       =     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					,	)
																												UpperCamelCase__			       =     pa.Table.from_batches([batch]					)
																												writer.write_table(__lowerCAmelCase					)

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

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

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

														UpperCamelCase__			       =     """-TTTTT-SSSSS-of-NNNNN"""
														UpperCamelCase__			       =     f"""{self.name}-{split_generator.name}{SUFFIX}.{file_format}"""
														UpperCamelCase__			       =     path_join(self._output_dir					,	__lowerCAmelCase					)

														UpperCamelCase__			       =     0
														UpperCamelCase__			       =     0
														UpperCamelCase__			       =     0
														UpperCamelCase__			       =     []
														UpperCamelCase__			       =     []

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

														UpperCamelCase__			       =     total_num_examples
														UpperCamelCase__			       =     total_num_bytes

														# should rename everything at the end
														logger.debug(f"""Renaming {total_shards} shards."""					)
														if total_shards > 1:
																					UpperCamelCase__			       =     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.
																					UpperCamelCase__			       =     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}"""					)					,	)

																					UpperCamelCase__			       =     []
																					UpperCamelCase__			       =     0
																					for i in range(len(__lowerCAmelCase					)					):
																												UpperCamelCase__       ,       UpperCamelCase__			       =     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
																					UpperCamelCase__			       =     0
																					UpperCamelCase__			       =     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 			_lowerCamelCase						(  self					,	__lowerCAmelCase					,	):
														return SparkExamplesIterable(self.df					)











from ...configuration_utils import PretrainedConfig
from ...utils import logging


UpperCamelCase__     =	logging.get_logger(__name__)

UpperCamelCase__     =	{
    "microsoft/cvt-13": "https://huggingface.co/microsoft/cvt-13/resolve/main/config.json",
    # See all Cvt models at https://huggingface.co/models?filter=cvt
}

class     __SCREAMING_SNAKE_CASE   (       _a    ):
							snake_case       : int         =    """cvt"""






							def __init__(  self					,	__lowerCAmelCase=3					,	__lowerCAmelCase=[7, 3, 3]					,	__lowerCAmelCase=[4, 2, 2]					,	__lowerCAmelCase=[2, 1, 1]					,	__lowerCAmelCase=[64, 192, 384]					,	__lowerCAmelCase=[1, 3, 6]					,	__lowerCAmelCase=[1, 2, 10]					,	__lowerCAmelCase=[4.0, 4.0, 4.0]					,	__lowerCAmelCase=[0.0, 0.0, 0.0]					,	__lowerCAmelCase=[0.0, 0.0, 0.0]					,	__lowerCAmelCase=[0.0, 0.0, 0.1]					,	__lowerCAmelCase=[True, True, True]					,	__lowerCAmelCase=[False, False, True]					,	__lowerCAmelCase=["dw_bn", "dw_bn", "dw_bn"]					,	__lowerCAmelCase=[3, 3, 3]					,	__lowerCAmelCase=[1, 1, 1]					,	__lowerCAmelCase=[2, 2, 2]					,	__lowerCAmelCase=[1, 1, 1]					,	__lowerCAmelCase=[1, 1, 1]					,	__lowerCAmelCase=0.02					,	__lowerCAmelCase=1E-12					,	**__lowerCAmelCase					,	):
														super().__init__(**__lowerCAmelCase					)
														UpperCamelCase__			       =     num_channels
														UpperCamelCase__			       =     patch_sizes
														UpperCamelCase__			       =     patch_stride
														UpperCamelCase__			       =     patch_padding
														UpperCamelCase__			       =     embed_dim
														UpperCamelCase__			       =     num_heads
														UpperCamelCase__			       =     depth
														UpperCamelCase__			       =     mlp_ratio
														UpperCamelCase__			       =     attention_drop_rate
														UpperCamelCase__			       =     drop_rate
														UpperCamelCase__			       =     drop_path_rate
														UpperCamelCase__			       =     qkv_bias
														UpperCamelCase__			       =     cls_token
														UpperCamelCase__			       =     qkv_projection_method
														UpperCamelCase__			       =     kernel_qkv
														UpperCamelCase__			       =     padding_kv
														UpperCamelCase__			       =     stride_kv
														UpperCamelCase__			       =     padding_q
														UpperCamelCase__			       =     stride_q
														UpperCamelCase__			       =     initializer_range
														UpperCamelCase__			       =     layer_norm_eps




