


"""simple docstring"""




import importlib.metadata
from typing import Union

from packaging.version import Version, parse

from .constants import STR_OPERATION_TO_FUNC


__UpperCamelCase						=      parse(importlib.metadata.version('''torch'''))


def     lowercase			(SCREAMING_SNAKE_CASE_		:				Union[str, Version] ,			SCREAMING_SNAKE_CASE_		:				str ,			SCREAMING_SNAKE_CASE_		:				str  )   ->						Dict:
						if operation not in STR_OPERATION_TO_FUNC.keys():
												raise ValueError(F'`operation` must be one of {list(STR_OPERATION_TO_FUNC.keys()  )}, received {operation}'  )
						SCREAMING_SNAKE_CASE													=					STR_OPERATION_TO_FUNC[operation]
						if isinstance(SCREAMING_SNAKE_CASE_ ,			SCREAMING_SNAKE_CASE_  ):
												SCREAMING_SNAKE_CASE													=					parse(importlib.metadata.version(SCREAMING_SNAKE_CASE_  )  )
						return operation(SCREAMING_SNAKE_CASE_ ,			parse(SCREAMING_SNAKE_CASE_  )  )


def     lowercase			(SCREAMING_SNAKE_CASE_		:				str ,			SCREAMING_SNAKE_CASE_		:				str  )   ->						str:
						return compare_versions(SCREAMING_SNAKE_CASE_ ,			SCREAMING_SNAKE_CASE_ ,			SCREAMING_SNAKE_CASE_  )






"""simple docstring"""




import os
import tempfile
from functools import partial
from unittest import TestCase
from unittest.mock import patch

import datasets
import datasets.config

from .utils import require_beam
class 					lowerCAmelCase	(					datasets.BeamBasedBuilder    ):




		'''simple docstring'''







		def 							__A						(	self				)     ->							Dict:
								return datasets.DatasetInfo(
								    features=datasets.Features({'content': datasets.Value('string'				)}				)						,       supervised_keys=lowerCAmelCase__						,       )







		def 							__A						(	self						,       lowerCAmelCase__						,       lowerCAmelCase__				)     ->							str:
								return [datasets.SplitGenerator(name=datasets.Split.TRAIN						,       gen_kwargs={'examples': get_test_dummy_examples()}				)]







		def 							__A						(	self						,       lowerCAmelCase__						,       lowerCAmelCase__				)     ->							Optional[int]:
								import apache_beam as beam

								return pipeline | "Load Examples" >> beam.Create(lowerCAmelCase__				)





class 					lowerCAmelCase	(					datasets.BeamBasedBuilder    ):




		'''simple docstring'''







		def 							__A						(	self				)     ->							int:
								return datasets.DatasetInfo(
								    features=datasets.Features({'a': datasets.Sequence({'b': datasets.Value('string'				)}				)}				)						,       supervised_keys=lowerCAmelCase__						,       )







		def 							__A						(	self						,       lowerCAmelCase__						,       lowerCAmelCase__				)     ->							Dict:
								return [
								    datasets.SplitGenerator(name=datasets.Split.TRAIN						,       gen_kwargs={'examples': get_test_nested_examples()}				)
								]







		def 							__A						(	self						,       lowerCAmelCase__						,       lowerCAmelCase__				)     ->							Optional[Any]:
								import apache_beam as beam

								return pipeline | "Load Examples" >> beam.Create(lowerCAmelCase__				)





def     lowercase			()   ->						str:
						return [(i, {"content": content}) for i, content in enumerate(['foo', 'bar', 'foobar']  )]


def     lowercase			()   ->						Optional[Any]:
						return [(i, {"a": {"b": [content]}}) for i, content in enumerate(['foo', 'bar', 'foobar']  )]


class 					lowerCAmelCase	(					lowerCamelCase_    ):




		'''simple docstring'''







		@require_beam
		def 							__A						(	self				)     ->							Optional[int]:
								SCREAMING_SNAKE_CASE													=					len(get_test_dummy_examples()				)
								with tempfile.TemporaryDirectory() as tmp_cache_dir:
														SCREAMING_SNAKE_CASE													=					DummyBeamDataset(cache_dir=lowerCAmelCase__						,       beam_runner='DirectRunner'				)
														builder.download_and_prepare()
														self.assertTrue(
														    os.path.exists(
														        os.path.join(lowerCAmelCase__						,       builder.name						,       'default'						,       '0.0.0'						,       F'{builder.name}-train.arrow'				)				)				)
														self.assertDictEqual(builder.info.features						,       datasets.Features({'content': datasets.Value('string'				)}				)				)
														SCREAMING_SNAKE_CASE													=					builder.as_dataset()
														self.assertEqual(dset['train'].num_rows						,       lowerCAmelCase__				)
														self.assertEqual(dset['train'].info.splits['train'].num_examples						,       lowerCAmelCase__				)
														self.assertDictEqual(dset['train'][0]						,       get_test_dummy_examples()[0][1]				)
														self.assertDictEqual(
														    dset['train'][expected_num_examples - 1]						,       get_test_dummy_examples()[expected_num_examples - 1][1]				)
														self.assertTrue(
														    os.path.exists(os.path.join(lowerCAmelCase__						,       builder.name						,       'default'						,       '0.0.0'						,       'dataset_info.json'				)				)				)
														del dset







		@require_beam
		def 							__A						(	self				)     ->							int:
								import apache_beam as beam

								SCREAMING_SNAKE_CASE													=					beam.io.parquetio.WriteToParquet

								SCREAMING_SNAKE_CASE													=					len(get_test_dummy_examples()				)
								with tempfile.TemporaryDirectory() as tmp_cache_dir:
														SCREAMING_SNAKE_CASE													=					DummyBeamDataset(cache_dir=lowerCAmelCase__						,       beam_runner='DirectRunner'				)
														with patch('apache_beam.io.parquetio.WriteToParquet'				) as write_parquet_mock:
																				SCREAMING_SNAKE_CASE													=					partial(lowerCAmelCase__						,       num_shards=2				)
																				builder.download_and_prepare()
														self.assertTrue(
														    os.path.exists(
														        os.path.join(
														            lowerCAmelCase__						,       builder.name						,       'default'						,       '0.0.0'						,       F'{builder.name}-train-00000-of-00002.arrow'				)				)				)
														self.assertTrue(
														    os.path.exists(
														        os.path.join(
														            lowerCAmelCase__						,       builder.name						,       'default'						,       '0.0.0'						,       F'{builder.name}-train-00000-of-00002.arrow'				)				)				)
														self.assertDictEqual(builder.info.features						,       datasets.Features({'content': datasets.Value('string'				)}				)				)
														SCREAMING_SNAKE_CASE													=					builder.as_dataset()
														self.assertEqual(dset['train'].num_rows						,       lowerCAmelCase__				)
														self.assertEqual(dset['train'].info.splits['train'].num_examples						,       lowerCAmelCase__				)
														# Order is not preserved when sharding, so we just check that all the elements are there
														self.assertListEqual(sorted(dset['train']['content']				)						,       sorted(['foo', 'bar', 'foobar']				)				)
														self.assertTrue(
														    os.path.exists(os.path.join(lowerCAmelCase__						,       builder.name						,       'default'						,       '0.0.0'						,       'dataset_info.json'				)				)				)
														del dset







		@require_beam
		def 							__A						(	self				)     ->							Optional[Any]:
								with tempfile.TemporaryDirectory() as tmp_cache_dir:
														SCREAMING_SNAKE_CASE													=					DummyBeamDataset(cache_dir=lowerCAmelCase__				)
														self.assertRaises(datasets.builder.MissingBeamOptions						,       builder.download_and_prepare				)







		@require_beam
		def 							__A						(	self				)     ->							int:
								SCREAMING_SNAKE_CASE													=					len(get_test_nested_examples()				)
								with tempfile.TemporaryDirectory() as tmp_cache_dir:
														SCREAMING_SNAKE_CASE													=					NestedBeamDataset(cache_dir=lowerCAmelCase__						,       beam_runner='DirectRunner'				)
														builder.download_and_prepare()
														self.assertTrue(
														    os.path.exists(
														        os.path.join(lowerCAmelCase__						,       builder.name						,       'default'						,       '0.0.0'						,       F'{builder.name}-train.arrow'				)				)				)
														self.assertDictEqual(
														    builder.info.features						,       datasets.Features({'a': datasets.Sequence({'b': datasets.Value('string'				)}				)}				)				)
														SCREAMING_SNAKE_CASE													=					builder.as_dataset()
														self.assertEqual(dset['train'].num_rows						,       lowerCAmelCase__				)
														self.assertEqual(dset['train'].info.splits['train'].num_examples						,       lowerCAmelCase__				)
														self.assertDictEqual(dset['train'][0]						,       get_test_nested_examples()[0][1]				)
														self.assertDictEqual(
														    dset['train'][expected_num_examples - 1]						,       get_test_nested_examples()[expected_num_examples - 1][1]				)
														self.assertTrue(
														    os.path.exists(os.path.join(lowerCAmelCase__						,       builder.name						,       'default'						,       '0.0.0'						,       'dataset_info.json'				)				)				)
														del dset


