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 		_SCREAMING_SNAKE_CASE    (   lowerCAmelCase__       ,**lowerCAmelCase__	):
    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       ,**lowerCAmelCase__       ,)
  
  
  
  
  
  
  class 	UpperCamelCase_     (     snake_case__       ):
  
  
  
       @staticmethod
       def      __a    (		lowerCamelCase    :    ArgumentParser	):
         lowerCamelCase_  :    Tuple   	=			parser.add_parser('run_beam'					,			help='Run a Beam dataset processing pipeline'	)
         run_beam_parser.add_argument('dataset'					,			type=lowerCamelCase					,			help='Name of the dataset to download'	)
         run_beam_parser.add_argument('--name'					,			type=lowerCamelCase					,			default=lowerCamelCase					,			help='Dataset config name'	)
         run_beam_parser.add_argument(
             '--cache_dir'					,			type=lowerCamelCase					,			default=lowerCamelCase					,			help='Cache directory where the datasets are stored'					,			)
         run_beam_parser.add_argument(
             '--beam_pipeline_options'					,			type=lowerCamelCase					,			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=lowerCamelCase					,			default=lowerCamelCase					,			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=lowerCamelCase	)
  
  
  
       def __init__(		self    :    Optional[Any]					,			lowerCamelCase    :    str					,			lowerCamelCase    :    str					,			lowerCamelCase    :    str					,			lowerCamelCase    :    str					,			lowerCamelCase    :    str					,			lowerCamelCase    :    bool					,			lowerCamelCase    :    bool					,			lowerCamelCase    :    bool					,			lowerCamelCase    :    bool					,			**lowerCamelCase    :    Any					,			):
         lowerCamelCase_  :    int   	=			dataset
         lowerCamelCase_  :    List[Any]   	=			name
         lowerCamelCase_  :    List[Any]   	=			cache_dir
         lowerCamelCase_  :    str   	=			beam_pipeline_options
         lowerCamelCase_  :    Any   	=			data_dir
         lowerCamelCase_  :    int   	=			all_configs
         lowerCamelCase_  :    int   	=			save_infos
         lowerCamelCase_  :    Optional[int]   	=			ignore_verifications
         lowerCamelCase_  :    List[Any]   	=			force_redownload
         lowerCamelCase_  :    Optional[Any]   	=			config_kwargs
  
  
  
  
  
       def      __a    (		self    :    Optional[int]	):
         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	)
         lowerCamelCase_  :    List[Any]   	=			self._dataset, self._name
         lowerCamelCase_  :    List[Any]   	=			dataset_module_factory(lowerCamelCase	)
         lowerCamelCase_  :    Optional[int]   	=			import_main_class(dataset_module.module_path	)
         lowerCamelCase_  :    List[DatasetBuilder]   	=			[]
         if self._beam_pipeline_options:
           lowerCamelCase_  :    str   	=			beam.options.pipeline_options.PipelineOptions(
               flags=[F"--{opt.strip()}" for opt in self._beam_pipeline_options.split(','	) if opt]	)
         else:
           lowerCamelCase_  :    Tuple   	=			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=lowerCamelCase					,			cache_dir=self._cache_dir					,			base_path=dataset_module.builder_kwargs.get('base_path'	)					,			)	)
         else:
           builders.append(
               builder_cls(
                   config_name=lowerCamelCase					,			data_dir=self._data_dir					,			beam_options=lowerCamelCase					,			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=lowerCamelCase					,			)
           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:
           lowerCamelCase_  :    Tuple   	=			os.path.join(builder_cls.get_imported_module_dir()					,			config.DATASETDICT_INFOS_FILENAME	)
  
           lowerCamelCase_  :    str   	=			Path(lowerCamelCase	).name + '.py'
  
           lowerCamelCase_  :    List[str]   	=			os.path.join(lowerCamelCase					,			lowerCamelCase	)
           if os.path.isfile(lowerCamelCase	):
             lowerCamelCase_  :    List[str]   	=			os.path.dirname(lowerCamelCase	)
           elif os.path.isfile(lowerCamelCase	):
             lowerCamelCase_  :    Tuple   	=			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
           lowerCamelCase_  :    Tuple   	=			os.path.join(lowerCamelCase					,			config.DATASETDICT_INFOS_FILENAME	)
           copyfile(lowerCamelCase					,			lowerCamelCase	)
           print(F"Dataset Infos file saved at {user_dataset_infos_path}"	)
  
style_context: 
  
  
  
  
  
  class 	UpperCamelCase_     :
  
  
  
  					def __init__(		self    :    List[Any]					,			lowerCamelCase    :    int	):
  							lowerCamelCase_  :    Union[str, Any]   	=			val
  							# Number of nodes in left subtree
  							lowerCamelCase_  :    Dict   	=			0
  							lowerCamelCase_  :    str   	=			None
  							lowerCamelCase_  :    int   	=			None
  							lowerCamelCase_  :    Dict   	=			None
  
  
  
  
  
  					def      __a    (		self    :    Optional[int]					,			lowerCamelCase    :    List[Any]	):
  							assert self.left_tree_size == other.left_tree_size, "Unequal Sizes of Blocks"
  
  							if self.val < other.val:
  									lowerCamelCase_  :    int   	=			self.right
  									lowerCamelCase_  :    Any   	=			None
  									if self.right:
  											lowerCamelCase_  :    Optional[Any]   	=			other
  									lowerCamelCase_  :    Union[str, Any]   	=			other
  									lowerCamelCase_  :    List[Any]   	=			self.left_tree_size * 2 + 1
  									return self
  							else:
  									lowerCamelCase_  :    str   	=			other.right
  									lowerCamelCase_  :    List[Any]   	=			None
  									if other.right:
  											lowerCamelCase_  :    Union[str, Any]   	=			self
  									lowerCamelCase_  :    Optional[int]   	=			self
  									lowerCamelCase_  :    Optional[int]   	=			other.left_tree_size * 2 + 1
  									return other
  
  
  
  
  
  
  class 	UpperCamelCase_     :
  
  
  
  					def __init__(		self    :    List[str]					,			lowerCamelCase    :    Optional[int]=None					,			lowerCamelCase    :    str=None					,			lowerCamelCase    :    List[Any]=0	):
  							lowerCamelCase_  :    Any   	=			heap_size
  							lowerCamelCase_  :    int   	=			bottom_root
  							lowerCamelCase_  :    Any   	=			min_node
  
  
  
  					def      __a    (		self    :    int					,			lowerCamelCase    :    int	):
  
  							# Empty heaps corner cases
  							if other.size == 0:
  									return None
  							if self.size == 0:
  									lowerCamelCase_  :    List[str]   	=			other.size
  									lowerCamelCase_  :    Tuple   	=			other.bottom_root
  									lowerCamelCase_  :    Tuple   	=			other.min_node
  									return None
  							# Update size
  							lowerCamelCase_  :    Optional[Any]   	=			self.size + other.size
  
  							# Update min.node
  							if self.min_node.val > other.min_node.val:
  									lowerCamelCase_  :    Optional[Any]   	=			other.min_node
  							# Merge
  
  							# Order roots by left_subtree_size
  							lowerCamelCase_  :    Dict   	=			[]
  							lowerCamelCase_	,			lowerCamelCase_  :    Optional[Any]   	=			self.bottom_root, other.bottom_root
  							while i or j:
  									if i and ((not j) or i.left_tree_size < j.left_tree_size):
  											combined_roots_list.append((i, True)	)
  											lowerCamelCase_  :    Tuple   	=			i.parent
  									else:
  											combined_roots_list.append((j, False)	)
  											lowerCamelCase_  :    Optional[Any]   	=			j.parent
          # Insert links between them
  							for i in range(len(lowerCamelCase	) - 1	):
  									if combined_roots_list[i][1] != combined_roots_list[i + 1][1]:
  											lowerCamelCase_  :    Dict   	=			combined_roots_list[i + 1][0]
  											lowerCamelCase_  :    Optional[int]   	=			combined_roots_list[i][0]
          # Consecutively merge roots with same left_tree_size
  							lowerCamelCase_  :    Any   	=			combined_roots_list[0][0]
  							while i.parent:
  									if (
  									    (i.left_tree_size == i.parent.left_tree_size) and (not i.parent.parent)
  									) or (
  									    i.left_tree_size == i.parent.left_tree_size
  									    and i.left_tree_size != i.parent.parent.left_tree_size
  									):
  											# Neighbouring Nodes
  											lowerCamelCase_  :    Dict   	=			i.left
  											lowerCamelCase_  :    int   	=			i.parent.parent
  
  											# Merging trees
  											lowerCamelCase_  :    str   	=			i.merge_trees(i.parent	)
  
  											# Updating links
  											lowerCamelCase_  :    Union[str, Any]   	=			previous_node
  											lowerCamelCase_  :    List[Any]   	=			next_node
  											if previous_node:
  													lowerCamelCase_  :    List[str]   	=			i
  											if next_node:
  													lowerCamelCase_  :    Dict   	=			i
  									else:
  											lowerCamelCase_  :    List[str]   	=			i.parent
          # Updating self.bottom_root
  							while i.left:
  									lowerCamelCase_  :    Union[str, Any]   	=			i.left
  							lowerCamelCase_  :    Any   	=			i
  
  							# Update other
  							lowerCamelCase_  :    Optional[int]   	=			self.size
  							lowerCamelCase_  :    Optional[Any]   	=			self.bottom_root
  							lowerCamelCase_  :    str   	=			self.min_node
  
  							# Return the merged heap
  							return self
  
  
  
  					def      __a    (		self    :    Any					,			lowerCamelCase    :    Dict	):
  							if self.size == 0:
  									lowerCamelCase_  :    Tuple   	=			Node(lowerCamelCase	)
  									lowerCamelCase_  :    Optional[int]   	=			1
  									lowerCamelCase_  :    Optional[int]   	=			self.bottom_root
  							else:
  									# Create new node
  									lowerCamelCase_  :    List[Any]   	=			Node(lowerCamelCase	)
  
  									# Update size
  									self.size += 1
  
  									# update min_node
  									if val < self.min_node.val:
  											lowerCamelCase_  :    List[str]   	=			new_node
  									# Put new_node as a bottom_root in heap
  									lowerCamelCase_  :    int   	=			new_node
  									lowerCamelCase_  :    Tuple   	=			self.bottom_root
  									lowerCamelCase_  :    Any   	=			new_node
  
  									# Consecutively merge roots with same left_tree_size
  									while (
  									    self.bottom_root.parent
  									    and self.bottom_root.left_tree_size
  									    == self.bottom_root.parent.left_tree_size
  									):
  											# Next node
  											lowerCamelCase_  :    Any   	=			self.bottom_root.parent.parent
  
  											# Merge
  											lowerCamelCase_  :    int   	=			self.bottom_root.merge_trees(self.bottom_root.parent	)
  
  											# Update Links
  											lowerCamelCase_  :    Dict   	=			next_node
  											lowerCamelCase_  :    Optional[Any]   	=			None
  											if next_node:
  													lowerCamelCase_  :    Union[str, Any]   	=			self.bottom_root
  
  
  
  					def      __a    (		self    :    Dict	):
  							return self.min_node.val
  
  
  
  					def      __a    (		self    :    int	):
  							return self.size == 0
  
  
  
  					def      __a    (		self    :    List[Any]	):
  							lowerCamelCase_  :    Tuple   	=			self.min_node.val
  
  							# Last element in heap corner case
  							if self.size == 1:
  									# Update size
  									lowerCamelCase_  :    Optional[Any]   	=			0
  
  									# Update bottom root
  									lowerCamelCase_  :    Any   	=			None
  
  									# Update min_node
  									lowerCamelCase_  :    List[str]   	=			None
  
  									return min_value
  							# No right subtree corner case
  							# The structure of the tree implies that this should be the bottom root
  							# and there is at least one other root
  							if self.min_node.right is None:
  									# Update size
  									self.size -= 1
  
  									# Update bottom root
  									lowerCamelCase_  :    Union[str, Any]   	=			self.bottom_root.parent
  									lowerCamelCase_  :    Union[str, Any]   	=			None
  
  									# Update min_node
  									lowerCamelCase_  :    Union[str, Any]   	=			self.bottom_root
  									lowerCamelCase_  :    Union[str, Any]   	=			self.bottom_root.parent
  									while i:
  											if i.val < self.min_node.val:
  													lowerCamelCase_  :    Union[str, Any]   	=			i
  											lowerCamelCase_  :    Optional[int]   	=			i.parent
  									return min_value
  							# General case
  							# Find the BinomialHeap of the right subtree of min_node
  							lowerCamelCase_  :    Dict   	=			self.min_node.right
  							lowerCamelCase_  :    Dict   	=			None
  							lowerCamelCase_  :    Tuple   	=			bottom_of_new
  							lowerCamelCase_  :    Dict   	=			1
  
  							# Size, min_node and bottom_root
  							while bottom_of_new.left:
  									lowerCamelCase_  :    Any   	=			size_of_new * 2 + 1
  									lowerCamelCase_  :    Tuple   	=			bottom_of_new.left
  									if bottom_of_new.val < min_of_new.val:
  											lowerCamelCase_  :    int   	=			bottom_of_new
          # Corner case of single root on top left path
  							if (not self.min_node.left) and (not self.min_node.parent):
  									lowerCamelCase_  :    Union[str, Any]   	=			size_of_new
  									lowerCamelCase_  :    Union[str, Any]   	=			bottom_of_new
  									lowerCamelCase_  :    Tuple   	=			min_of_new
  									# print("Single root, multiple nodes case")
  									return min_value
  							# Remaining cases
  							# Construct heap of right subtree
  							lowerCamelCase_  :    Optional[Any]   	=			BinomialHeap(
  							    bottom_root=lowerCamelCase					,			min_node=lowerCamelCase					,			heap_size=lowerCamelCase	)
  
  							# Update size
  							lowerCamelCase_  :    Union[str, Any]   	=			self.size - 1 - size_of_new
  
  							# Neighbour nodes
  							lowerCamelCase_  :    Union[str, Any]   	=			self.min_node.left
  							lowerCamelCase_  :    Optional[Any]   	=			self.min_node.parent
  
  							# Initialize new bottom_root and min_node
  							lowerCamelCase_  :    Optional[int]   	=			previous_node or next_node
  							lowerCamelCase_  :    Tuple   	=			next_node
  
  							# Update links of previous_node and search below for new min_node and
  							# bottom_root
  							if previous_node:
  									lowerCamelCase_  :    Dict   	=			next_node
  
  									# Update bottom_root and search for min_node below
  									lowerCamelCase_  :    int   	=			previous_node
  									lowerCamelCase_  :    Tuple   	=			previous_node
  									while self.bottom_root.left:
  											lowerCamelCase_  :    Union[str, Any]   	=			self.bottom_root.left
  											if self.bottom_root.val < self.min_node.val:
  													lowerCamelCase_  :    Any   	=			self.bottom_root
  							if next_node:
  									lowerCamelCase_  :    Any   	=			previous_node
  
  									# Search for new min_node above min_node
  									lowerCamelCase_  :    List[str]   	=			next_node
  									while i:
  											if i.val < self.min_node.val:
  													lowerCamelCase_  :    List[Any]   	=			i
  											lowerCamelCase_  :    Dict   	=			i.parent
          # Merge heaps
  							self.merge_heaps(lowerCamelCase	)
  
  							return min_value
  
  
  
  					def      __a    (		self    :    Optional[int]	):
  							lowerCamelCase_  :    List[Any]   	=			self.bottom_root
  							while top_root.parent:
  									lowerCamelCase_  :    Tuple   	=			top_root.parent
  							# preorder
  							lowerCamelCase_  :    str   	=			[]
  							self.__traversal(lowerCamelCase					,			lowerCamelCase	)
  							return heap_pre_order
  
  
  
  					def      __a    (		self    :    Dict					,			lowerCamelCase    :    Tuple					,			lowerCamelCase    :    Dict					,			lowerCamelCase    :    List[str]=0	):
  							if curr_node:
  									preorder.append((curr_node.val, level)	)
  									self.__traversal(curr_node.left					,			lowerCamelCase					,			level + 1	)
  									self.__traversal(curr_node.right					,			lowerCamelCase					,			level + 1	)
  							else:
  									preorder.append(('#', level)	)
  
  
  
  
  
  					def __str__(		self    :    List[Any]	):
  							if self.is_empty():
  									return ""
  							lowerCamelCase_  :    str   	=			self.pre_order()
  
  							return "\n".join(('-' * level + str(lowerCamelCase	)) for value, level in preorder_heap	)
  
  
  # Unit Tests
  if __name__ == "__main__":
  		import doctest
  
  		doctest.testmod()
  