





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



     '''simple docstring'''

     def        SCREAMING_SNAKE_CASE__						(							self		:					List[Any] )  ->	Optional[Any]:




           '''simple docstring'''


           return datasets.DatasetInfo(
               features=datasets.Features({"""content""": datasets.Value("""string""" )} )		,supervised_keys=lowerCamelCase__		,)






     def        SCREAMING_SNAKE_CASE__						(							self		:					Any		,lowerCamelCase__		:					Tuple		,lowerCamelCase__		:					Dict )  ->	List[str]:




           '''simple docstring'''


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






     def        SCREAMING_SNAKE_CASE__						(							self		:					Optional[int]		,lowerCamelCase__		:					List[Any]		,lowerCamelCase__		:					Any )  ->	Tuple:




           '''simple docstring'''


           import apache_beam as beam

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

class 	UpperCamelCase__				(   datasets.BeamBasedBuilder     ):



     '''simple docstring'''

     def        SCREAMING_SNAKE_CASE__						(							self		:					Any )  ->	List[str]:




           '''simple docstring'''


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






     def        SCREAMING_SNAKE_CASE__						(							self		:					List[str]		,lowerCamelCase__		:					Tuple		,lowerCamelCase__		:					List[Any] )  ->	List[str]:




           '''simple docstring'''


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






     def        SCREAMING_SNAKE_CASE__						(							self		:					List[Any]		,lowerCamelCase__		:					List[Any]		,lowerCamelCase__		:					int )  ->	Tuple:




           '''simple docstring'''


           import apache_beam as beam

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



def      __lowercase     (   )				->     Tuple:


      '''simple docstring'''


      return [(i, {"content": content}) for i, content in enumerate(["""foo""", """bar""", """foobar"""]			)]



def      __lowercase     (   )				->     Union[str, Any]:


      '''simple docstring'''


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



class 	UpperCamelCase__				(   lowerCAmelCase_     ):



     '''simple docstring'''

     @require_beam
     def        SCREAMING_SNAKE_CASE__						(							self		:					Union[str, Any] )  ->	Optional[Any]:




           '''simple docstring'''


           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        SCREAMING_SNAKE_CASE__						(							self		:					Any )  ->	Dict:




           '''simple docstring'''


           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        SCREAMING_SNAKE_CASE__						(							self		:					Any )  ->	Tuple:




           '''simple docstring'''


           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        SCREAMING_SNAKE_CASE__						(							self		:					int )  ->	Tuple:




           '''simple docstring'''


           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









# We ignore warnings about stepping the scheduler since we step it ourselves during gradient accumulation

import warnings

from .state import AcceleratorState, GradientState


warnings.filterwarnings("""ignore""", category=UserWarning, module="""torch.optim.lr_scheduler""")

class 	UpperCamelCase__				:



     '''simple docstring'''

     def __init__(							self		:					Optional[Any]		,lowerCamelCase__		:					Optional[int]		,lowerCamelCase__		:					Tuple		,lowerCamelCase__		:					bool = True		,lowerCamelCase__		:					bool = False )  ->	Tuple:




           '''simple docstring'''


           SCREAMING_SNAKE_CASE	=		scheduler
           SCREAMING_SNAKE_CASE	=		optimizers if isinstance(lowerCamelCase__		,(list, tuple) ) else [optimizers]
           SCREAMING_SNAKE_CASE	=		split_batches
           SCREAMING_SNAKE_CASE	=		step_with_optimizer
           SCREAMING_SNAKE_CASE	=		GradientState()






     def        SCREAMING_SNAKE_CASE__						(							self		:					str		,*lowerCamelCase__		:					Union[str, Any]		,**lowerCamelCase__		:					Optional[int] )  ->	Dict:




           '''simple docstring'''


           if not self.step_with_optimizer:
                 # No link between scheduler and optimizer -> just step
                 self.scheduler.step(*lowerCamelCase__		,**lowerCamelCase__ )
                 return

           # Otherwise, first make sure the optimizer was stepped.
           if not self.gradient_state.sync_gradients:
                 if self.gradient_state.adjust_scheduler:
                       self.scheduler._step_count += 1
                 return

           for opt in self.optimizers:
                 if opt.step_was_skipped:
                       return
           if self.split_batches:
                 # Split batches -> the training dataloader batch size is not changed so one step per training step
                 self.scheduler.step(*lowerCamelCase__		,**lowerCamelCase__ )
           else:
                 # Otherwise the training dataloader batch size was multiplied by `num_processes`, so we need to do
                 # num_processes steps per training step
                 SCREAMING_SNAKE_CASE	=		AcceleratorState().num_processes
                 for _ in range(lowerCamelCase__ ):
                       # Special case when using OneCycle and `drop_last` was not used
                       if hasattr(self.scheduler		,"""total_steps""" ):
                             if self.scheduler._step_count <= self.scheduler.total_steps:
                                   self.scheduler.step(*lowerCamelCase__		,**lowerCamelCase__ )
                       else:
                             self.scheduler.step(*lowerCamelCase__		,**lowerCamelCase__ )






     def        SCREAMING_SNAKE_CASE__						(							self		:					int )  ->	Optional[Any]:




           '''simple docstring'''


           return self.scheduler.get_last_lr()






     def        SCREAMING_SNAKE_CASE__						(							self		:					List[Any] )  ->	Optional[Any]:




           '''simple docstring'''


           return self.scheduler.state_dict()






     def        SCREAMING_SNAKE_CASE__						(							self		:					Optional[int]		,lowerCamelCase__		:					str )  ->	int:




           '''simple docstring'''


           self.scheduler.load_state_dict(lowerCamelCase__ )






     def        SCREAMING_SNAKE_CASE__						(							self		:					List[Any] )  ->	Optional[Any]:




           '''simple docstring'''


           return self.scheduler.get_lr()






     def        SCREAMING_SNAKE_CASE__						(							self		:					Dict		,*lowerCamelCase__		:					int		,**lowerCamelCase__		:					Tuple )  ->	Optional[int]:




           '''simple docstring'''


           return self.scheduler.print_lr(*lowerCamelCase__		,**lowerCamelCase__ )


