





from typing import List, Optional, TypeVar

from .arrow_dataset import Dataset, _concatenate_map_style_datasets, _interleave_map_style_datasets
from .dataset_dict import DatasetDict, IterableDatasetDict
from .info import DatasetInfo
from .iterable_dataset import IterableDataset, _concatenate_iterable_datasets, _interleave_iterable_datasets
from .splits import NamedSplit
from .utils import logging
from .utils.py_utils import Literal


A_												=     logging.get_logger(__name__)


A_												=     TypeVar("DatasetType", Dataset, IterableDataset)





def    __UpperCamelCase      (    a,     a = None,     a = None,     a = None,     a = None,     a = "first_exhausted",     ) ->DatasetType:
					from .arrow_dataset import Dataset
					from .iterable_dataset import IterableDataset

					if not datasets:
										raise ValueError("Unable to interleave an empty list of datasets.")
					for i, dataset in enumerate(a):
										if not isinstance(a,     (Dataset, IterableDataset)):
															if isinstance(a,     (DatasetDict, IterableDatasetDict)):
																				if not dataset:
																									raise ValueError(
																									    f"Expected a list of Dataset objects or a list of IterableDataset objects, but element at position {i} "
																									    "is an empty dataset dictionary.")
																				raise ValueError(
																				    f"Dataset at position {i} has at least one split: {list(a)}\n"
																				    f"Please pick one to interleave with the other datasets, for example: dataset['{next(iter(a))}']")
															raise ValueError(
															    f"Expected a list of Dataset objects or a list of IterableDataset objects, but element at position {i} is a {type(a).__name__}.")
										if i == 0:
															lowerCamelCase__    ,		lowerCamelCase__       =       (
															    (Dataset, IterableDataset) if isinstance(a,     a) else (IterableDataset, Dataset)
															)
										elif not isinstance(a,     a):
															raise ValueError(
															    f"Unable to interleave a {dataset_type.__name__} (at position 0) with a {other_type.__name__} (at position {i}). Expected a list of Dataset objects or a list of IterableDataset objects.")
					if stopping_strategy not in ["first_exhausted", "all_exhausted"]:
										raise ValueError(f"{stopping_strategy} is not supported. Please enter a valid stopping_strategy.")
					if dataset_type is Dataset:
										return _interleave_map_style_datasets(
										    a,     a,     a,     info=a,     split=a,     stopping_strategy=a)
					else:
										return _interleave_iterable_datasets(
										    a,     a,     a,     info=a,     split=a,     stopping_strategy=a)







def    __UpperCamelCase      (    a,     a = None,     a = None,     a = 0,     ) ->DatasetType:

					if not dsets:
										raise ValueError("Unable to concatenate an empty list of datasets.")
					for i, dataset in enumerate(a):
										if not isinstance(a,     (Dataset, IterableDataset)):
															if isinstance(a,     (DatasetDict, IterableDatasetDict)):
																				if not dataset:
																									raise ValueError(
																									    f"Expected a list of Dataset objects or a list of IterableDataset objects, but element at position {i} "
																									    "is an empty dataset dictionary.")
																				raise ValueError(
																				    f"Dataset at position {i} has at least one split: {list(a)}\n"
																				    f"Please pick one to interleave with the other datasets, for example: dataset['{next(iter(a))}']")
															raise ValueError(
															    f"Expected a list of Dataset objects or a list of IterableDataset objects, but element at position {i} is a {type(a).__name__}.")
										if i == 0:
															lowerCamelCase__    ,		lowerCamelCase__       =       (
															    (Dataset, IterableDataset) if isinstance(a,     a) else (IterableDataset, Dataset)
															)
										elif not isinstance(a,     a):
															raise ValueError(
															    f"Unable to interleave a {dataset_type.__name__} (at position 0) with a {other_type.__name__} (at position {i}). Expected a list of Dataset objects or a list of IterableDataset objects.")
					if dataset_type is Dataset:
										return _concatenate_map_style_datasets(a,     info=a,     split=a,     axis=a)
					else:
										return _concatenate_iterable_datasets(a,     info=a,     split=a,     axis=a)








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 SCREAMING_SNAKE_CASE_ (					datasets.BeamBasedBuilder   ):





					"""simple docstring"""


					def 					__magic_name__ ( self	):
										return datasets.DatasetInfo(
										    features=datasets.Features({"content": datasets.Value("string"	)}	)	,  supervised_keys=_lowerCAmelCase	,  )





					def 					__magic_name__ ( self	,  _lowerCAmelCase	,  _lowerCAmelCase	):
										return [datasets.SplitGenerator(name=datasets.Split.TRAIN	,  gen_kwargs={"examples": get_test_dummy_examples()}	)]






					def 					__magic_name__ ( self	,  _lowerCAmelCase	,  _lowerCAmelCase	):
										import apache_beam as beam

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

class SCREAMING_SNAKE_CASE_ (					datasets.BeamBasedBuilder   ):





					"""simple docstring"""


					def 					__magic_name__ ( self	):
										return datasets.DatasetInfo(
										    features=datasets.Features({"a": datasets.Sequence({"b": datasets.Value("string"	)}	)}	)	,  supervised_keys=_lowerCAmelCase	,  )





					def 					__magic_name__ ( self	,  _lowerCAmelCase	,  _lowerCAmelCase	):
										return [
										    datasets.SplitGenerator(name=datasets.Split.TRAIN	,  gen_kwargs={"examples": get_test_nested_examples()}	)
										]






					def 					__magic_name__ ( self	,  _lowerCAmelCase	,  _lowerCAmelCase	):
										import apache_beam as beam

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





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





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







class SCREAMING_SNAKE_CASE_ (					lowercase_   ):





					"""simple docstring"""


					@require_beam
					def 					__magic_name__ ( self	):
										lowerCamelCase__       =       len(get_test_dummy_examples()	)
										with tempfile.TemporaryDirectory() as tmp_cache_dir:
															lowerCamelCase__       =       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"	)}	)	)
															lowerCamelCase__       =       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 					__magic_name__ ( self	):
										import apache_beam as beam

										lowerCamelCase__       =       beam.io.parquetio.WriteToParquet

										lowerCamelCase__       =       len(get_test_dummy_examples()	)
										with tempfile.TemporaryDirectory() as tmp_cache_dir:
															lowerCamelCase__       =       DummyBeamDataset(cache_dir=_lowerCAmelCase	,  beam_runner="DirectRunner"	)
															with patch("apache_beam.io.parquetio.WriteToParquet"	) as write_parquet_mock:
																				lowerCamelCase__       =       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"	)}	)	)
															lowerCamelCase__       =       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 					__magic_name__ ( self	):
										with tempfile.TemporaryDirectory() as tmp_cache_dir:
															lowerCamelCase__       =       DummyBeamDataset(cache_dir=_lowerCAmelCase	)
															self.assertRaises(datasets.builder.MissingBeamOptions	,  builder.download_and_prepare	)






					@require_beam
					def 					__magic_name__ ( self	):
										lowerCamelCase__       =       len(get_test_nested_examples()	)
										with tempfile.TemporaryDirectory() as tmp_cache_dir:
															lowerCamelCase__       =       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"	)}	)}	)	)
															lowerCamelCase__       =       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

