


"""simple docstring"""







import sys
from pathlib import Path


lowerCAmelCase__				    =       Path(__file__).resolve().parents[3] / 'src'
sys.path.insert(1, str(git_repo_path))

import dataclasses  # noqa
import io  # noqa
import itertools  # noqa
import json  # noqa
import os  # noqa
import unittest  # noqa
from copy import deepcopy  # noqa

from parameterized import parameterized  # noqa
from transformers import TrainingArguments, is_torch_available  # noqa
from transformers.deepspeed import is_deepspeed_available  # noqa
from transformers.file_utils import WEIGHTS_NAME  # noqa
from transformers.testing_utils import (  # noqa
    CaptureLogger,
    ExtendSysPath,
    TestCasePlus,
    execute_subprocess_async,
    get_gpu_count,
    mockenv_context,
    require_deepspeed,
    require_torch_gpu,
    require_torch_multi_gpu,
    slow,
)
from transformers.trainer_utils import set_seed  # noqa


set_seed(42)

lowerCAmelCase__				    =       {'base': 'patrickvonplaten/wav2vec2_tiny_random', 'robust': 'patrickvonplaten/wav2vec2_tiny_random_robust'}

lowerCAmelCase__				    =       'zero2'
lowerCAmelCase__				    =       'zero3'
lowerCAmelCase__				    =       [ZEROa, ZEROa]
def  a__      (    SCREAMING_SNAKE_CASE   :   int    ,				SCREAMING_SNAKE_CASE   :   Optional[int]    ,				SCREAMING_SNAKE_CASE   :   Union[str, Any]    ):


  '''simple docstring'''







  lowerCAmelCase					:					Union[str, Any]								=							parameterized.to_safe_name("_".join(str(UpperCAmelCase_    ) for x in param.args    )    )
  return f"""{func.__name__}_{param_based_name}"""


# Cartesian-product of zero stages with models to test
lowerCAmelCase__				    =       list(itertools.product(stages, models.keys()))





@slow
@require_deepspeed
@require_torch_gpu
class    SCREAMING_SNAKE_CASE__    (		_lowerCamelCase    ):
       """simple docstring"""


       @parameterized.expand(snake_case__		,					name_func=snake_case__						)
       def 	lowercase__							(						self		,					snake_case__		,					snake_case__						):






         """simple docstring"""





         self.run_and_check(
             stage=snake_case__		,					model=snake_case__		,					distributed=snake_case__		,					fpaa=snake_case__		,					)

       @require_torch_multi_gpu
       @parameterized.expand(snake_case__		,					name_func=snake_case__						)
       def 	lowercase__							(						self		,					snake_case__		,					snake_case__						):






         """simple docstring"""





         self.run_and_check(
             stage=snake_case__		,					model=snake_case__		,					distributed=snake_case__		,					fpaa=snake_case__		,					)

       @parameterized.expand(snake_case__		,					name_func=snake_case__						)
       def 	lowercase__							(						self		,					snake_case__		,					snake_case__						):






         """simple docstring"""





         self.run_and_check(
             stage=snake_case__		,					model=snake_case__		,					distributed=snake_case__		,					fpaa=snake_case__		,					)

       @require_torch_multi_gpu
       @parameterized.expand(snake_case__		,					name_func=snake_case__						)
       def 	lowercase__							(						self		,					snake_case__		,					snake_case__						):






         """simple docstring"""





         self.run_and_check(
             stage=snake_case__		,					model=snake_case__		,					distributed=snake_case__		,					fpaa=snake_case__		,					)

       def 	lowercase__							(						self		,					snake_case__						):






         """simple docstring"""





         pass

       def 	lowercase__							(						self		,					snake_case__		,					snake_case__		,					snake_case__ = 10		,					snake_case__ = True		,					snake_case__ = True		,					snake_case__ = True		,					):






         """simple docstring"""





         lowerCAmelCase					:					Union[str, Any]								=							models[model]

         lowerCAmelCase					:					List[str]								=							self.run_trainer(
             stage=snake_case__		,					model_name=snake_case__		,					eval_steps=snake_case__		,					num_train_epochs=1		,					distributed=snake_case__		,					fpaa=snake_case__		,					)

         self.do_checks(snake_case__						)

         return output_dir

       def 	lowercase__							(						self		,					snake_case__		,					snake_case__		,					snake_case__ = 10		,					snake_case__ = 1		,					snake_case__ = True		,					snake_case__ = True		,					):






         """simple docstring"""





         lowerCAmelCase					:					Optional[int]								=							self.get_auto_remove_tmp_dir("./xxx"		,					after=snake_case__						)
         lowerCAmelCase					:					Union[str, Any]								=							f"""
            --model_name_or_path {model_name}
            --dataset_name hf-internal-testing/librispeech_asr_dummy
            --dataset_config_name clean
            --train_split_name validation
            --validation_split_name validation
            --output_dir {output_dir}
            --num_train_epochs {str(snake_case__						)}
            --per_device_train_batch_size 2
            --per_device_eval_batch_size 2
            --evaluation_strategy steps
            --learning_rate 5e-4
            --warmup_steps 8
            --orthography timit
            --preprocessing_num_workers 1
            --group_by_length
            --freeze_feature_extractor
            --report_to none
            --save_steps 0
            --eval_steps {eval_steps}
            --report_to none
        """.split()

         if fpaa:
           args.extend(["--fp16"]						)

         # currently ds_config_wav2vec2_zero.json requires "zero_optimization.find_unused_parameters": true,
         # hence the separate config files
         lowerCAmelCase					:					Any								=							f"""--deepspeed {self.test_file_dir_str}/ds_config_wav2vec2_{stage}.json""".split()
         lowerCAmelCase					:					Union[str, Any]								=							[f"""{self.examples_dir_str}/research_projects/wav2vec2/run_asr.py"""]
         lowerCAmelCase					:					Optional[Any]								=							self.get_launcher(snake_case__						)

         lowerCAmelCase					:					Optional[Any]								=							launcher + script + args + ds_args
         # keep for quick debug
         # print(" ".join([f"\nPYTHONPATH={self.src_dir_str}"] +cmd)); die
         execute_subprocess_async(snake_case__		,					env=self.get_env()						)

         return output_dir







       def 	lowercase__							(						self		,					snake_case__=False						):






         """simple docstring"""





         lowerCAmelCase					:					Any								=							min(2		,					get_gpu_count()						) if distributed else 1
         return f"""deepspeed --num_nodes 1 --num_gpus {num_gpus}""".split()









"""simple docstring"""







import multiprocessing
from typing import TYPE_CHECKING, Optional, Union

from .. import Dataset, Features, config
from ..formatting import query_table
from ..packaged_modules.sql.sql import Sql
from ..utils import logging
from .abc import AbstractDatasetInputStream


if TYPE_CHECKING:
     import sqlitea

     import sqlalchemy


class 				SCREAMING_SNAKE_CASE__    (		lowercase    ):
       """simple docstring"""


       def __init__(						self		,					snake_case__		,					snake_case__		,					snake_case__ = None		,					snake_case__ = None		,					snake_case__ = False		,					**snake_case__		,					):






         """simple docstring"""





         super().__init__(features=snake_case__		,					cache_dir=snake_case__		,					keep_in_memory=snake_case__		,					**snake_case__						)
         lowerCAmelCase					:					Dict								=							Sql(
             cache_dir=snake_case__		,					features=snake_case__		,					sql=snake_case__		,					con=snake_case__		,					**snake_case__		,					)







       def 	lowercase__							(						self						):






         """simple docstring"""





         lowerCAmelCase					:					Tuple								=							None
         lowerCAmelCase					:					Optional[int]								=							None
         lowerCAmelCase					:					str								=							None
         lowerCAmelCase					:					int								=							None

         self.builder.download_and_prepare(
             download_config=snake_case__		,					download_mode=snake_case__		,					verification_mode=snake_case__		,					base_path=snake_case__		,					)

         # Build dataset for splits
         lowerCAmelCase					:					str								=							self.builder.as_dataset(
             split="train"		,					verification_mode=snake_case__		,					in_memory=self.keep_in_memory						)
         return dataset


class 				SCREAMING_SNAKE_CASE__    :
       """simple docstring"""


       def __init__(						self		,					snake_case__		,					snake_case__		,					snake_case__		,					snake_case__ = None		,					snake_case__ = None		,					**snake_case__		,					):






         """simple docstring"""





         if num_proc is not None and num_proc <= 0:
           raise ValueError(f"""num_proc {num_proc} must be an integer > 0."""						)

         lowerCAmelCase					:					Tuple								=							dataset
         lowerCAmelCase					:					Tuple								=							name
         lowerCAmelCase					:					List[str]								=							con
         lowerCAmelCase					:					List[Any]								=							batch_size if batch_size else config.DEFAULT_MAX_BATCH_SIZE
         lowerCAmelCase					:					Optional[int]								=							num_proc
         lowerCAmelCase					:					Optional[int]								=							to_sql_kwargs

       def 	lowercase__							(						self						):






         """simple docstring"""





         lowerCAmelCase					:					List[str]								=							self.to_sql_kwargs.pop("sql"		,					snake_case__						)
         lowerCAmelCase					:					List[Any]								=							self.to_sql_kwargs.pop("con"		,					snake_case__						)
         lowerCAmelCase					:					Dict								=							self.to_sql_kwargs.pop("index"		,					snake_case__						)

         lowerCAmelCase					:					Dict								=							self._write(index=snake_case__		,					**self.to_sql_kwargs						)
         return written

       def 	lowercase__							(						self		,					snake_case__						):






         """simple docstring"""





         lowerCAmelCase						,  lowerCAmelCase						,  lowerCAmelCase					:					Optional[int]								=							args
         lowerCAmelCase					:					Optional[int]								=							{**to_sql_kwargs, "if_exists": "append"} if offset > 0 else to_sql_kwargs
         lowerCAmelCase					:					List[Any]								=							query_table(
             table=self.dataset.data		,					key=slice(snake_case__		,					offset + self.batch_size						)		,					indices=self.dataset._indices		,					)
         lowerCAmelCase					:					Tuple								=							batch.to_pandas()
         lowerCAmelCase					:					str								=							df.to_sql(self.name		,					self.con		,					index=snake_case__		,					**snake_case__						)
         return num_rows or len(snake_case__						)







       def 	lowercase__							(						self		,					snake_case__		,					**snake_case__						):






         """simple docstring"""





         lowerCAmelCase					:					Tuple								=							0

         if self.num_proc is None or self.num_proc == 1:
           for offset in logging.tqdm(
               range(0		,					len(self.dataset						)		,					self.batch_size						)		,					unit="ba"		,					disable=not logging.is_progress_bar_enabled()		,					desc="Creating SQL from Arrow format"		,					):
             written += self._batch_sql((offset, index, to_sql_kwargs)						)
         else:
           lowerCAmelCase						,  lowerCAmelCase					:					Dict								=							len(self.dataset						), self.batch_size
           with multiprocessing.Pool(self.num_proc						) as pool:
             for num_rows in logging.tqdm(
                 pool.imap(
                     self._batch_sql		,					[(offset, index, to_sql_kwargs) for offset in range(0		,					snake_case__		,					snake_case__						)]		,					)		,					total=(num_rows // batch_size) + 1 if num_rows % batch_size else num_rows // batch_size		,					unit="ba"		,					disable=not logging.is_progress_bar_enabled()		,					desc="Creating SQL from Arrow format"		,					):
               written += num_rows

         return written





