
"""simple docstring"""





import inspect
import unittest

from transformers import is_torch_available, is_vision_available
from transformers.models.auto import get_values
from transformers.testing_utils import require_torch, require_vision, slow, torch_device

from ...test_configuration_common import ConfigTester
from ...test_modeling_common import ModelTesterMixin, floats_tensor, ids_tensor
from ...test_pipeline_mixin import PipelineTesterMixin


if is_torch_available():
    import torch

    from transformers import MODEL_MAPPING, GLPNConfig, GLPNForDepthEstimation, GLPNModel
    from transformers.models.glpn.modeling_glpn import GLPN_PRETRAINED_MODEL_ARCHIVE_LIST


if is_vision_available():
    from PIL import Image

    from transformers import GLPNImageProcessor


class A_      (lowercase__ ):




  '''simple docstring'''







  def 			UpperCamelCase__   (						self    ):

     """simple docstring"""
     UpperCAmelCase_   :   Any     	=       self.config_class(**self.inputs_dict    )
     self.parent.assertTrue(hasattr(lowercase_							,  "hidden_sizes"    )    )
     self.parent.assertTrue(hasattr(lowercase_							,  "num_attention_heads"    )    )
     self.parent.assertTrue(hasattr(lowercase_							,  "num_encoder_blocks"    )    )




class A_      :




  '''simple docstring'''







  def __init__(						self							,  lowercase_							,  lowercase_=13							,  lowercase_=64							,  lowercase_=3							,  lowercase_=4							,  lowercase_=[2, 2, 2, 2]							,  lowercase_=[8, 4, 2, 1]							,  lowercase_=[16, 32, 64, 128]							,  lowercase_=[1, 4, 8, 16]							,  lowercase_=[1, 2, 4, 8]							,  lowercase_=True							,  lowercase_=True							,  lowercase_="gelu"							,  lowercase_=0.1							,  lowercase_=0.1							,  lowercase_=0.02							,  lowercase_=16							,  lowercase_=3							,  lowercase_=None							,  ):

     """simple docstring"""
     UpperCAmelCase_   :   str     	=       parent
     UpperCAmelCase_   :   Dict     	=       batch_size
     UpperCAmelCase_   :   Tuple     	=       image_size
     UpperCAmelCase_   :   Any     	=       num_channels
     UpperCAmelCase_   :   Tuple     	=       num_encoder_blocks
     UpperCAmelCase_   :   Union[str, Any]     	=       sr_ratios
     UpperCAmelCase_   :   int     	=       depths
     UpperCAmelCase_   :   Any     	=       hidden_sizes
     UpperCAmelCase_   :   Optional[int]     	=       downsampling_rates
     UpperCAmelCase_   :   str     	=       num_attention_heads
     UpperCAmelCase_   :   List[Any]     	=       is_training
     UpperCAmelCase_   :   Dict     	=       use_labels
     UpperCAmelCase_   :   Optional[Any]     	=       hidden_act
     UpperCAmelCase_   :   Any     	=       hidden_dropout_prob
     UpperCAmelCase_   :   List[str]     	=       attention_probs_dropout_prob
     UpperCAmelCase_   :   Dict     	=       initializer_range
     UpperCAmelCase_   :   List[str]     	=       decoder_hidden_size
     UpperCAmelCase_   :   Optional[Any]     	=       num_labels
     UpperCAmelCase_   :   List[str]     	=       scope







  def 			UpperCamelCase__   (						self    ):

     """simple docstring"""
     UpperCAmelCase_   :   int     	=       floats_tensor([self.batch_size, self.num_channels, self.image_size, self.image_size]    )

     UpperCAmelCase_   :   int     	=       None
     if self.use_labels:
        UpperCAmelCase_   :   Optional[Any]     	=       ids_tensor([self.batch_size, self.image_size, self.image_size]							,  self.num_labels    )

     UpperCAmelCase_   :   Optional[Any]     	=       self.get_config()
     return config, pixel_values, labels







  def 			UpperCamelCase__   (						self    ):

     """simple docstring"""
     return GLPNConfig(
         image_size=self.image_size							,  num_channels=self.num_channels							,  num_encoder_blocks=self.num_encoder_blocks							,  depths=self.depths							,  hidden_sizes=self.hidden_sizes							,  num_attention_heads=self.num_attention_heads							,  hidden_act=self.hidden_act							,  hidden_dropout_prob=self.hidden_dropout_prob							,  attention_probs_dropout_prob=self.attention_probs_dropout_prob							,  initializer_range=self.initializer_range							,  decoder_hidden_size=self.decoder_hidden_size							,  )







  def 			UpperCamelCase__   (						self							,  lowercase_							,  lowercase_							,  lowercase_    ):

     """simple docstring"""
     UpperCAmelCase_   :   Optional[int]     	=       GLPNModel(config=lowercase_    )
     model.to(lowercase_    )
     model.eval()
     UpperCAmelCase_   :   Dict     	=       model(lowercase_    )
     UpperCAmelCase_   :   Optional[int]     	=       self.image_size // (self.downsampling_rates[-1] * 2)
     self.parent.assertEqual(
         result.last_hidden_state.shape							,  (self.batch_size, self.hidden_sizes[-1], expected_height, expected_width)    )







  def 			UpperCamelCase__   (						self							,  lowercase_							,  lowercase_							,  lowercase_    ):

     """simple docstring"""
     UpperCAmelCase_   :   Dict     	=       self.num_labels
     UpperCAmelCase_   :   Union[str, Any]     	=       GLPNForDepthEstimation(lowercase_    )
     model.to(lowercase_    )
     model.eval()
     UpperCAmelCase_   :   Tuple     	=       model(lowercase_    )
     self.parent.assertEqual(result.predicted_depth.shape							,  (self.batch_size, self.image_size, self.image_size)    )
     UpperCAmelCase_   :   str     	=       model(lowercase_							,  labels=lowercase_    )
     self.parent.assertEqual(result.predicted_depth.shape							,  (self.batch_size, self.image_size, self.image_size)    )







  def 			UpperCamelCase__   (						self    ):

     """simple docstring"""
     UpperCAmelCase_   :   str     	=       self.prepare_config_and_inputs()
     UpperCAmelCase_	,  UpperCAmelCase_	,  UpperCAmelCase_   :   Tuple     	=       config_and_inputs
     UpperCAmelCase_   :   Dict     	=       {"pixel_values": pixel_values}
     return config, inputs_dict




@require_torch
class A_      (lowercase__							,lowercase__							,unittest.TestCase ):




  '''simple docstring'''
  SCREAMING_SNAKE_CASE__     :		Union[str, Any]    	=						(GLPNModel, GLPNForDepthEstimation) if is_torch_available() else ()
  SCREAMING_SNAKE_CASE__     :		Any    	=						(
      {"""depth-estimation""": GLPNForDepthEstimation, """feature-extraction""": GLPNModel} if is_torch_available() else {}
  )

  SCREAMING_SNAKE_CASE__     :		Union[str, Any]    	=						False
  SCREAMING_SNAKE_CASE__     :		Optional[int]    	=						False
  SCREAMING_SNAKE_CASE__     :		Any    	=						False







  def 			UpperCamelCase__   (						self    ):

     """simple docstring"""
     UpperCAmelCase_   :   Any     	=       GLPNModelTester(self    )
     UpperCAmelCase_   :   Optional[Any]     	=       GLPNConfigTester(self							,  config_class=lowercase_    )







  def 			UpperCamelCase__   (						self    ):

     """simple docstring"""
     self.config_tester.run_common_tests()







  def 			UpperCamelCase__   (						self    ):

     """simple docstring"""
     UpperCAmelCase_   :   Any     	=       self.model_tester.prepare_config_and_inputs()
     self.model_tester.create_and_check_model(*lowercase_    )







  def 			UpperCamelCase__   (						self    ):

     """simple docstring"""
     UpperCAmelCase_   :   int     	=       self.model_tester.prepare_config_and_inputs()
     self.model_tester.create_and_check_for_depth_estimation(*lowercase_    )







  @unittest.skip("GLPN does not use inputs_embeds"    )
  def 			UpperCamelCase__   (						self    ):

     """simple docstring"""
     pass







  @unittest.skip("GLPN does not have get_input_embeddings method and get_output_embeddings methods"    )
  def 			UpperCamelCase__   (						self    ):

     """simple docstring"""
     pass







  def 			UpperCamelCase__   (						self    ):

     """simple docstring"""
     UpperCAmelCase_	,  UpperCAmelCase_   :   Optional[Any]     	=       self.model_tester.prepare_config_and_inputs_for_common()

     for model_class in self.all_model_classes:
        UpperCAmelCase_   :   List[str]     	=       model_class(lowercase_    )
        UpperCAmelCase_   :   List[Any]     	=       inspect.signature(model.forward    )
        # signature.parameters is an OrderedDict => so arg_names order is deterministic
        UpperCAmelCase_   :   Optional[int]     	=       [*signature.parameters.keys()]

        UpperCAmelCase_   :   Any     	=       ["pixel_values"]
        self.assertListEqual(arg_names[:1]							,  lowercase_    )







  def 			UpperCamelCase__   (						self    ):

     """simple docstring"""
     UpperCAmelCase_	,  UpperCAmelCase_   :   Optional[int]     	=       self.model_tester.prepare_config_and_inputs_for_common()
     UpperCAmelCase_   :   Union[str, Any]     	=       True

     for model_class in self.all_model_classes:
        UpperCAmelCase_   :   Optional[int]     	=       True
        UpperCAmelCase_   :   int     	=       False
        UpperCAmelCase_   :   List[str]     	=       True
        UpperCAmelCase_   :   Union[str, Any]     	=       model_class(lowercase_    )
        model.to(lowercase_    )
        model.eval()
        with torch.no_grad():
           UpperCAmelCase_   :   Optional[int]     	=       model(**self._prepare_for_class(lowercase_							,  lowercase_    )    )
        UpperCAmelCase_   :   Union[str, Any]     	=       outputs.attentions

        UpperCAmelCase_   :   int     	=       sum(self.model_tester.depths    )
        self.assertEqual(len(lowercase_    )							,  lowercase_    )

        # check that output_attentions also work using config
        del inputs_dict["output_attentions"]
        UpperCAmelCase_   :   Optional[int]     	=       True
        UpperCAmelCase_   :   Dict     	=       model_class(lowercase_    )
        model.to(lowercase_    )
        model.eval()
        with torch.no_grad():
           UpperCAmelCase_   :   Tuple     	=       model(**self._prepare_for_class(lowercase_							,  lowercase_    )    )
        UpperCAmelCase_   :   Union[str, Any]     	=       outputs.attentions

        self.assertEqual(len(lowercase_    )							,  lowercase_    )

        # verify the first attentions (first block, first layer)
        UpperCAmelCase_   :   Union[str, Any]     	=       (self.model_tester.image_size // 4) ** 2
        UpperCAmelCase_   :   Optional[Any]     	=       (self.model_tester.image_size // (4 * self.model_tester.sr_ratios[0])) ** 2
        self.assertListEqual(
            list(attentions[0].shape[-3:]    )							,  [self.model_tester.num_attention_heads[0], expected_seq_len, expected_reduced_seq_len]							,  )

        # verify the last attentions (last block, last layer)
        UpperCAmelCase_   :   int     	=       (self.model_tester.image_size // 32) ** 2
        UpperCAmelCase_   :   Tuple     	=       (self.model_tester.image_size // (32 * self.model_tester.sr_ratios[-1])) ** 2
        self.assertListEqual(
            list(attentions[-1].shape[-3:]    )							,  [self.model_tester.num_attention_heads[-1], expected_seq_len, expected_reduced_seq_len]							,  )
        UpperCAmelCase_   :   Union[str, Any]     	=       len(lowercase_    )

        # Check attention is always last and order is fine
        UpperCAmelCase_   :   Optional[Any]     	=       True
        UpperCAmelCase_   :   Any     	=       True
        UpperCAmelCase_   :   str     	=       model_class(lowercase_    )
        model.to(lowercase_    )
        model.eval()
        with torch.no_grad():
           UpperCAmelCase_   :   str     	=       model(**self._prepare_for_class(lowercase_							,  lowercase_    )    )

        self.assertEqual(out_len + 1							,  len(lowercase_    )    )

        UpperCAmelCase_   :   List[str]     	=       outputs.attentions

        self.assertEqual(len(lowercase_    )							,  lowercase_    )
        # verify the first attentions (first block, first layer)
        UpperCAmelCase_   :   Tuple     	=       (self.model_tester.image_size // 4) ** 2
        UpperCAmelCase_   :   Any     	=       (self.model_tester.image_size // (4 * self.model_tester.sr_ratios[0])) ** 2
        self.assertListEqual(
            list(self_attentions[0].shape[-3:]    )							,  [self.model_tester.num_attention_heads[0], expected_seq_len, expected_reduced_seq_len]							,  )







  def 			UpperCamelCase__   (						self    ):

     """simple docstring"""
     def check_hidden_states_output(lowercase_							,  lowercase_							,  lowercase_    ):
        UpperCAmelCase_   :   str     	=       model_class(lowercase_    )
        model.to(lowercase_    )
        model.eval()

        with torch.no_grad():
           UpperCAmelCase_   :   Union[str, Any]     	=       model(**self._prepare_for_class(lowercase_							,  lowercase_    )    )

        UpperCAmelCase_   :   Optional[int]     	=       outputs.hidden_states

        UpperCAmelCase_   :   Union[str, Any]     	=       self.model_tester.num_encoder_blocks
        self.assertEqual(len(lowercase_    )							,  lowercase_    )

        # verify the first hidden states (first block)
        self.assertListEqual(
            list(hidden_states[0].shape[-3:]    )							,  [
                self.model_tester.hidden_sizes[0],
                self.model_tester.image_size // 4,
                self.model_tester.image_size // 4,
            ]							,  )

     UpperCAmelCase_	,  UpperCAmelCase_   :   List[str]     	=       self.model_tester.prepare_config_and_inputs_for_common()

     for model_class in self.all_model_classes:
        UpperCAmelCase_   :   Dict     	=       True
        check_hidden_states_output(lowercase_							,  lowercase_							,  lowercase_    )

        # check that output_hidden_states also work using config
        del inputs_dict["output_hidden_states"]
        UpperCAmelCase_   :   Dict     	=       True

        check_hidden_states_output(lowercase_							,  lowercase_							,  lowercase_    )







  def 			UpperCamelCase__   (						self    ):

     """simple docstring"""
     if not self.model_tester.is_training:
        return

     UpperCAmelCase_	,  UpperCAmelCase_   :   Optional[int]     	=       self.model_tester.prepare_config_and_inputs_for_common()
     UpperCAmelCase_   :   Optional[int]     	=       True

     for model_class in self.all_model_classes:
        if model_class in get_values(lowercase_    ):
           continue
        # TODO: remove the following 3 lines once we have a MODEL_FOR_DEPTH_ESTIMATION_MAPPING
        # this can then be incorporated into _prepare_for_class in test_modeling_common.py
        if model_class.__name__ == "GLPNForDepthEstimation":
           UpperCAmelCase_	,  UpperCAmelCase_	,  UpperCAmelCase_	,  UpperCAmelCase_   :   List[str]     	=       inputs_dict["pixel_values"].shape
           UpperCAmelCase_   :   str     	=       torch.zeros(
               [self.model_tester.batch_size, height, width]							,  device=lowercase_    ).long()
        UpperCAmelCase_   :   Tuple     	=       model_class(lowercase_    )
        model.to(lowercase_    )
        model.train()
        UpperCAmelCase_   :   Union[str, Any]     	=       self._prepare_for_class(lowercase_							,  lowercase_							,  return_labels=lowercase_    )
        UpperCAmelCase_   :   Optional[Any]     	=       model(**lowercase_    ).loss
        loss.backward()







  @slow
  def 			UpperCamelCase__   (						self    ):

     """simple docstring"""
     for model_name in GLPN_PRETRAINED_MODEL_ARCHIVE_LIST[:1]:
        UpperCAmelCase_   :   List[str]     	=       GLPNModel.from_pretrained(lowercase_    )
        self.assertIsNotNone(lowercase_    )







def       __a	(						):
   UpperCAmelCase_   :   Optional[int]     	=       Image.open("./tests/fixtures/tests_samples/COCO/000000039769.png"    )
   return image


@require_torch
@require_vision
@slow
class A_      (unittest.TestCase ):




  '''simple docstring'''







  @slow
  def 			UpperCamelCase__   (						self    ):

     """simple docstring"""
     UpperCAmelCase_   :   List[Any]     	=       GLPNImageProcessor.from_pretrained(GLPN_PRETRAINED_MODEL_ARCHIVE_LIST[0]    )
     UpperCAmelCase_   :   List[Any]     	=       GLPNForDepthEstimation.from_pretrained(GLPN_PRETRAINED_MODEL_ARCHIVE_LIST[0]    ).to(lowercase_    )

     UpperCAmelCase_   :   List[Any]     	=       prepare_img()
     UpperCAmelCase_   :   str     	=       image_processor(images=lowercase_							,  return_tensors="pt"    ).to(lowercase_    )

     # forward pass
     with torch.no_grad():
        UpperCAmelCase_   :   Any     	=       model(**lowercase_    )

     # verify the predicted depth
     UpperCAmelCase_   :   Optional[int]     	=       torch.Size([1, 480, 640]    )
     self.assertEqual(outputs.predicted_depth.shape							,  lowercase_    )

     UpperCAmelCase_   :   Dict     	=       torch.tensor(
         [[3.42_91, 2.78_65, 2.51_51], [3.28_41, 2.70_21, 2.35_02], [3.11_47, 2.46_25, 2.24_81]]    ).to(lowercase_    )

     self.assertTrue(torch.allclose(outputs.predicted_depth[0, :3, :3]							,  lowercase_							,  atol=1E-4    )    )





import logging
import pathlib
import re
import sys
from dataclasses import dataclass, field
from typing import Any, Callable, Dict, List, Optional, Set, Union

import datasets
import librosa
import numpy as np
import torch
from lang_trans import arabic
from packaging import version
from torch import nn

from transformers import (
    HfArgumentParser,
    Trainer,
    TrainingArguments,
    WavaVecaCTCTokenizer,
    WavaVecaFeatureExtractor,
    WavaVecaForCTC,
    WavaVecaProcessor,
    is_apex_available,
    trainer_utils,
)


if is_apex_available():
       from apex import amp

if version.parse(version.parse(torch.__version__).base_version) >= version.parse("1.6"):
       UpperCamelCase__	:   str								=     True
       from torch.cuda.amp import autocast


UpperCamelCase__	:   Tuple								=     logging.getLogger(__name__)



@dataclass
class    __snake_case	:
  __lowerCAmelCase  :						str    			=	field(
      metadata={'help': 'Path to pretrained model or model identifier from huggingface.co/models'}							)
  __lowerCAmelCase  :						Optional[str]    			=	field(
      default=lowerCAmelCase__  ,			metadata={'help': 'Where do you want to store the pretrained models downloaded from huggingface.co'}  ,			)
  __lowerCAmelCase  :						Optional[bool]    			=	field(
      default=lowerCAmelCase__  ,			metadata={'help': 'Whether to freeze the feature extractor layers of the model.'}							)
  __lowerCAmelCase  :						Optional[bool]    			=	field(
      default=lowerCAmelCase__  ,			metadata={'help': 'Whether to log verbose messages or not.'}  ,			)






def        _UpperCAmelCase							(	_SCREAMING_SNAKE_CASE       :		ModelArguments			, _SCREAMING_SNAKE_CASE       :		TrainingArguments							):





    """simple docstring"""


    logging.basicConfig(
        format='%(asctime)s - %(levelname)s - %(name)s - %(message)s'			, datefmt='%m/%d/%Y %H:%M:%S'			, handlers=[logging.StreamHandler(sys.stdout							)]			, )
    SCREAMING_SNAKE_CASE_						=    logging.WARNING
    if model_args.verbose_logging:
        SCREAMING_SNAKE_CASE_						=    logging.DEBUG
    elif trainer_utils.is_main_process(training_args.local_rank							):
        SCREAMING_SNAKE_CASE_						=    logging.INFO
    logger.setLevel(_SCREAMING_SNAKE_CASE							)



@dataclass
class    __snake_case	:
  __lowerCAmelCase  :						str    			=	field(
      default=lowerCAmelCase__  ,			metadata={'help': 'The name of the dataset to use (via the datasets library).'}							)
  __lowerCAmelCase  :						Optional[str]    			=	field(
      default=lowerCAmelCase__  ,			metadata={'help': 'The configuration name of the dataset to use (via the datasets library).'}							)
  __lowerCAmelCase  :						Optional[str]    			=	field(
      default='train'  ,			metadata={
          'help': 'The name of the training data set split to use (via the datasets library). Defaults to \'train\''
      }  ,			)
  __lowerCAmelCase  :						Optional[str]    			=	field(
      default='validation'  ,			metadata={
          'help': (
              'The name of the validation data set split to use (via the datasets library). Defaults to \'validation\''
          )
      }  ,			)
  __lowerCAmelCase  :						Optional[str]    			=	field(
      default='text'  ,			metadata={'help': 'Column in the dataset that contains label (target text). Defaults to \'text\''}  ,			)
  __lowerCAmelCase  :						Optional[str]    			=	field(
      default='file'  ,			metadata={'help': 'Column in the dataset that contains speech file path. Defaults to \'file\''}  ,			)
  __lowerCAmelCase  :						Optional[bool]    			=	field(
      default=lowerCAmelCase__  ,			metadata={'help': 'Resample loaded audio to target feature extractor\'s sampling rate or not.'}  ,			)
  __lowerCAmelCase  :						Optional[float]    			=	field(
      default=lowerCAmelCase__  ,			metadata={'help': 'Filters out examples longer than specified. Defaults to no filtering.'}  ,			)
  __lowerCAmelCase  :						Optional[str]    			=	field(
      default='librispeech'  ,			metadata={
          'help': (
              'Orthography used for normalization and tokenization: \'librispeech\' (default), \'timit\', or'
              ' \'buckwalter\'.'
          )
      }  ,			)
  __lowerCAmelCase  :						bool    			=	field(
      default=lowerCAmelCase__  ,			metadata={'help': 'Overwrite the cached preprocessed datasets or not.'}							)
  __lowerCAmelCase  :						Optional[int]    			=	field(
      default=lowerCAmelCase__  ,			metadata={'help': 'The number of processes to use for the preprocessing.'}  ,			)






@dataclass
class    __snake_case	:
  __lowerCAmelCase  :						bool    			=	False
  __lowerCAmelCase  :						Optional[str]    			=	None
  __lowerCAmelCase  :						Optional[str]    			=	"|"
  __lowerCAmelCase  :						Optional[Dict[str, str]]    			=	field(default_factory=lowerCAmelCase__							)
  __lowerCAmelCase  :						Optional[Set[str]]    			=	field(default_factory=lowerCAmelCase__							)
  __lowerCAmelCase  :						Optional[Callable[[str], str]]    			=	None

  @classmethod
  def     lowerCAmelCase__  (       cls    ,		_A):
      if name == "librispeech":
          return cls()
      if name == "timit":
          return cls(
              do_lower_case=_A    ,		translation_table=str.maketrans({'-': ' '})    ,		)
      if name == "buckwalter":
          SCREAMING_SNAKE_CASE_						=    {
              '-': ' ',  # sometimes used to represent pauses
              '^': 'v',  # fixing "tha" in arabic_speech_corpus dataset
          }
          return cls(
              vocab_file=pathlib.Path(_A).parent.joinpath('vocab/buckwalter.json')    ,		word_delimiter_token='/'    ,		translation_table=str.maketrans(_A)    ,		words_to_remove={'sil'}    ,		untransliterator=arabic.buckwalter.untransliterate    ,		)
      raise ValueError(f"""Unsupported orthography: '{name}'.""")

  def     lowerCAmelCase__  (       self    ,		_A):
      # TODO(elgeish) return a pipeline (e.g., from jiwer) instead? Or rely on branch predictor as is
      if len(self.translation_table) > 0:
          SCREAMING_SNAKE_CASE_						=    text.translate(self.translation_table)
      if len(self.words_to_remove) == 0:
          SCREAMING_SNAKE_CASE_						=    ' '.join(text.split())  # clean up whitespaces
      else:
          SCREAMING_SNAKE_CASE_						=    ' '.join(w for w in text.split() if w not in self.words_to_remove)  # and clean up whilespaces
      return text


  def     lowerCAmelCase__  (       self    ,		_A):
      SCREAMING_SNAKE_CASE_						=    WavaVecaFeatureExtractor.from_pretrained(
          model_args.model_name_or_path    ,		cache_dir=model_args.cache_dir)
      if self.vocab_file:
          SCREAMING_SNAKE_CASE_						=    WavaVecaCTCTokenizer(
              self.vocab_file    ,		cache_dir=model_args.cache_dir    ,		do_lower_case=self.do_lower_case    ,		word_delimiter_token=self.word_delimiter_token    ,		)
      else:
          SCREAMING_SNAKE_CASE_						=    WavaVecaCTCTokenizer.from_pretrained(
              model_args.model_name_or_path    ,		cache_dir=model_args.cache_dir    ,		do_lower_case=self.do_lower_case    ,		word_delimiter_token=self.word_delimiter_token    ,		)
      return WavaVecaProcessor(_A    ,		_A)






@dataclass
class    __snake_case	:
  __lowerCAmelCase  :						WavaVecaProcessor
  __lowerCAmelCase  :						Union[bool, str]    			=	True
  __lowerCAmelCase  :						Optional[int]    			=	None
  __lowerCAmelCase  :						Optional[int]    			=	None
  __lowerCAmelCase  :						Optional[int]    			=	None
  __lowerCAmelCase  :						Optional[int]    			=	None

  def __call__(       self    ,		_A):
      # split inputs and labels since they have to be of different lenghts and need
      # different padding methods
      SCREAMING_SNAKE_CASE_						=    [{'input_values': feature['input_values']} for feature in features]
      SCREAMING_SNAKE_CASE_						=    [{'input_ids': feature['labels']} for feature in features]

      SCREAMING_SNAKE_CASE_						=    self.processor.pad(
          _A    ,		padding=self.padding    ,		max_length=self.max_length    ,		pad_to_multiple_of=self.pad_to_multiple_of    ,		return_tensors='pt'    ,		)
      SCREAMING_SNAKE_CASE_						=    self.processor.pad(
          labels=_A    ,		padding=self.padding    ,		max_length=self.max_length_labels    ,		pad_to_multiple_of=self.pad_to_multiple_of_labels    ,		return_tensors='pt'    ,		)

      # replace padding with -100 to ignore loss correctly
      SCREAMING_SNAKE_CASE_						=    labels_batch['input_ids'].masked_fill(labels_batch.attention_mask.ne(1)    ,		-100)

      SCREAMING_SNAKE_CASE_						=    labels

      return batch






class    __snake_case	(     lowerCAmelCase__							):

  def     lowerCAmelCase__  (       self    ,		_A    ,		_A):
      model.train()
      SCREAMING_SNAKE_CASE_						=    self._prepare_inputs(_A)

      if self.use_amp:
          with autocast():
              SCREAMING_SNAKE_CASE_						=    self.compute_loss(_A    ,		_A)
      else:
          SCREAMING_SNAKE_CASE_						=    self.compute_loss(_A    ,		_A)

      if self.args.n_gpu > 1:
          if model.module.config.ctc_loss_reduction == "mean":
              SCREAMING_SNAKE_CASE_						=    loss.mean()
          elif model.module.config.ctc_loss_reduction == "sum":
              SCREAMING_SNAKE_CASE_						=    loss.sum() / (inputs['labels'] >= 0).sum()
          else:
              raise ValueError(f"""{model.config.ctc_loss_reduction} is not valid. Choose one of ['mean', 'sum']""")

      if self.args.gradient_accumulation_steps > 1:
          SCREAMING_SNAKE_CASE_						=    loss / self.args.gradient_accumulation_steps

      if self.use_amp:
          self.scaler.scale(_A).backward()
      elif self.use_apex:
          with amp.scale_loss(_A    ,		self.optimizer) as scaled_loss:
              scaled_loss.backward()
      elif self.deepspeed:
          self.deepspeed.backward(_A)
      else:
          loss.backward()

      return loss.detach()






def        _UpperCAmelCase							(	):





    """simple docstring"""


    SCREAMING_SNAKE_CASE_						=    HfArgumentParser((ModelArguments, DataTrainingArguments, TrainingArguments)							)

    SCREAMING_SNAKE_CASE_				,      SCREAMING_SNAKE_CASE_				,      SCREAMING_SNAKE_CASE_						=    parser.parse_args_into_dataclasses()
    configure_logger(_SCREAMING_SNAKE_CASE			, _SCREAMING_SNAKE_CASE							)

    SCREAMING_SNAKE_CASE_						=    Orthography.from_name(data_args.orthography.lower()							)
    SCREAMING_SNAKE_CASE_						=    orthography.create_processor(_SCREAMING_SNAKE_CASE							)
    SCREAMING_SNAKE_CASE_						=    WavaVecaForCTC.from_pretrained(
        model_args.model_name_or_path			, cache_dir=model_args.cache_dir			, gradient_checkpointing=training_args.gradient_checkpointing			, vocab_size=len(processor.tokenizer							)			, )

    SCREAMING_SNAKE_CASE_						=    datasets.load_dataset(
        data_args.dataset_name			, data_args.dataset_config_name			, split=data_args.train_split_name							)
    SCREAMING_SNAKE_CASE_						=    datasets.load_dataset(
        data_args.dataset_name			, data_args.dataset_config_name			, split=data_args.validation_split_name							)

    SCREAMING_SNAKE_CASE_						=    datasets.load_metric('wer'							)
    SCREAMING_SNAKE_CASE_						=    processor.feature_extractor.sampling_rate if data_args.target_feature_extractor_sampling_rate else None
    SCREAMING_SNAKE_CASE_						=    ''.join(t for t in processor.tokenizer.get_vocab().keys() if len(_SCREAMING_SNAKE_CASE							) == 1							)
    SCREAMING_SNAKE_CASE_						=    re.compile(  # remove characters not in vocabulary
        rf"""[^\s{re.escape(_SCREAMING_SNAKE_CASE							)}]"""			, flags=re.IGNORECASE if processor.tokenizer.do_lower_case else 0			, )
    SCREAMING_SNAKE_CASE_						=    []

    def prepare_example(_SCREAMING_SNAKE_CASE       :		Any							):  # TODO(elgeish) make use of multiprocessing?
        SCREAMING_SNAKE_CASE_				,      SCREAMING_SNAKE_CASE_						=    librosa.load(example[data_args.speech_file_column]			, sr=_SCREAMING_SNAKE_CASE							)
        if data_args.max_duration_in_seconds is not None:
            SCREAMING_SNAKE_CASE_						=    len(example['speech']							) / example['sampling_rate']
        # Normalize and clean up text; order matters!
        SCREAMING_SNAKE_CASE_						=    orthography.preprocess_for_training(example[data_args.target_text_column]							)
        SCREAMING_SNAKE_CASE_						=    vocabulary_text_cleaner.sub(''			, _SCREAMING_SNAKE_CASE							)
        if updated_text != example[data_args.target_text_column]:
            text_updates.append((example[data_args.target_text_column], updated_text)							)
            SCREAMING_SNAKE_CASE_						=    updated_text
        return example

    SCREAMING_SNAKE_CASE_						=    train_dataset.map(_SCREAMING_SNAKE_CASE			, remove_columns=[data_args.speech_file_column]							)
    SCREAMING_SNAKE_CASE_						=    val_dataset.map(_SCREAMING_SNAKE_CASE			, remove_columns=[data_args.speech_file_column]							)

    if data_args.max_duration_in_seconds is not None:

        def filter_by_max_duration(_SCREAMING_SNAKE_CASE       :		Tuple							):
            return example["duration_in_seconds"] <= data_args.max_duration_in_seconds

        SCREAMING_SNAKE_CASE_						=    len(_SCREAMING_SNAKE_CASE							)
        SCREAMING_SNAKE_CASE_						=    len(_SCREAMING_SNAKE_CASE							)
        SCREAMING_SNAKE_CASE_						=    train_dataset.filter(_SCREAMING_SNAKE_CASE			, remove_columns=['duration_in_seconds']							)
        SCREAMING_SNAKE_CASE_						=    val_dataset.filter(_SCREAMING_SNAKE_CASE			, remove_columns=['duration_in_seconds']							)
        if len(_SCREAMING_SNAKE_CASE							) > old_train_size:
            logger.warning(
                f"""Filtered out {len(_SCREAMING_SNAKE_CASE							) - old_train_size} train example(s) longer than"""
                f""" {data_args.max_duration_in_seconds} second(s)."""							)
        if len(_SCREAMING_SNAKE_CASE							) > old_val_size:
            logger.warning(
                f"""Filtered out {len(_SCREAMING_SNAKE_CASE							) - old_val_size} validation example(s) longer than"""
                f""" {data_args.max_duration_in_seconds} second(s)."""							)
    logger.info(f"""Split sizes: {len(_SCREAMING_SNAKE_CASE							)} train and {len(_SCREAMING_SNAKE_CASE							)} validation."""							)

    logger.warning(f"""Updated {len(_SCREAMING_SNAKE_CASE							)} transcript(s) using '{data_args.orthography}' orthography rules."""							)
    if logger.isEnabledFor(logging.DEBUG							):
        for original_text, updated_text in text_updates:
            logger.debug(f"""Updated text: \"{original_text}\" -> \"{updated_text}\""""							)
    SCREAMING_SNAKE_CASE_						=    None

    def prepare_dataset(_SCREAMING_SNAKE_CASE       :		List[Any]							):
        # check that all files have the correct sampling rate
        assert (
            len(set(batch['sampling_rate']							)							) == 1
        ), f"""Make sure all inputs have the same sampling rate of {processor.feature_extractor.sampling_rate}."""

        SCREAMING_SNAKE_CASE_						=    processor(
            audio=batch['speech']			, text=batch[data_args.target_text_column]			, sampling_rate=batch['sampling_rate'][0]							)
        batch.update(_SCREAMING_SNAKE_CASE							)
        return batch

    SCREAMING_SNAKE_CASE_						=    train_dataset.map(
        _SCREAMING_SNAKE_CASE			, batch_size=training_args.per_device_train_batch_size			, batched=_SCREAMING_SNAKE_CASE			, num_proc=data_args.preprocessing_num_workers			, )
    SCREAMING_SNAKE_CASE_						=    val_dataset.map(
        _SCREAMING_SNAKE_CASE			, batch_size=training_args.per_device_train_batch_size			, batched=_SCREAMING_SNAKE_CASE			, num_proc=data_args.preprocessing_num_workers			, )

    SCREAMING_SNAKE_CASE_						=    DataCollatorCTCWithPadding(processor=_SCREAMING_SNAKE_CASE			, padding=_SCREAMING_SNAKE_CASE							)

    def compute_metrics(_SCREAMING_SNAKE_CASE       :		Optional[int]							):
        SCREAMING_SNAKE_CASE_						=    pred.predictions
        SCREAMING_SNAKE_CASE_						=    np.argmax(_SCREAMING_SNAKE_CASE			, axis=-1							)

        SCREAMING_SNAKE_CASE_						=    processor.tokenizer.pad_token_id

        SCREAMING_SNAKE_CASE_						=    processor.batch_decode(_SCREAMING_SNAKE_CASE							)
        # we do not want to group tokens when computing the metrics
        SCREAMING_SNAKE_CASE_						=    processor.batch_decode(pred.label_ids			, group_tokens=_SCREAMING_SNAKE_CASE							)
        if logger.isEnabledFor(logging.DEBUG							):
            for reference, predicted in zip(_SCREAMING_SNAKE_CASE			, _SCREAMING_SNAKE_CASE							):
                logger.debug(f"""reference: \"{reference}\""""							)
                logger.debug(f"""predicted: \"{predicted}\""""							)
                if orthography.untransliterator is not None:
                    logger.debug(f"""reference (untransliterated): \"{orthography.untransliterator(_SCREAMING_SNAKE_CASE							)}\""""							)
                    logger.debug(f"""predicted (untransliterated): \"{orthography.untransliterator(_SCREAMING_SNAKE_CASE							)}\""""							)

        SCREAMING_SNAKE_CASE_						=    wer_metric.compute(predictions=_SCREAMING_SNAKE_CASE			, references=_SCREAMING_SNAKE_CASE							)

        return {"wer": wer}

    if model_args.freeze_feature_extractor:
        model.freeze_feature_extractor()

    SCREAMING_SNAKE_CASE_						=    CTCTrainer(
        model=_SCREAMING_SNAKE_CASE			, data_collator=_SCREAMING_SNAKE_CASE			, args=_SCREAMING_SNAKE_CASE			, compute_metrics=_SCREAMING_SNAKE_CASE			, train_dataset=_SCREAMING_SNAKE_CASE			, eval_dataset=_SCREAMING_SNAKE_CASE			, tokenizer=processor.feature_extractor			, )

    trainer.train()


if __name__ == "__main__":
       main()


