


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





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





     def  SCREAMING_SNAKE_CASE__       (    self					:				Optional[Any]    ,	SCREAMING_SNAKE_CASE_					:				Tuple    ,	SCREAMING_SNAKE_CASE_					:				Dict							):
        return [datasets.SplitGenerator(name=datasets.Split.TRAIN    ,	gen_kwargs={"""examples""": get_test_dummy_examples()}							)]







     def  SCREAMING_SNAKE_CASE__       (    self					:				Tuple    ,	SCREAMING_SNAKE_CASE_					:				int    ,	SCREAMING_SNAKE_CASE_					:				Optional[int]							):
        import apache_beam as beam

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



class        a					(			datasets.BeamBasedBuilder   ):





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





     def  SCREAMING_SNAKE_CASE__       (    self					:				List[Any]    ,	SCREAMING_SNAKE_CASE_					:				List[Any]    ,	SCREAMING_SNAKE_CASE_					:				Tuple							):
        return [
            datasets.SplitGenerator(name=datasets.Split.TRAIN    ,	gen_kwargs={"""examples""": get_test_nested_examples()}							)
        ]







     def  SCREAMING_SNAKE_CASE__       (    self					:				Tuple    ,	SCREAMING_SNAKE_CASE_					:				Optional[Any]    ,	SCREAMING_SNAKE_CASE_					:				Optional[int]							):
        import apache_beam as beam

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



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



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







class        a					(			a__   ):





     @require_beam
     def  SCREAMING_SNAKE_CASE__       (    self					:				Any							):
        __lowerCamelCase:				List[Any]     =	len(get_test_dummy_examples()							)
        with tempfile.TemporaryDirectory() as tmp_cache_dir:
           __lowerCamelCase:				Any     =	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:				Optional[Any]     =	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  SCREAMING_SNAKE_CASE__       (    self					:				Union[str, Any]							):
        import apache_beam as beam

        __lowerCamelCase:				int     =	beam.io.parquetio.WriteToParquet

        __lowerCamelCase:				Any     =	len(get_test_dummy_examples()							)
        with tempfile.TemporaryDirectory() as tmp_cache_dir:
           __lowerCamelCase:				Dict     =	DummyBeamDataset(cache_dir=lowerCAmelCase__    ,	beam_runner="""DirectRunner"""							)
           with patch("""apache_beam.io.parquetio.WriteToParquet"""							) as write_parquet_mock:
              __lowerCamelCase:				str     =	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:				List[str]     =	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  SCREAMING_SNAKE_CASE__       (    self					:				int							):
        with tempfile.TemporaryDirectory() as tmp_cache_dir:
           __lowerCamelCase:				int     =	DummyBeamDataset(cache_dir=lowerCAmelCase__							)
           self.assertRaises(datasets.builder.MissingBeamOptions    ,	builder.download_and_prepare							)







     @require_beam
     def  SCREAMING_SNAKE_CASE__       (    self					:				Optional[int]							):
        __lowerCamelCase:				List[Any]     =	len(get_test_nested_examples()							)
        with tempfile.TemporaryDirectory() as tmp_cache_dir:
           __lowerCamelCase:				int     =	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:				Dict     =	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






def     __lowerCAmelCase						(							snake_case	:    int = 100 )       ->							int:
   __lowerCamelCase:				List[Any]     =	0
   __lowerCamelCase:				int     =	0
   for i in range(1						,      n + 1 ):
      sum_of_squares += i**2
      sum_of_ints += i
   return sum_of_ints**2 - sum_of_squares


if __name__ == "__main__":
   print(F"""{solution() = }""")


