


"""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


__UpperCamelCase						:				Tuple             =      datasets.utils.logging.get_logger(__name__)

if TYPE_CHECKING:
	import pyspark



@dataclass
class 			lowerCamelCase__						(   datasets.BuilderConfig			):
						lowerCAmelCase       =     None







def  __snake_case       (   UpperCamelCase__ ,			UpperCamelCase__ ,			)      ->		Optional[int]:







		"""simple docstring"""





		import pyspark

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

		return generate_fn



class 			lowerCamelCase__						(   _BaseExamplesIterable			):







						def __init__(							self   :    Union[str, Any]			,			_lowercase   :    "pyspark.sql.DataFrame"			,			_lowercase   :    Optional[int]=None			,			):
								A					       = df
								A					       = partition_order or range(self.df.rdd.getNumPartitions()		)
								A					       = _generate_iterable_examples(self.df			,			self.partition_order		)







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







						def __a						(							self   :    Optional[Any]			,			_lowercase   :    np.random.Generator		):
								A					       = list(range(self.df.rdd.getNumPartitions()		)		)
								generator.shuffle(_lowercase		)
								return SparkExamplesIterable(self.df			,			partition_order=_lowercase		)







						def __a						(							self   :    List[str]			,			_lowercase   :    int			,			_lowercase   :    int		):
								A					       = self.split_shard_indices_by_worker(_lowercase			,			_lowercase		)
								return SparkExamplesIterable(self.df			,			partition_order=_lowercase		)







						@property
						def __a						(							self   :    Union[str, Any]		):
								return len(self.partition_order		)



class 			lowerCamelCase__						(   datasets.DatasetBuilder			):
						lowerCAmelCase       =     SparkConfig







						def __init__(							self   :    Optional[int]			,			_lowercase   :    "pyspark.sql.DataFrame"			,			_lowercase   :    str = None			,			_lowercase   :    str = None			,			**_lowercase   :    str			,			):
								import pyspark

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

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







						def __a						(							self   :    str		):
								# Returns the path of the created file.
								def create_cache_and_write_probe(_lowercase   :    List[Any]		):
										# 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=_lowercase		)
										A					       = 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(_lowercase			,			'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					       = (
										    self._spark.sparkContext.parallelize(range(1		)			,			1		).mapPartitions(_lowercase		).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   :    List[str]		):
								return datasets.DatasetInfo(features=self.config.features		)







						def __a						(							self   :    str			,			_lowercase   :    datasets.download.download_manager.DownloadManager		):
								return [datasets.SplitGenerator(name=datasets.Split.TRAIN		)]







						def __a						(							self   :    Tuple			,			_lowercase   :    Tuple		):
								import pyspark

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

								A					       = self.df.count()
								A					       = 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					       = (
								    self.df.limit(_lowercase		)
								    .repartition(1		)
								    .mapInArrow(_lowercase			,			'batch_bytes: long'		)
								    .agg(pyspark.sql.functions.sum('batch_bytes'		).alias('sample_bytes'		)		)
								    .collect()[0]
								    .sample_bytes
								    / sample_num_rows
								)
								A					       = 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					       = min(_lowercase			,			int(approx_total_size / max_shard_size		)		)
										A					       = self.df.repartition(_lowercase		)







						def __a						(							self   :    Dict			,			_lowercase   :    str			,			_lowercase   :    str			,			_lowercase   :    int			,			):
								import pyspark

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

								def write_arrow(_lowercase   :    Union[str, Any]		):
										# Within the same SparkContext, no two task attempts will share the same attempt ID.
										A					       = pyspark.TaskContext().taskAttemptId()
										A					       = next(_lowercase			,			_lowercase		)
										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					       = 0
										A					       = writer_class(
										    features=_lowercase			,			path=working_fpath.replace('SSSSS'			,			f'{shard_id:05d}'		).replace('TTTTT'			,			f'{task_id:05d}'		)			,			writer_batch_size=_lowercase			,			storage_options=_lowercase			,			embed_local_files=_lowercase			,			)
										A					       = pa.Table.from_batches([first_batch]		)
										writer.write_table(_lowercase		)
										for batch in it:
												if max_shard_size is not None and writer._num_bytes >= max_shard_size:
														A       , A					       = 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					       = writer_class(
														    features=writer._features			,			path=working_fpath.replace('SSSSS'			,			f'{shard_id:05d}'		).replace('TTTTT'			,			f'{task_id:05d}'		)			,			writer_batch_size=_lowercase			,			storage_options=_lowercase			,			embed_local_files=_lowercase			,			)
												A					       = pa.Table.from_batches([batch]		)
												writer.write_table(_lowercase		)

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

								A					       = (
								    self.df.mapInArrow(_lowercase			,			'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   :    Union[str, Any]			,			_lowercase   :    "datasets.SplitGenerator"			,			_lowercase   :    str = "arrow"			,			_lowercase   :    Optional[Union[str, int]] = None			,			_lowercase   :    Optional[int] = None			,			**_lowercase   :    Optional[Any]			,			):
								self._validate_cache_dir()

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

								A					       = '-TTTTT-SSSSS-of-NNNNN'
								A					       = f'{self.name}-{split_generator.name}{SUFFIX}.{file_format}'
								A					       = path_join(self._output_dir			,			_lowercase		)

								A					       = 0
								A					       = 0
								A					       = 0
								A					       = []
								A					       = []

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

								A					       = total_num_examples
								A					       = total_num_bytes

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

										# use the -SSSSS-of-NNNNN pattern
										def _rename_shard(
										    _lowercase   :    int			,			_lowercase   :    int			,			_lowercase   :    int			,			):
												rename(
												    _lowercase			,			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					       = []
										A					       = 0
										for i in range(len(_lowercase		)		):
												A       , A					       = task_id_and_num_shards[i]
												for shard_id in range(_lowercase		):
														args.append([task_id, shard_id, global_shard_id]		)
														global_shard_id += 1
										self._spark.sparkContext.parallelize(_lowercase			,			len(_lowercase		)		).map(lambda _lowercase		: _rename_shard(*_lowercase		)		).collect()
								else:
										# don't use any pattern
										A					       = 0
										A					       = task_id_and_num_shards[0][0]
										self._rename(
										    fpath.replace('SSSSS'			,			f'{shard_id:05d}'		).replace('TTTTT'			,			f'{task_id:05d}'		)			,			fpath.replace(_lowercase			,			''		)			,			)







						def __a						(							self   :    Optional[int]			,			_lowercase   :    "datasets.SplitGenerator"			,			):
								return SparkExamplesIterable(self.df		)









"""simple docstring"""


import unittest

from transformers import MODEL_FOR_ZERO_SHOT_OBJECT_DETECTION_MAPPING, is_vision_available, pipeline
from transformers.testing_utils import (
    is_pipeline_test,
    nested_simplify,
    require_tf,
    require_torch,
    require_vision,
    slow,
)

from .test_pipelines_common import ANY


if is_vision_available():
	from PIL import Image
else:



	class 			lowerCamelCase__						:







							@staticmethod
							def __a						(							*_lowercase   :    int			,			**_lowercase   :    Optional[int]		):
									pass



@is_pipeline_test
@require_vision
@require_torch
class 			lowerCamelCase__						(   unittest.TestCase			):
						lowerCAmelCase       =     MODEL_FOR_ZERO_SHOT_OBJECT_DETECTION_MAPPING







						def __a						(							self   :    int			,			_lowercase   :    Optional[Any]			,			_lowercase   :    int			,			_lowercase   :    Union[str, Any]		):
								A					       = pipeline(
								    'zero-shot-object-detection'			,			model='hf-internal-testing/tiny-random-owlvit-object-detection'		)

								A					       = [
								    {
								        'image': './tests/fixtures/tests_samples/COCO/000000039769.png',
								        'candidate_labels': ['cat', 'remote', 'couch'],
								    }
								]
								return object_detector, examples







						def __a						(							self   :    Tuple			,			_lowercase   :    int			,			_lowercase   :    Tuple		):
								A					       = object_detector(examples[0]			,			threshold=0.0		)

								A					       = len(_lowercase		)
								self.assertGreater(_lowercase			,			0		)
								self.assertEqual(
								    _lowercase			,			[
								        {
								            'score': ANY(_lowercase		),
								            'label': ANY(_lowercase		),
								            'box': {'xmin': ANY(_lowercase		), 'ymin': ANY(_lowercase		), 'xmax': ANY(_lowercase		), 'ymax': ANY(_lowercase		)},
								        }
								        for i in range(_lowercase		)
								    ]			,			)







						@require_tf
						@unittest.skip('Zero Shot Object Detection not implemented in TF'		)
						def __a						(							self   :    Optional[int]		):
								pass







						@require_torch
						def __a						(							self   :    Optional[Any]		):
								A					       = pipeline(
								    'zero-shot-object-detection'			,			model='hf-internal-testing/tiny-random-owlvit-object-detection'		)

								A					       = object_detector(
								    './tests/fixtures/tests_samples/COCO/000000039769.png'			,			candidate_labels=['cat', 'remote', 'couch']			,			threshold=0.6_4			,			)

								self.assertEqual(
								    nested_simplify(_lowercase			,			decimals=4		)			,			[
								        {'score': 0.7_2_3_5, 'label': 'cat', 'box': {'xmin': 204, 'ymin': 167, 'xmax': 232, 'ymax': 190}},
								        {'score': 0.7_2_1_8, 'label': 'remote', 'box': {'xmin': 204, 'ymin': 167, 'xmax': 232, 'ymax': 190}},
								        {'score': 0.7_1_8_4, 'label': 'couch', 'box': {'xmin': 204, 'ymin': 167, 'xmax': 232, 'ymax': 190}},
								        {'score': 0.6_7_4_8, 'label': 'remote', 'box': {'xmin': 571, 'ymin': 83, 'xmax': 598, 'ymax': 103}},
								        {'score': 0.6_6_5_6, 'label': 'cat', 'box': {'xmin': 571, 'ymin': 83, 'xmax': 598, 'ymax': 103}},
								        {'score': 0.6_6_1_4, 'label': 'couch', 'box': {'xmin': 571, 'ymin': 83, 'xmax': 598, 'ymax': 103}},
								        {'score': 0.6_4_5_6, 'label': 'remote', 'box': {'xmin': 494, 'ymin': 105, 'xmax': 521, 'ymax': 127}},
								        {'score': 0.6_4_2, 'label': 'remote', 'box': {'xmin': 67, 'ymin': 274, 'xmax': 93, 'ymax': 297}},
								        {'score': 0.6_4_1_9, 'label': 'cat', 'box': {'xmin': 494, 'ymin': 105, 'xmax': 521, 'ymax': 127}},
								    ]			,			)

								A					       = object_detector(
								    [
								        {
								            'image': './tests/fixtures/tests_samples/COCO/000000039769.png',
								            'candidate_labels': ['cat', 'remote', 'couch'],
								        }
								    ]			,			threshold=0.6_4			,			)

								self.assertEqual(
								    nested_simplify(_lowercase			,			decimals=4		)			,			[
								        [
								            {'score': 0.7_2_3_5, 'label': 'cat', 'box': {'xmin': 204, 'ymin': 167, 'xmax': 232, 'ymax': 190}},
								            {'score': 0.7_2_1_8, 'label': 'remote', 'box': {'xmin': 204, 'ymin': 167, 'xmax': 232, 'ymax': 190}},
								            {'score': 0.7_1_8_4, 'label': 'couch', 'box': {'xmin': 204, 'ymin': 167, 'xmax': 232, 'ymax': 190}},
								            {'score': 0.6_7_4_8, 'label': 'remote', 'box': {'xmin': 571, 'ymin': 83, 'xmax': 598, 'ymax': 103}},
								            {'score': 0.6_6_5_6, 'label': 'cat', 'box': {'xmin': 571, 'ymin': 83, 'xmax': 598, 'ymax': 103}},
								            {'score': 0.6_6_1_4, 'label': 'couch', 'box': {'xmin': 571, 'ymin': 83, 'xmax': 598, 'ymax': 103}},
								            {'score': 0.6_4_5_6, 'label': 'remote', 'box': {'xmin': 494, 'ymin': 105, 'xmax': 521, 'ymax': 127}},
								            {'score': 0.6_4_2, 'label': 'remote', 'box': {'xmin': 67, 'ymin': 274, 'xmax': 93, 'ymax': 297}},
								            {'score': 0.6_4_1_9, 'label': 'cat', 'box': {'xmin': 494, 'ymin': 105, 'xmax': 521, 'ymax': 127}},
								        ]
								    ]			,			)







						@require_torch
						@slow
						def __a						(							self   :    List[str]		):
								A					       = pipeline('zero-shot-object-detection'		)

								A					       = object_detector(
								    'http://images.cocodataset.org/val2017/000000039769.jpg'			,			candidate_labels=['cat', 'remote', 'couch']			,			)
								self.assertEqual(
								    nested_simplify(_lowercase			,			decimals=4		)			,			[
								        {'score': 0.2_8_6_8, 'label': 'cat', 'box': {'xmin': 324, 'ymin': 20, 'xmax': 640, 'ymax': 373}},
								        {'score': 0.2_7_7, 'label': 'remote', 'box': {'xmin': 40, 'ymin': 72, 'xmax': 177, 'ymax': 115}},
								        {'score': 0.2_5_3_7, 'label': 'cat', 'box': {'xmin': 1, 'ymin': 55, 'xmax': 315, 'ymax': 472}},
								        {'score': 0.1_4_7_4, 'label': 'remote', 'box': {'xmin': 335, 'ymin': 74, 'xmax': 371, 'ymax': 187}},
								        {'score': 0.1_2_0_8, 'label': 'couch', 'box': {'xmin': 4, 'ymin': 0, 'xmax': 642, 'ymax': 476}},
								    ]			,			)

								A					       = object_detector(
								    [
								        {
								            'image': 'http://images.cocodataset.org/val2017/000000039769.jpg',
								            'candidate_labels': ['cat', 'remote', 'couch'],
								        },
								        {
								            'image': 'http://images.cocodataset.org/val2017/000000039769.jpg',
								            'candidate_labels': ['cat', 'remote', 'couch'],
								        },
								    ]			,			)
								self.assertEqual(
								    nested_simplify(_lowercase			,			decimals=4		)			,			[
								        [
								            {'score': 0.2_8_6_8, 'label': 'cat', 'box': {'xmin': 324, 'ymin': 20, 'xmax': 640, 'ymax': 373}},
								            {'score': 0.2_7_7, 'label': 'remote', 'box': {'xmin': 40, 'ymin': 72, 'xmax': 177, 'ymax': 115}},
								            {'score': 0.2_5_3_7, 'label': 'cat', 'box': {'xmin': 1, 'ymin': 55, 'xmax': 315, 'ymax': 472}},
								            {'score': 0.1_4_7_4, 'label': 'remote', 'box': {'xmin': 335, 'ymin': 74, 'xmax': 371, 'ymax': 187}},
								            {'score': 0.1_2_0_8, 'label': 'couch', 'box': {'xmin': 4, 'ymin': 0, 'xmax': 642, 'ymax': 476}},
								        ],
								        [
								            {'score': 0.2_8_6_8, 'label': 'cat', 'box': {'xmin': 324, 'ymin': 20, 'xmax': 640, 'ymax': 373}},
								            {'score': 0.2_7_7, 'label': 'remote', 'box': {'xmin': 40, 'ymin': 72, 'xmax': 177, 'ymax': 115}},
								            {'score': 0.2_5_3_7, 'label': 'cat', 'box': {'xmin': 1, 'ymin': 55, 'xmax': 315, 'ymax': 472}},
								            {'score': 0.1_4_7_4, 'label': 'remote', 'box': {'xmin': 335, 'ymin': 74, 'xmax': 371, 'ymax': 187}},
								            {'score': 0.1_2_0_8, 'label': 'couch', 'box': {'xmin': 4, 'ymin': 0, 'xmax': 642, 'ymax': 476}},
								        ],
								    ]			,			)







						@require_tf
						@unittest.skip('Zero Shot Object Detection not implemented in TF'		)
						def __a						(							self   :    Optional[int]		):
								pass







						@require_torch
						@slow
						def __a						(							self   :    Optional[Any]		):
								A					       = 0.2
								A					       = pipeline('zero-shot-object-detection'		)

								A					       = object_detector(
								    'http://images.cocodataset.org/val2017/000000039769.jpg'			,			candidate_labels=['cat', 'remote', 'couch']			,			threshold=_lowercase			,			)
								self.assertEqual(
								    nested_simplify(_lowercase			,			decimals=4		)			,			[
								        {'score': 0.2_8_6_8, 'label': 'cat', 'box': {'xmin': 324, 'ymin': 20, 'xmax': 640, 'ymax': 373}},
								        {'score': 0.2_7_7, 'label': 'remote', 'box': {'xmin': 40, 'ymin': 72, 'xmax': 177, 'ymax': 115}},
								        {'score': 0.2_5_3_7, 'label': 'cat', 'box': {'xmin': 1, 'ymin': 55, 'xmax': 315, 'ymax': 472}},
								    ]			,			)







						@require_torch
						@slow
						def __a						(							self   :    Optional[int]		):
								A					       = 2
								A					       = pipeline('zero-shot-object-detection'		)

								A					       = object_detector(
								    'http://images.cocodataset.org/val2017/000000039769.jpg'			,			candidate_labels=['cat', 'remote', 'couch']			,			top_k=_lowercase			,			)
								self.assertEqual(
								    nested_simplify(_lowercase			,			decimals=4		)			,			[
								        {'score': 0.2_8_6_8, 'label': 'cat', 'box': {'xmin': 324, 'ymin': 20, 'xmax': 640, 'ymax': 373}},
								        {'score': 0.2_7_7, 'label': 'remote', 'box': {'xmin': 40, 'ymin': 72, 'xmax': 177, 'ymax': 115}},
								    ]			,			)





