






"""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     lowerCamelCase_					(	__A	:		Optional[int]      ,						**__A	:		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 							__lowercase(							_UpperCamelCase	):






						@staticmethod
						def 		lowerCAmelCase (       _SCREAMING_SNAKE_CASE)      ->	Optional[Any]:




											"""simple docstring"""





											lowerCamelCase__					=parser.add_parser('run_beam'     ,    help='Run a Beam dataset processing pipeline')
											run_beam_parser.add_argument('dataset'     ,    type=_SCREAMING_SNAKE_CASE     ,    help='Name of the dataset to download')
											run_beam_parser.add_argument('--name'     ,    type=_SCREAMING_SNAKE_CASE     ,    default=_SCREAMING_SNAKE_CASE     ,    help='Dataset config name')
											run_beam_parser.add_argument(
											    '--cache_dir'     ,    type=_SCREAMING_SNAKE_CASE     ,    default=_SCREAMING_SNAKE_CASE     ,    help='Cache directory where the datasets are stored'     ,    )
											run_beam_parser.add_argument(
											    '--beam_pipeline_options'     ,    type=_SCREAMING_SNAKE_CASE     ,    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=_SCREAMING_SNAKE_CASE     ,    default=_SCREAMING_SNAKE_CASE     ,    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=_SCREAMING_SNAKE_CASE)






						def __init__(       self     ,    _SCREAMING_SNAKE_CASE     ,    _SCREAMING_SNAKE_CASE     ,    _SCREAMING_SNAKE_CASE     ,    _SCREAMING_SNAKE_CASE     ,    _SCREAMING_SNAKE_CASE     ,    _SCREAMING_SNAKE_CASE     ,    _SCREAMING_SNAKE_CASE     ,    _SCREAMING_SNAKE_CASE     ,    _SCREAMING_SNAKE_CASE     ,    **_SCREAMING_SNAKE_CASE     ,    )      ->	Any:




											"""simple docstring"""





											lowerCamelCase__					=dataset
											lowerCamelCase__					=name
											lowerCamelCase__					=cache_dir
											lowerCamelCase__					=beam_pipeline_options
											lowerCamelCase__					=data_dir
											lowerCamelCase__					=all_configs
											lowerCamelCase__					=save_infos
											lowerCamelCase__					=ignore_verifications
											lowerCamelCase__					=force_redownload
											lowerCamelCase__					=config_kwargs






						def 		lowerCAmelCase (       self)      ->	Tuple:




											"""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)
											lowerCamelCase__					,  lowerCamelCase__					=self._dataset, self._name
											lowerCamelCase__					=dataset_module_factory(_SCREAMING_SNAKE_CASE)
											lowerCamelCase__					=import_main_class(dataset_module.module_path)
											lowerCamelCase__					=[]
											if self._beam_pipeline_options:
																lowerCamelCase__					=beam.options.pipeline_options.PipelineOptions(
																    flags=[F"--{opt.strip()}" for opt in self._beam_pipeline_options.split(',') if opt])
											else:
																lowerCamelCase__					=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=_SCREAMING_SNAKE_CASE     ,    cache_dir=self._cache_dir     ,    base_path=dataset_module.builder_kwargs.get('base_path')     ,    ))
											else:
																builders.append(
																    builder_cls(
																        config_name=_SCREAMING_SNAKE_CASE     ,    data_dir=self._data_dir     ,    beam_options=_SCREAMING_SNAKE_CASE     ,    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=_SCREAMING_SNAKE_CASE     ,    )
																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__					=os.path.join(builder_cls.get_imported_module_dir()     ,    config.DATASETDICT_INFOS_FILENAME)

																lowerCamelCase__					=Path(_SCREAMING_SNAKE_CASE).name + '.py'

																lowerCamelCase__					=os.path.join(_SCREAMING_SNAKE_CASE     ,    _SCREAMING_SNAKE_CASE)
																if os.path.isfile(_SCREAMING_SNAKE_CASE):
																					lowerCamelCase__					=os.path.dirname(_SCREAMING_SNAKE_CASE)
																elif os.path.isfile(_SCREAMING_SNAKE_CASE):
																					lowerCamelCase__					=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__					=os.path.join(_SCREAMING_SNAKE_CASE     ,    config.DATASETDICT_INFOS_FILENAME)
																copyfile(_SCREAMING_SNAKE_CASE     ,    _SCREAMING_SNAKE_CASE)
																print(F"Dataset Infos file saved at {user_dataset_infos_path}")







"""simple docstring"""



import timeit
from typing import Callable, Optional

from ..configuration_utils import PretrainedConfig
from ..models.auto.modeling_auto import MODEL_MAPPING, MODEL_WITH_LM_HEAD_MAPPING
from ..utils import is_pyanvml_available, is_torch_available, logging
from .benchmark_utils import (
    Benchmark,
    Memory,
    MemorySummary,
    measure_peak_memory_cpu,
    start_memory_tracing,
    stop_memory_tracing,
)


if is_torch_available():
						import torch

						from .benchmark_args import PyTorchBenchmarkArguments


if is_pyanvml_available():
						import pyanvml.pyanvml as nvml


_snake_case							:   str	  =       logging.get_logger(__name__)




class 							__lowercase(							_UpperCamelCase	):
						lowerCAmelCase_		:				PyTorchBenchmarkArguments
						lowerCAmelCase_		:				PretrainedConfig
						lowerCAmelCase_		:				str=							"PyTorch"






						@property
						def 		lowerCAmelCase (       self)      ->	List[Any]:




											"""simple docstring"""





											return torch.__version__






						def 		lowerCAmelCase (       self     ,    _SCREAMING_SNAKE_CASE     ,    _SCREAMING_SNAKE_CASE     ,    _SCREAMING_SNAKE_CASE)      ->	float:




											"""simple docstring"""





											lowerCamelCase__					=self._prepare_inference_func(_SCREAMING_SNAKE_CASE     ,    _SCREAMING_SNAKE_CASE     ,    _SCREAMING_SNAKE_CASE)
											return self._measure_speed(_inference)






						def 		lowerCAmelCase (       self     ,    _SCREAMING_SNAKE_CASE     ,    _SCREAMING_SNAKE_CASE     ,    _SCREAMING_SNAKE_CASE)      ->	[Memory, Optional[MemorySummary]]:




											"""simple docstring"""





											lowerCamelCase__					=self._prepare_inference_func(_SCREAMING_SNAKE_CASE     ,    _SCREAMING_SNAKE_CASE     ,    _SCREAMING_SNAKE_CASE)
											return self._measure_memory(_inference)






						def 		lowerCAmelCase (       self     ,    _SCREAMING_SNAKE_CASE     ,    _SCREAMING_SNAKE_CASE     ,    _SCREAMING_SNAKE_CASE)      ->	float:




											"""simple docstring"""





											lowerCamelCase__					=self._prepare_train_func(_SCREAMING_SNAKE_CASE     ,    _SCREAMING_SNAKE_CASE     ,    _SCREAMING_SNAKE_CASE)
											return self._measure_speed(_train)






						def 		lowerCAmelCase (       self     ,    _SCREAMING_SNAKE_CASE     ,    _SCREAMING_SNAKE_CASE     ,    _SCREAMING_SNAKE_CASE)      ->	[Memory, Optional[MemorySummary]]:




											"""simple docstring"""





											lowerCamelCase__					=self._prepare_train_func(_SCREAMING_SNAKE_CASE     ,    _SCREAMING_SNAKE_CASE     ,    _SCREAMING_SNAKE_CASE)
											return self._measure_memory(_train)






						def 		lowerCAmelCase (       self     ,    _SCREAMING_SNAKE_CASE     ,    _SCREAMING_SNAKE_CASE     ,    _SCREAMING_SNAKE_CASE)      ->	Callable[[], None]:




											"""simple docstring"""





											lowerCamelCase__					=self.config_dict[model_name]

											if self.args.torchscript:
																lowerCamelCase__					=True

											lowerCamelCase__					=(
											    hasattr(_SCREAMING_SNAKE_CASE     ,    'architectures')
											    and isinstance(config.architectures     ,    _SCREAMING_SNAKE_CASE)
											    and len(config.architectures) > 0
											)
											if not self.args.only_pretrain_model and has_model_class_in_config:
																try:
																					lowerCamelCase__					=config.architectures[0]
																					lowerCamelCase__					=__import__('transformers'     ,    fromlist=[model_class])
																					lowerCamelCase__					=getattr(_SCREAMING_SNAKE_CASE     ,    _SCREAMING_SNAKE_CASE)
																					lowerCamelCase__					=model_cls(_SCREAMING_SNAKE_CASE)
																except ImportError:
																					raise ImportError(
																					    F"{model_class} does not exist. If you just want to test the pretrained model, you might want to"
																					    ' set `--only_pretrain_model` or `args.only_pretrain_model=True`.')
											else:
																lowerCamelCase__					=MODEL_MAPPING[config.__class__](_SCREAMING_SNAKE_CASE)

											model.eval()
											model.to(self.args.device)

											# encoder-decoder has vocab size saved differently
											lowerCamelCase__					=config.vocab_size if hasattr(_SCREAMING_SNAKE_CASE     ,    'vocab_size') else config.encoder.vocab_size
											lowerCamelCase__					=torch.randint(_SCREAMING_SNAKE_CASE     ,    (batch_size, sequence_length)     ,    dtype=torch.long     ,    device=self.args.device)

											if self.args.fpaa:
																logger.info('Running training in Mixed Precision...')
																if not self.args.is_gpu:
																					raise ValueError('Mixed precision is possible only for GPU.')
																# amp seems to have memory leaks so that memory usage
																# is measured using .half() for now https://github.com/NVIDIA/apex/issues/439
																model.half()

											if self.args.torchscript:
																with torch.no_grad():
																					lowerCamelCase__					=torch.jit.trace(_SCREAMING_SNAKE_CASE     ,    _SCREAMING_SNAKE_CASE)
											else:
																lowerCamelCase__					=model

											def encoder_decoder_forward():
																with torch.no_grad():
																					lowerCamelCase__					=inference_model(_SCREAMING_SNAKE_CASE     ,    decoder_input_ids=_SCREAMING_SNAKE_CASE)
																return outputs

											def encoder_forward():
																with torch.no_grad():
																					lowerCamelCase__					=inference_model(_SCREAMING_SNAKE_CASE)
																return outputs

											lowerCamelCase__					=encoder_decoder_forward if config.is_encoder_decoder else encoder_forward
											return _forward






						def 		lowerCAmelCase (       self     ,    _SCREAMING_SNAKE_CASE     ,    _SCREAMING_SNAKE_CASE     ,    _SCREAMING_SNAKE_CASE)      ->	Callable[[], None]:




											"""simple docstring"""





											lowerCamelCase__					=self.config_dict[model_name]

											lowerCamelCase__					=(
											    hasattr(_SCREAMING_SNAKE_CASE     ,    'architectures')
											    and isinstance(config.architectures     ,    _SCREAMING_SNAKE_CASE)
											    and len(config.architectures) > 0
											)
											if not self.args.only_pretrain_model and has_model_class_in_config:
																try:
																					lowerCamelCase__					=config.architectures[0]
																					lowerCamelCase__					=__import__('transformers'     ,    fromlist=[model_class])
																					lowerCamelCase__					=getattr(_SCREAMING_SNAKE_CASE     ,    _SCREAMING_SNAKE_CASE)
																					lowerCamelCase__					=model_cls(_SCREAMING_SNAKE_CASE)
																except ImportError:
																					raise ImportError(
																					    F"{model_class} does not exist. If you just want to test the pretrained model, you might want to"
																					    ' set `--only_pretrain_model` or `args.only_pretrain_model=True`.')
											else:
																lowerCamelCase__					=MODEL_WITH_LM_HEAD_MAPPING[config.__class__](_SCREAMING_SNAKE_CASE)

											if self.args.torchscript:
																raise NotImplementedError('Training for torchscript is currently not implemented')
											else:
																lowerCamelCase__					=model

											model.train()
											model.to(self.args.device)

											# encoder-decoder has vocab size saved differently
											lowerCamelCase__					=config.vocab_size if hasattr(_SCREAMING_SNAKE_CASE     ,    'vocab_size') else config.encoder.vocab_size
											lowerCamelCase__					=torch.randint(_SCREAMING_SNAKE_CASE     ,    (batch_size, sequence_length)     ,    dtype=torch.long     ,    device=self.args.device)

											if self.args.fpaa:
																logger.info('Running training in Mixed Precision...')
																if not self.args.is_gpu:
																					raise ValueError('Mixed precision is possible only for GPU.')

																# amp seems to have memory leaks so that memory usage
																# is measured using .half() for now https://github.com/NVIDIA/apex/issues/439
																model.half()

											def compute_loss_and_backprob_encoder():
																lowerCamelCase__					=train_model(_SCREAMING_SNAKE_CASE     ,    labels=_SCREAMING_SNAKE_CASE)[0]
																loss.backward()
																return loss

											def compute_loss_and_backprob_encoder_decoder():
																lowerCamelCase__					=train_model(_SCREAMING_SNAKE_CASE     ,    decoder_input_ids=_SCREAMING_SNAKE_CASE     ,    labels=_SCREAMING_SNAKE_CASE)[0]
																loss.backward()
																return loss

											lowerCamelCase__					=(
											    compute_loss_and_backprob_encoder_decoder
											    if config.is_encoder_decoder
											    else compute_loss_and_backprob_encoder
											)
											return _train






						def 		lowerCAmelCase (       self     ,    _SCREAMING_SNAKE_CASE)      ->	float:




											"""simple docstring"""





											try:
																if self.args.is_tpu or self.args.torchscript:
																					# run additional 10 times to stabilize compilation for tpu and torchscript
																					logger.info('Do inference on TPU or torchscript. Running model 5 times to stabilize compilation')
																					timeit.repeat(
																					    _SCREAMING_SNAKE_CASE     ,    repeat=1     ,    number=5     ,    )

																# as written in https://docs.python.org/2/library/timeit.html#timeit.Timer.repeat, min should be taken rather than the average
																lowerCamelCase__					=timeit.repeat(
																    _SCREAMING_SNAKE_CASE     ,    repeat=self.args.repeat     ,    number=10     ,    )

																if self.args.is_tpu and self.args.torch_xla_tpu_print_metrics:
																					import torch_xla.debug.metrics as met

																					self.print_fn(met.metrics_report())

																return min(_SCREAMING_SNAKE_CASE) / 10.0
											except RuntimeError as e:
																self.print_fn(F"Doesn't fit on GPU. {e}")
																return "N/A"






						def 		lowerCAmelCase (       self     ,    _SCREAMING_SNAKE_CASE)      ->	[Memory, MemorySummary]:




											"""simple docstring"""





											try:
																if self.args.trace_memory_line_by_line:
																					lowerCamelCase__					=start_memory_tracing('transformers')

																if self.args.is_tpu:
																					# tpu
																					raise NotImplementedError(
																					    'Memory Benchmarking is currently not implemented for TPU. Please disable memory benchmarking with'
																					    ' `--no-memory` or `args.memory=False`')
																elif self.args.is_gpu:
																					if not is_pyanvml_available():
																										logger.warning(
																										    'py3nvml not installed, we won\'t log GPU memory usage. '
																										    'Install py3nvml (pip install py3nvml) to log information about GPU.')
																										lowerCamelCase__					='N/A'
																					else:
																										logger.info(
																										    'Measuring total GPU usage on GPU device. Make sure to not have additional processes running'
																										    ' on the same GPU.')
																										# init nvml
																										nvml.nvmlInit()
																										func()
																										lowerCamelCase__					=nvml.nvmlDeviceGetHandleByIndex(self.args.device_idx)
																										lowerCamelCase__					=nvml.nvmlDeviceGetMemoryInfo(_SCREAMING_SNAKE_CASE)
																										lowerCamelCase__					=meminfo.used
																										lowerCamelCase__					=Memory(_SCREAMING_SNAKE_CASE)
																										# shutdown nvml
																										nvml.nvmlShutdown()
																else:
																					# cpu
																					lowerCamelCase__					=measure_peak_memory_cpu(_SCREAMING_SNAKE_CASE)
																					lowerCamelCase__					=Memory(_SCREAMING_SNAKE_CASE) if isinstance(_SCREAMING_SNAKE_CASE     ,    _SCREAMING_SNAKE_CASE) else memory_bytes

																if self.args.trace_memory_line_by_line:
																					lowerCamelCase__					=stop_memory_tracing(_SCREAMING_SNAKE_CASE)
																else:
																					lowerCamelCase__					=None

																return memory, summary
											except RuntimeError as e:
																self.print_fn(F"Doesn't fit on GPU. {e}")
																return "N/A", None