





"""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        __UpperCamelCase				(						datasets.BuilderConfig						):
							lowerCamelCase   :					Optional[datasets.Features]      					=None






def       _SCREAMING_SNAKE_CASE		(	_lowercase						:    "pyspark.sql.DataFrame"    ,       _lowercase						:    List[int]    ,       ) ->Union[str, Any]:






	'''simple docstring'''






	import pyspark

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

	return generate_fn


class        __UpperCamelCase				(						_BaseExamplesIterable						):






							def __init__(   self      ,					lowerCAmelCase__      ,					lowerCAmelCase__=None      ,					)	-> List[str]:
								a      :		int	=				df
								a      :		Tuple	=				partition_order or range(self.df.rdd.getNumPartitions()      )
								a      :		Dict	=				_generate_iterable_examples(self.df      ,					self.partition_order      )






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






							def     __a   (   self      ,					lowerCAmelCase__      )	-> "SparkExamplesIterable":
								a      :		Dict	=				list(range(self.df.rdd.getNumPartitions()      )      )
								generator.shuffle(lowerCAmelCase__      )
								return SparkExamplesIterable(self.df      ,					partition_order=lowerCAmelCase__      )






							def     __a   (   self      ,					lowerCAmelCase__      ,					lowerCAmelCase__      )	-> "SparkExamplesIterable":
								a      :		Optional[Any]	=				self.split_shard_indices_by_worker(lowerCAmelCase__      ,					lowerCAmelCase__      )
								return SparkExamplesIterable(self.df      ,					partition_order=lowerCAmelCase__      )






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


class        __UpperCamelCase				(						datasets.DatasetBuilder						):
							lowerCamelCase   :					Dict      					=SparkConfig






							def __init__(   self      ,					lowerCAmelCase__      ,					lowerCAmelCase__ = None      ,					lowerCAmelCase__ = None      ,					**lowerCAmelCase__      ,					)	-> Tuple:
								import pyspark

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

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






							def     __a   (   self      )	-> Tuple:
								# 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      :		Dict	=				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      :		Union[str, Any]	=				(
									    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     __a   (   self      )	-> str:
								return datasets.DatasetInfo(features=self.config.features      )






							def     __a   (   self      ,					lowerCAmelCase__      )	-> Dict:
								return [datasets.SplitGenerator(name=datasets.Split.TRAIN      )]






							def     __a   (   self      ,					lowerCAmelCase__      )	-> List[str]:
								import pyspark

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

								a      :		Dict	=				self.df.count()
								a      :		str	=				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      :		Optional[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      :		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.
									a      :		Any	=				min(lowerCAmelCase__      ,					int(approx_total_size / max_shard_size      )      )
									a      :		Union[str, Any]	=				self.df.repartition(lowerCAmelCase__      )






							def     __a   (   self      ,					lowerCAmelCase__      ,					lowerCAmelCase__      ,					lowerCAmelCase__      ,					)	-> Iterable[Tuple[int, bool, Union[int, tuple]]]:
								import pyspark

								a      :		List[Any]	=				ParquetWriter if file_format == "parquet" else ArrowWriter
								a      :		str	=				os.path.join(self._working_dir      ,					os.path.basename(lowerCAmelCase__      )      ) if self._working_dir else fpath
								a      :		List[Any]	=				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      :		Optional[Any]	=				self.config.features
								a      :		Tuple	=				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[int]	=				pyspark.TaskContext().taskAttemptId()
									a      :		Dict	=				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      :		List[str]	=				0
									a      :		Optional[int]	=				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      :		int	=				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      :		List[str]	=				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[str]	=				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      :		List[str]	=				pa.Table.from_batches([batch]      )
										writer.write_table(lowerCAmelCase__      )

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

								a      :		Any	=				(
								    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     __a   (   self      ,					lowerCAmelCase__      ,					lowerCAmelCase__ = "arrow"      ,					lowerCAmelCase__ = None      ,					lowerCAmelCase__ = None      ,					**lowerCAmelCase__      ,					)	-> Optional[int]:
								self._validate_cache_dir()

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

								a      :		Optional[Any]	=				"-TTTTT-SSSSS-of-NNNNN"
								a      :		Optional[Any]	=				f"""{self.name}-{split_generator.name}{SUFFIX}.{file_format}"""
								a      :		int	=				path_join(self._output_dir      ,					lowerCAmelCase__      )

								a      :		List[Any]	=				0
								a      :		Optional[Any]	=				0
								a      :		List[Any]	=				0
								a      :		int	=				[]
								a      :		List[Any]	=				[]

								for task_id, content in self._prepare_split_single(lowerCAmelCase__      ,					lowerCAmelCase__      ,					lowerCAmelCase__      ):
									(
									    (
									    a
									),       (
									    a
									),       (
									    a
									),       (
									    a
									),       
									)      :		Optional[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      :		Tuple	=				total_num_examples
								a      :		List[Any]	=				total_num_bytes

								# should rename everything at the end
								logger.debug(f"""Renaming {total_shards} shards."""      )
								if total_shards > 1:
									a      :		List[str]	=				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      :		List[Any]	=				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      :		Dict	=				[]
									a      :		List[Any]	=				0
									for i in range(len(lowerCAmelCase__      )      ):
										a,       a      :		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      :		Optional[Any]	=				0
									a      :		Optional[Any]	=				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     __a   (   self      ,					lowerCAmelCase__      ,					)	-> SparkExamplesIterable:
								return SparkExamplesIterable(self.df      )









"""simple docstring"""




from unittest.mock import Mock, patch

from file_transfer.send_file import send_file






@patch("socket.socket"					)
@patch("builtins.open"					)
def       _SCREAMING_SNAKE_CASE		(	_lowercase						:    List[Any]    ,       _lowercase						:    int					) ->str:






	'''simple docstring'''






	a      :		Optional[Any]	=				Mock()
	a      :		Dict	=				conn, Mock()
	a      :		Union[str, Any]	=				iter([1, None]					)
	a      :		Optional[int]	=				lambda _lowercase					: next(_lowercase					)

	# ===== invoke =====
	send_file(filename="mytext.txt"    ,       testing=_lowercase					)

	# ===== ensurance =====
	sock.assert_called_once()
	sock.return_value.bind.assert_called_once()
	sock.return_value.listen.assert_called_once()
	sock.return_value.accept.assert_called_once()
	conn.recv.assert_called_once()

	file.return_value.__enter__.assert_called_once()
	file.return_value.__enter__.return_value.read.assert_called()

	conn.send.assert_called_once()
	conn.close.assert_called_once()
	sock.return_value.shutdown.assert_called_once()
	sock.return_value.close.assert_called_once()


