code: 
  
  
  
  
  
  '''simple docstring'''
  
  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 		UpperCamelCase       (							_lowerCamelCase      :      Optional[int]		, **_lowerCamelCase      :      Optional[int] ):
      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			( lowerCAmelCase_       ):
   @staticmethod
   def   UpperCAmelCase_							(       lowercase_   :Any							)->							List[Any]:
       A__       =    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   :List[Any]		, lowercase_   :Dict		, lowercase_   :Union[str, Any]		, lowercase_   :List[str]		, lowercase_   :List[Any]		, lowercase_   :Union[str, Any]		, lowercase_   :Dict		, lowercase_   :str		, lowercase_   :Any		, lowercase_   :Optional[Any]		, **lowercase_   :str		, )->							Optional[Any]:
       A__       =    dataset
       A__       =    name
       A__       =    cache_dir
       A__       =    beam_pipeline_options
       A__       =    data_dir
       A__       =    all_configs
       A__       =    save_infos
       A__       =    ignore_verifications
       A__       =    force_redownload
       A__       =    config_kwargs
  
  
  
   def   UpperCAmelCase_							(       self   :Dict							)->							List[str]:
       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							)
       A__,			A__       =    self._dataset, self._name
       A__       =    dataset_module_factory(lowerCamelCase_							)
       A__       =    import_main_class(dataset_module.module_path							)
       A__       =    []
       if self._beam_pipeline_options:
           A__       =    beam.options.pipeline_options.PipelineOptions(
               flags=[F"--{opt.strip()}" for opt in self._beam_pipeline_options.split(","							) if opt]							)
       else:
           A__       =    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:
           A__       =    os.path.join(builder_cls.get_imported_module_dir()		, config.DATASETDICT_INFOS_FILENAME							)
  
           A__       =    Path(lowerCamelCase_							).name + ".py"
  
           A__       =    os.path.join(lowerCamelCase_		, lowerCamelCase_							)
           if os.path.isfile(lowerCamelCase_							):
               A__       =    os.path.dirname(lowerCamelCase_							)
           elif os.path.isfile(lowerCamelCase_							):
               A__       =    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
           A__       =    os.path.join(lowerCamelCase_		, config.DATASETDICT_INFOS_FILENAME							)
           copyfile(lowerCamelCase_		, lowerCamelCase_							)
           print(F"Dataset Infos file saved at {user_dataset_infos_path}"							)
  
  
style_context: 
  
  
  
  
  
  '''simple docstring'''
  
  import bisect
  import copy
  from collections import defaultdict
  
  import numpy as np
  from torch.utils.data import BatchSampler, Sampler
  
  from utils import logger
  
  
  
  
  
  def 		UpperCamelCase       (							_lowerCamelCase      :      str		, _lowerCamelCase      :      Optional[Any] ):
      A__       =    copy.deepcopy(_lowerCamelCase )
      A__       =    sorted(_lowerCamelCase )
      A__       =    [bisect.bisect_right(_lowerCamelCase		, _lowerCamelCase ) for y in x]
      return quantized
  
  
  
  
  
  def 		UpperCamelCase       (							_lowerCamelCase      :      Optional[int]		, _lowerCamelCase      :      Tuple=0 ):
      A__       =    np.arange(start=3		, stop=_lowerCamelCase		, step=4 ).tolist() if k > 0 else [10]
      A__       =    _quantize(_lowerCamelCase		, _lowerCamelCase )
      # count number of elements per group
      A__       =    np.unique(_lowerCamelCase		, return_counts=_lowerCamelCase )[1]
      A__       =    [0] + bins + [np.inf]
      logger.info("Using {} as bins for aspect lengths quantization".format(_lowerCamelCase ) )
      logger.info("Count of instances per bin: {}".format(_lowerCamelCase ) )
      return groups
  
  
  
  
  
  
  class      UpperCAmelCase			( UpperCamelCase__       ):
   def __init__(       self   :Optional[Any]		, lowercase_   :List[Any]		, lowercase_   :Optional[int]		, lowercase_   :str							)->							Dict:
       if not isinstance(lowercase_		, lowercase_							):
           raise ValueError(
               "sampler should be an instance of torch.utils.data.Sampler, but got sampler={}".format(lowercase_							)							)
       A__       =    sampler
       A__       =    group_ids
       A__       =    batch_size
   def __iter__(       self   :int							)->							Tuple:
       A__       =    defaultdict(lowercase_							)
       A__       =    defaultdict(lowercase_							)
  
       A__       =    0
       for idx in self.sampler:
           A__       =    self.group_ids[idx]
           buffer_per_group[group_id].append(lowercase_							)
           samples_per_group[group_id].append(lowercase_							)
           if len(buffer_per_group[group_id]							) == self.batch_size:
               yield buffer_per_group[group_id]  # TODO
               num_batches += 1
               del buffer_per_group[group_id]
           assert len(buffer_per_group[group_id]							) < self.batch_size
  
       # now we have run out of elements that satisfy
       # the group criteria, let's return the remaining
       # elements so that the size of the sampler is
       # deterministic
       A__       =    len(self							)
       A__       =    expected_num_batches - num_batches
       if num_remaining > 0:
           # for the remaining batches, group the batches by similar lengths
           A__       =    []
           for group_id, idxs in sorted(buffer_per_group.items()		, key=lambda lowercase_							: x[0]							):
               batch_idx.extend(lowercase_							)
               if len(lowercase_							) >= self.batch_size:
                   yield batch_idx[: self.batch_size]
                   A__       =    batch_idx[self.batch_size :]
                   num_remaining -= 1
           if len(lowercase_							) > 0:
               yield batch_idx
               num_remaining -= 1
       assert num_remaining == 0
  
  
  
   def __len__(       self   :int							)->							Tuple:
       return (len(self.sampler							) + self.batch_size - 1) // self.batch_size
  
  
  
  