code: 
  
  
  
  import os
  from argparse import ArgumentParser
  from pathlib import Path
  from shutil import copyfile
  from typing import List
  
  from datasets import config
  from datasets.builder import DatasetBuilder
  from datasets.commands import BaseDatasetsCLICommand
  from datasets.download.download_config import DownloadConfig
  from datasets.download.download_manager import DownloadMode
  from datasets.load import dataset_module_factory, import_main_class
  from datasets.utils.info_utils import VerificationMode
  
  
  
  
  
  
  def 			__lowercase     (				a__							,     **a__     )       ->				List[str]:
  							return RunBeamCommand(
  							    args.dataset							,     args.name							,     args.cache_dir							,     args.beam_pipeline_options							,     args.data_dir							,     args.all_configs							,     args.save_info or args.save_infos							,     args.ignore_verifications							,     args.force_redownload							,     **a__							,     )
  
  class      UpperCAmelCase_						(     UpperCamelCase_ ):
  
  						'''simple docstring'''
  
  
  
  
  						@staticmethod
  						def  _A			(					_A	):
  
  
  
  
  
  
  
  													'''simple docstring'''
  
  													__SCREAMING_SNAKE_CASE								=      parser.add_parser('run_beam'   ,							help='Run a Beam dataset processing pipeline'	)
  													run_beam_parser.add_argument('dataset'   ,							type=_A   ,							help='Name of the dataset to download'	)
  													run_beam_parser.add_argument('--name'   ,							type=_A   ,							default=_A   ,							help='Dataset config name'	)
  													run_beam_parser.add_argument(
  													    '--cache_dir'   ,							type=_A   ,							default=_A   ,							help='Cache directory where the datasets are stored'   ,							)
  													run_beam_parser.add_argument(
  													    '--beam_pipeline_options'   ,							type=_A   ,							default=''   ,							help='Beam pipeline options, separated by commas. Example:: `--beam_pipeline_options=job_name=my-job,project=my-project`'   ,							)
  													run_beam_parser.add_argument(
  													    '--data_dir'   ,							type=_A   ,							default=_A   ,							help='Can be used to specify a manual directory to get the files from'   ,							)
  													run_beam_parser.add_argument('--all_configs'   ,							action='store_true'   ,							help='Test all dataset configurations'	)
  													run_beam_parser.add_argument('--save_info'   ,							action='store_true'   ,							help='Save the dataset infos file'	)
  													run_beam_parser.add_argument(
  													    '--ignore_verifications'   ,							action='store_true'   ,							help='Run the test without checksums and splits checks'	)
  													run_beam_parser.add_argument('--force_redownload'   ,							action='store_true'   ,							help='Force dataset redownload'	)
  													# aliases
  													run_beam_parser.add_argument('--save_infos'   ,							action='store_true'   ,							help='alias for save_info'	)
  													run_beam_parser.set_defaults(func=_A	)
  
  
  
  
  
  
  						def __init__(					self   ,							_A   ,							_A   ,							_A   ,							_A   ,							_A   ,							_A   ,							_A   ,							_A   ,							_A   ,							**_A   ,							):
  
  
  
  
  
  
  
  													'''simple docstring'''
  
  													__SCREAMING_SNAKE_CASE								=      dataset
  													__SCREAMING_SNAKE_CASE								=      name
  													__SCREAMING_SNAKE_CASE								=      cache_dir
  													__SCREAMING_SNAKE_CASE								=      beam_pipeline_options
  													__SCREAMING_SNAKE_CASE								=      data_dir
  													__SCREAMING_SNAKE_CASE								=      all_configs
  													__SCREAMING_SNAKE_CASE								=      save_infos
  													__SCREAMING_SNAKE_CASE								=      ignore_verifications
  													__SCREAMING_SNAKE_CASE								=      force_redownload
  													__SCREAMING_SNAKE_CASE								=      config_kwargs
  
  
  
  
  
  
  						def  _A			(					self	):
  
  
  
  
  
  
  
  													'''simple docstring'''
  
  													import apache_beam as beam
  
  													if self._name is not None and self._all_configs:
  																				print('Both parameters `name` and `all_configs` can\'t be used at once.'	)
  																				exit(1	)
  													__SCREAMING_SNAKE_CASE							, __SCREAMING_SNAKE_CASE								=      self._dataset, self._name
  													__SCREAMING_SNAKE_CASE								=      dataset_module_factory(_A	)
  													__SCREAMING_SNAKE_CASE								=      import_main_class(dataset_module.module_path	)
  													__SCREAMING_SNAKE_CASE								=      []
  													if self._beam_pipeline_options:
  																				__SCREAMING_SNAKE_CASE								=      beam.options.pipeline_options.PipelineOptions(
  																				    flags=[f"""--{opt.strip()}""" for opt in self._beam_pipeline_options.split(','	) if opt]	)
  													else:
  																				__SCREAMING_SNAKE_CASE								=      None
  													if self._all_configs and len(builder_cls.BUILDER_CONFIGS	) > 0:
  																				for builder_config in builder_cls.BUILDER_CONFIGS:
  																											builders.append(
  																											    builder_cls(
  																											        config_name=builder_config.name   ,							data_dir=self._data_dir   ,							hash=dataset_module.hash   ,							beam_options=_A   ,							cache_dir=self._cache_dir   ,							base_path=dataset_module.builder_kwargs.get('base_path'	)   ,							)	)
  													else:
  																				builders.append(
  																				    builder_cls(
  																				        config_name=_A   ,							data_dir=self._data_dir   ,							beam_options=_A   ,							cache_dir=self._cache_dir   ,							base_path=dataset_module.builder_kwargs.get('base_path'	)   ,							**self._config_kwargs   ,							)	)
  
  													for builder in builders:
  																				builder.download_and_prepare(
  																				    download_mode=DownloadMode.REUSE_CACHE_IF_EXISTS
  																				    if not self._force_redownload
  																				    else DownloadMode.FORCE_REDOWNLOAD   ,							download_config=DownloadConfig(cache_dir=config.DOWNLOADED_DATASETS_PATH	)   ,							verification_mode=VerificationMode.NO_CHECKS
  																				    if self._ignore_verifications
  																				    else VerificationMode.ALL_CHECKS   ,							try_from_hf_gcs=_A   ,							)
  																				if self._save_infos:
  																											builder._save_infos()
  
  													print('Apache beam run successful.'	)
  
  													# If save_infos=True, the dataset infos file is created next to the loaded module file.
  													# Let's move it to the original directory of the dataset script, to allow the user to
  													# upload them on S3 at the same time afterwards.
  													if self._save_infos:
  																				__SCREAMING_SNAKE_CASE								=      os.path.join(builder_cls.get_imported_module_dir()   ,							config.DATASETDICT_INFOS_FILENAME	)
  
  																				__SCREAMING_SNAKE_CASE								=      Path(_A	).name + '.py'
  
  																				__SCREAMING_SNAKE_CASE								=      os.path.join(_A   ,							_A	)
  																				if os.path.isfile(_A	):
  																											__SCREAMING_SNAKE_CASE								=      os.path.dirname(_A	)
  																				elif os.path.isfile(_A	):
  																											__SCREAMING_SNAKE_CASE								=      path
  																				else:  # in case of a remote dataset
  																											print(f"""Dataset Infos file saved at {dataset_infos_path}"""	)
  																											exit(1	)
  
  																				# Move datasetinfo back to the user
  																				__SCREAMING_SNAKE_CASE								=      os.path.join(_A   ,							config.DATASETDICT_INFOS_FILENAME	)
  																				copyfile(_A   ,							_A	)
  																				print(f"""Dataset Infos file saved at {user_dataset_infos_path}"""	)
  
  
  
  
  
  
  
style_context: 
  
  
  
  from __future__ import annotations
  
  from collections.abc import Iterable
  from dataclasses import dataclass
  from typing import Any
  
  class      UpperCAmelCase_						:
  
  						'''simple docstring'''
  
  
  
  
  						UpperCamelCase__    :				List[str]       						=						('''_front''', '''_back''', '''_len''')
  
  						@dataclass
  						class      UpperCAmelCase_						:
  
  												'''simple docstring'''
  
  
  
  
  												UpperCamelCase__    :				Any       						=						None
  												UpperCamelCase__    :				Deque._Node | None       						=						None
  												UpperCamelCase__    :				Deque._Node | None       						=						None
  
  
  
  
  
  
  
  						class      UpperCAmelCase_						:
  
  												'''simple docstring'''
  
  
  
  
  												UpperCamelCase__    :				Optional[int]       						=						('''_cur''',)
  
  
  
  
  
  
  												def __init__(					self   ,							_A	):
  
  
  
  
  
  
  
  																			'''simple docstring'''
  
  																			__SCREAMING_SNAKE_CASE								=      cur
  
  
  
  
  
  
  												def __iter__(					self	):
  
  
  
  
  
  
  
  																			'''simple docstring'''
  
  																			return self
  
  
  
  
  
  
  												def  _A			(					self	):
  
  
  
  
  
  
  
  																			'''simple docstring'''
  
  																			if self._cur is None:
  																										# finished iterating
  																										raise StopIteration
  																			__SCREAMING_SNAKE_CASE								=      self._cur.val
  																			__SCREAMING_SNAKE_CASE								=      self._cur.next_node
  
  																			return val
  
  
  
  
  
  
  
  						def __init__(					self   ,							_A = None	):
  
  
  
  
  
  
  
  													'''simple docstring'''
  
  													__SCREAMING_SNAKE_CASE								=      None
  													__SCREAMING_SNAKE_CASE								=      None
  													__SCREAMING_SNAKE_CASE								=      0
  
  													if iterable is not None:
  																				# append every value to the deque
  																				for val in iterable:
  																											self.append(_A	)
  
  
  
  
  
  
  						def  _A			(					self   ,							_A	):
  
  
  
  
  
  
  
  													'''simple docstring'''
  
  													__SCREAMING_SNAKE_CASE								=      self._Node(_A   ,							_A   ,							_A	)
  													if self.is_empty():
  																				# front = back
  																				__SCREAMING_SNAKE_CASE								=      __SCREAMING_SNAKE_CASE								=      node
  																				__SCREAMING_SNAKE_CASE								=      1
  													else:
  																				# connect nodes
  																				__SCREAMING_SNAKE_CASE								=      node
  																				__SCREAMING_SNAKE_CASE								=      self._back
  																				__SCREAMING_SNAKE_CASE								=      node  # assign new back to the new node
  
  																				self._len += 1
  
  																				# make sure there were no errors
  																				assert not self.is_empty(), "Error on appending value."
  
  
  
  
  
  
  						def  _A			(					self   ,							_A	):
  
  
  
  
  
  
  
  													'''simple docstring'''
  
  													__SCREAMING_SNAKE_CASE								=      self._Node(_A   ,							_A   ,							_A	)
  													if self.is_empty():
  																				# front = back
  																				__SCREAMING_SNAKE_CASE								=      __SCREAMING_SNAKE_CASE								=      node
  																				__SCREAMING_SNAKE_CASE								=      1
  													else:
  																				# connect nodes
  																				__SCREAMING_SNAKE_CASE								=      self._front
  																				__SCREAMING_SNAKE_CASE								=      node
  																				__SCREAMING_SNAKE_CASE								=      node  # assign new front to the new node
  
  																				self._len += 1
  
  																				# make sure there were no errors
  																				assert not self.is_empty(), "Error on appending value."
  
  
  
  
  
  
  						def  _A			(					self   ,							_A	):
  
  
  
  
  
  
  
  													'''simple docstring'''
  
  													for val in iterable:
  																				self.append(_A	)
  
  
  
  
  
  
  						def  _A			(					self   ,							_A	):
  
  
  
  
  
  
  
  													'''simple docstring'''
  
  													for val in iterable:
  																				self.appendleft(_A	)
  
  
  
  
  
  
  						def  _A			(					self	):
  
  
  
  
  
  
  
  													'''simple docstring'''
  
  													assert not self.is_empty(), "Deque is empty."
  
  													__SCREAMING_SNAKE_CASE								=      self._back
  													__SCREAMING_SNAKE_CASE								=      self._back.prev_node  # set new back
  													# drop the last node - python will deallocate memory automatically
  													__SCREAMING_SNAKE_CASE								=      None
  
  													self._len -= 1
  
  													return topop.val
  
  
  
  
  
  
  						def  _A			(					self	):
  
  
  
  
  
  
  
  													'''simple docstring'''
  
  													assert not self.is_empty(), "Deque is empty."
  
  													__SCREAMING_SNAKE_CASE								=      self._front
  													__SCREAMING_SNAKE_CASE								=      self._front.next_node  # set new front and drop the first node
  													__SCREAMING_SNAKE_CASE								=      None
  
  													self._len -= 1
  
  													return topop.val
  
  
  
  
  
  
  						def  _A			(					self	):
  
  
  
  
  
  
  
  													'''simple docstring'''
  
  													return self._front is None
  
  
  
  
  
  
  						def __len__(					self	):
  
  
  
  
  
  
  
  													'''simple docstring'''
  
  													return self._len
  
  
  
  
  
  
  						def __eq__(					self   ,							_A	):
  
  
  
  
  
  
  
  													'''simple docstring'''
  
  													if not isinstance(_A   ,							_A	):
  																				return NotImplemented
  
  													__SCREAMING_SNAKE_CASE								=      self._front
  													__SCREAMING_SNAKE_CASE								=      other._front
  
  													# if the length of the dequeues are not the same, they are not equal
  													if len(self	) != len(_A	):
  																				return False
  
  													while me is not None and oth is not None:
  																				# compare every value
  																				if me.val != oth.val:
  																											return False
  																				__SCREAMING_SNAKE_CASE								=      me.next_node
  																				__SCREAMING_SNAKE_CASE								=      oth.next_node
  
  													return True
  
  
  
  
  
  
  						def __iter__(					self	):
  
  
  
  
  
  
  
  													'''simple docstring'''
  
  													return Deque._Iterator(self._front	)
  
  
  
  
  
  
  						def __repr__(					self	):
  
  
  
  
  
  
  
  													'''simple docstring'''
  
  													__SCREAMING_SNAKE_CASE								=      []
  													__SCREAMING_SNAKE_CASE								=      self._front
  													while aux is not None:
  																				# append the values in a list to display
  																				values_list.append(aux.val	)
  																				__SCREAMING_SNAKE_CASE								=      aux.next_node
  
  													return f"""[{", ".join(repr(_A	) for val in values_list	)}]"""
  
  
  if __name__ == "__main__":
  			import doctest
  
  			doctest.testmod()
  
  
  
  
  
  
  