






'''simple docstring'''

import argparse
import json
from pathlib import Path

import requests
import torch
from huggingface_hub import hf_hub_download
from PIL import Image

from transformers import (
    SwiftFormerConfig,
    SwiftFormerForImageClassification,
    ViTImageProcessor,
)
from transformers.utils import logging


logging.set_verbosity_info()
UpperCamelCase							=logging.get_logger(__name__)

UpperCamelCase							=torch.device("cpu")



def 							snake_case					(     )       ->				List[Any]:





      """simple docstring"""

      UpperCamelCase_				:   Optional[Any]					    =							"""http://images.cocodataset.org/val2017/000000039769.jpg"""
      UpperCamelCase_				:   str					    =							Image.open(requests.get(__snake_case    ,   stream=__snake_case	).raw	)
      return im



def 							snake_case					(     a_							: Optional[int]	)       ->				int:





      """simple docstring"""

      if swiftformer_name == "swiftformer_xs":
            return torch.tensor([-2.1_703e00, 2.1_107e00, -2.0_811e00, 8.8_685e-01, 2.4_360e-01]	)

      elif swiftformer_name == "swiftformer_s":
            return torch.tensor([3.9_636e-01, 2.3_478e-01, -1.6_963e00, -1.7_381e00, -8.6_337e-01]	)

      elif swiftformer_name == "swiftformer_l1":
            return torch.tensor([-4.2_768e-01, -4.7_429e-01, -1.0_897e00, -1.0_248e00, 3.5_523e-02]	)

      elif swiftformer_name == "swiftformer_l3":
            return torch.tensor([-2.5_330e-01, 2.4_211e-01, -6.0_185e-01, -8.2_789e-01, -6.0_446e-02]	)



def 							snake_case					(     a_							: Optional[int]    ,   a_							: Optional[int]    ,   a_							: List[str]	)       ->				Tuple:





      """simple docstring"""

      UpperCamelCase_				:   List[str]					    =							dct.pop(__snake_case	)
      UpperCamelCase_				:   List[Any]					    =							val



def 							snake_case					(     a_							: Optional[Any]	)       ->				int:





      """simple docstring"""

      UpperCamelCase_				:   Any					    =							[]
      for k in state_dict.keys():
            UpperCamelCase_				:   List[Any]					    =							k
            if ".pwconv" in k:
                  UpperCamelCase_				:   Optional[int]					    =							k_new.replace(""".pwconv"""    ,   """.point_wise_conv"""	)
            if ".dwconv" in k:
                  UpperCamelCase_				:   Tuple					    =							k_new.replace(""".dwconv"""    ,   """.depth_wise_conv"""	)
            if ".Proj." in k:
                  UpperCamelCase_				:   int					    =							k_new.replace(""".Proj."""    ,   """.proj."""	)
            if "patch_embed" in k_new:
                  UpperCamelCase_				:   Tuple					    =							k_new.replace("""patch_embed"""    ,   """swiftformer.patch_embed.patch_embedding"""	)
            if "network" in k_new:
                  UpperCamelCase_				:   Optional[Any]					    =							k_new.split("""."""	)
                  if ls[2].isdigit():
                        UpperCamelCase_				:   Any					    =							"""swiftformer.encoder.network.""" + ls[1] + """.blocks.""" + ls[2] + """.""" + """.""".join(ls[3:]	)
                  else:
                        UpperCamelCase_				:   Optional[Any]					    =							k_new.replace("""network"""    ,   """swiftformer.encoder.network"""	)
            rename_keys.append((k, k_new)	)
      return rename_keys



@torch.no_grad()
def 							snake_case					(     a_							: Optional[int]    ,   a_							: int    ,   a_							: Dict	)       ->				Optional[Any]:





      """simple docstring"""

      UpperCamelCase_				:   List[str]					    =							SwiftFormerConfig()

      # dataset (ImageNet-21k only or also fine-tuned on ImageNet 2012), patch_size and image_size
      UpperCamelCase_				:   Optional[int]					    =							1_000
      UpperCamelCase_				:   Any					    =							"""huggingface/label-files"""
      UpperCamelCase_				:   Dict					    =							"""imagenet-1k-id2label.json"""
      UpperCamelCase_				:   List[Any]					    =							json.load(open(hf_hub_download(__snake_case    ,   __snake_case    ,   repo_type="""dataset"""	)    ,   """r"""	)	)
      UpperCamelCase_				:   Optional[Any]					    =							{int(__snake_case	): v for k, v in idalabel.items()}
      UpperCamelCase_				:   List[str]					    =							idalabel
      UpperCamelCase_				:   List[Any]					    =							{v: k for k, v in idalabel.items()}

      # size of the architecture
      if swiftformer_name == "swiftformer_xs":
            UpperCamelCase_				:   Tuple					    =							[3, 3, 6, 4]
            UpperCamelCase_				:   Tuple					    =							[48, 56, 112, 220]

      elif swiftformer_name == "swiftformer_s":
            UpperCamelCase_				:   Any					    =							[3, 3, 9, 6]
            UpperCamelCase_				:   Any					    =							[48, 64, 168, 224]

      elif swiftformer_name == "swiftformer_l1":
            UpperCamelCase_				:   Any					    =							[4, 3, 10, 5]
            UpperCamelCase_				:   Optional[int]					    =							[48, 96, 192, 384]

      elif swiftformer_name == "swiftformer_l3":
            UpperCamelCase_				:   int					    =							[4, 4, 12, 6]
            UpperCamelCase_				:   str					    =							[64, 128, 320, 512]

      # load state_dict of original model, remove and rename some keys
      if original_ckpt:
            if original_ckpt.startswith("""https"""	):
                  UpperCamelCase_				:   Optional[int]					    =							torch.hub.load_state_dict_from_url(__snake_case    ,   map_location="""cpu"""    ,   check_hash=__snake_case	)
            else:
                  UpperCamelCase_				:   Optional[int]					    =							torch.load(__snake_case    ,   map_location="""cpu"""	)
      UpperCamelCase_				:   Any					    =							checkpoint

      UpperCamelCase_				:   List[Any]					    =							create_rename_keys(__snake_case	)
      for rename_key_src, rename_key_dest in rename_keys:
            rename_key(__snake_case    ,   __snake_case    ,   __snake_case	)

      # load HuggingFace model
      UpperCamelCase_				:   Tuple					    =							SwiftFormerForImageClassification(__snake_case	).eval()
      hf_model.load_state_dict(__snake_case	)

      # prepare test inputs
      UpperCamelCase_				:   Optional[Any]					    =							prepare_img()
      UpperCamelCase_				:   List[Any]					    =							ViTImageProcessor.from_pretrained("""preprocessor_config"""	)
      UpperCamelCase_				:   List[str]					    =							processor(images=__snake_case    ,   return_tensors="""pt"""	)

      # compare outputs from both models
      UpperCamelCase_				:   str					    =							get_expected_output(__snake_case	)
      UpperCamelCase_				:   Dict					    =							hf_model(inputs["""pixel_values"""]	).logits

      assert hf_logits.shape == torch.Size([1, 1_000]	)
      assert torch.allclose(hf_logits[0, 0:5]    ,   __snake_case    ,   atol=1e-3	)

      Path(__snake_case	).mkdir(exist_ok=__snake_case	)
      print(f"Saving model {swiftformer_name} to {pytorch_dump_folder_path}"	)
      hf_model.save_pretrained(__snake_case	)


if __name__ == "__main__":
      UpperCamelCase							=argparse.ArgumentParser()
      # Required parameters
      parser.add_argument(
          "--swiftformer_name",
          default="swiftformer_xs",
          choices=["swiftformer_xs", "swiftformer_s", "swiftformer_l1", "swiftformer_l3"],
          type=str,
          help="Name of the SwiftFormer model you'd like to convert.",
      )
      parser.add_argument(
          "--pytorch_dump_folder_path",
          default="./converted_outputs/",
          type=str,
          help="Path to the output PyTorch model directory.",
      )
      parser.add_argument("--original_ckpt", default=None, type=str, help="Path to the original model checkpoint.")

      UpperCamelCase							=parser.parse_args()
      convert_swiftformer_checkpoint(args.swiftformer_name, args.pytorch_dump_folder_path, args.original_ckpt)













'''simple docstring'''

import json
import os
import shutil
import tempfile
import unittest
from multiprocessing import get_context
from pathlib import Path

import datasets
import numpy as np
from datasets import load_dataset
from parameterized import parameterized

from transformers import AutoProcessor
from transformers.models.wavaveca import WavaVecaCTCTokenizer, WavaVecaFeatureExtractor
from transformers.models.wavaveca.tokenization_wavaveca import VOCAB_FILES_NAMES
from transformers.testing_utils import require_pyctcdecode, require_torch, require_torchaudio, slow
from transformers.utils import FEATURE_EXTRACTOR_NAME, is_pyctcdecode_available, is_torch_available

from ..wavaveca.test_feature_extraction_wavaveca import floats_list


if is_pyctcdecode_available():
      from huggingface_hub import snapshot_download
      from pyctcdecode import BeamSearchDecoderCTC

      from transformers.models.wavaveca_with_lm import WavaVecaProcessorWithLM
      from transformers.models.wavaveca_with_lm.processing_wavaveca_with_lm import WavaVecaDecoderWithLMOutput

if is_torch_available():
      from transformers import WavaVecaForCTC




@require_pyctcdecode
class     A    (      unittest.TestCase						):






 """simple docstring"""





 def   _UpperCAmelCase       (							self ):
       UpperCamelCase_				:   str					    =							"""| <pad> <unk> <s> </s> a b c d e f g h i j k""".split()
       UpperCamelCase_				:   int					    =							dict(zip(__lowerCAmelCase			,     range(len(__lowerCAmelCase ) ) ) )

       UpperCamelCase_				:   Tuple					    =							{
           """unk_token""": """<unk>""",
           """bos_token""": """<s>""",
           """eos_token""": """</s>""",
       }
       UpperCamelCase_				:   Optional[Any]					    =							{
           """feature_size""": 1,
           """padding_value""": 0.0,
           """sampling_rate""": 1_60_00,
           """return_attention_mask""": False,
           """do_normalize""": True,
       }

       UpperCamelCase_				:   int					    =							tempfile.mkdtemp()
       UpperCamelCase_				:   Tuple					    =							os.path.join(self.tmpdirname			,     VOCAB_FILES_NAMES["""vocab_file"""] )
       UpperCamelCase_				:   List[Any]					    =							os.path.join(self.tmpdirname			,     __lowerCAmelCase )
       with open(self.vocab_file			,     """w"""			,     encoding="""utf-8""" ) as fp:
             fp.write(json.dumps(__lowerCAmelCase ) + """\n""" )

       with open(self.feature_extraction_file			,     """w"""			,     encoding="""utf-8""" ) as fp:
             fp.write(json.dumps(__lowerCAmelCase ) + """\n""" )

       # load decoder from hub
       UpperCamelCase_				:   Union[str, Any]					    =							"""hf-internal-testing/ngram-beam-search-decoder"""
 def   _UpperCAmelCase       (							self			,     **__lowerCAmelCase ):
       UpperCamelCase_				:   str					    =							self.add_kwargs_tokens_map.copy()
       kwargs.update(__lowerCAmelCase )
       return WavaVecaCTCTokenizer.from_pretrained(self.tmpdirname			,     **__lowerCAmelCase )
 def   _UpperCAmelCase       (							self			,     **__lowerCAmelCase ):
       return WavaVecaFeatureExtractor.from_pretrained(self.tmpdirname			,     **__lowerCAmelCase )
 def   _UpperCAmelCase       (							self			,     **__lowerCAmelCase ):
       return BeamSearchDecoderCTC.load_from_hf_hub(self.decoder_name			,     **__lowerCAmelCase )
 def   _UpperCAmelCase       (							self ):
       shutil.rmtree(self.tmpdirname )
 def   _UpperCAmelCase       (							self ):
       UpperCamelCase_				:   Union[str, Any]					    =							self.get_tokenizer()
       UpperCamelCase_				:   str					    =							self.get_feature_extractor()
       UpperCamelCase_				:   Tuple					    =							self.get_decoder()

       UpperCamelCase_				:   int					    =							WavaVecaProcessorWithLM(tokenizer=__lowerCAmelCase			,     feature_extractor=__lowerCAmelCase			,     decoder=__lowerCAmelCase )

       processor.save_pretrained(self.tmpdirname )
       UpperCamelCase_				:   Dict					    =							WavaVecaProcessorWithLM.from_pretrained(self.tmpdirname )

       # tokenizer
       self.assertEqual(processor.tokenizer.get_vocab()			,     tokenizer.get_vocab() )
       self.assertIsInstance(processor.tokenizer			,     __lowerCAmelCase )

       # feature extractor
       self.assertEqual(processor.feature_extractor.to_json_string()			,     feature_extractor.to_json_string() )
       self.assertIsInstance(processor.feature_extractor			,     __lowerCAmelCase )

       # decoder
       self.assertEqual(processor.decoder._alphabet.labels			,     decoder._alphabet.labels )
       self.assertEqual(
           processor.decoder.model_container[decoder._model_key]._unigram_set			,     decoder.model_container[decoder._model_key]._unigram_set			,     )
       self.assertIsInstance(processor.decoder			,     __lowerCAmelCase )
 def   _UpperCAmelCase       (							self ):
       UpperCamelCase_				:   Union[str, Any]					    =							WavaVecaProcessorWithLM(
           tokenizer=self.get_tokenizer()			,     feature_extractor=self.get_feature_extractor()			,     decoder=self.get_decoder() )
       processor.save_pretrained(self.tmpdirname )

       # make sure that error is thrown when decoder alphabet doesn't match
       UpperCamelCase_				:   Optional[Any]					    =							WavaVecaProcessorWithLM.from_pretrained(
           self.tmpdirname			,     alpha=5.0			,     beta=3.0			,     score_boundary=-7.0			,     unk_score_offset=3 )

       # decoder
       self.assertEqual(processor.language_model.alpha			,     5.0 )
       self.assertEqual(processor.language_model.beta			,     3.0 )
       self.assertEqual(processor.language_model.score_boundary			,     -7.0 )
       self.assertEqual(processor.language_model.unk_score_offset			,     3 )
 def   _UpperCAmelCase       (							self ):
       UpperCamelCase_				:   int					    =							self.get_tokenizer()
       # add token to trigger raise
       tokenizer.add_tokens(["""xx"""] )
       with self.assertRaisesRegex(__lowerCAmelCase			,     """include""" ):
             WavaVecaProcessorWithLM(
                 tokenizer=__lowerCAmelCase			,     feature_extractor=self.get_feature_extractor()			,     decoder=self.get_decoder() )
 def   _UpperCAmelCase       (							self ):
       UpperCamelCase_				:   Tuple					    =							self.get_feature_extractor()
       UpperCamelCase_				:   Tuple					    =							self.get_tokenizer()
       UpperCamelCase_				:   Any					    =							self.get_decoder()

       UpperCamelCase_				:   List[Any]					    =							WavaVecaProcessorWithLM(tokenizer=__lowerCAmelCase			,     feature_extractor=__lowerCAmelCase			,     decoder=__lowerCAmelCase )

       UpperCamelCase_				:   List[Any]					    =							floats_list((3, 10_00) )

       UpperCamelCase_				:   Tuple					    =							feature_extractor(__lowerCAmelCase			,     return_tensors="""np""" )
       UpperCamelCase_				:   str					    =							processor(__lowerCAmelCase			,     return_tensors="""np""" )

       for key in input_feat_extract.keys():
             self.assertAlmostEqual(input_feat_extract[key].sum()			,     input_processor[key].sum()			,     delta=1E-2 )
 def   _UpperCAmelCase       (							self ):
       UpperCamelCase_				:   Dict					    =							self.get_feature_extractor()
       UpperCamelCase_				:   List[Any]					    =							self.get_tokenizer()
       UpperCamelCase_				:   List[Any]					    =							self.get_decoder()

       UpperCamelCase_				:   Optional[Any]					    =							WavaVecaProcessorWithLM(tokenizer=__lowerCAmelCase			,     feature_extractor=__lowerCAmelCase			,     decoder=__lowerCAmelCase )

       UpperCamelCase_				:   List[str]					    =							"""This is a test string"""

       UpperCamelCase_				:   Optional[Any]					    =							processor(text=__lowerCAmelCase )

       UpperCamelCase_				:   int					    =							tokenizer(__lowerCAmelCase )

       for key in encoded_tok.keys():
             self.assertListEqual(encoded_tok[key]			,     encoded_processor[key] )
 def   _UpperCAmelCase       (							self			,     __lowerCAmelCase=(2, 10, 16)			,     __lowerCAmelCase=77 ):
       np.random.seed(__lowerCAmelCase )
       return np.random.rand(*__lowerCAmelCase )
 def   _UpperCAmelCase       (							self ):
       UpperCamelCase_				:   int					    =							self.get_feature_extractor()
       UpperCamelCase_				:   Tuple					    =							self.get_tokenizer()
       UpperCamelCase_				:   Optional[int]					    =							self.get_decoder()

       UpperCamelCase_				:   int					    =							WavaVecaProcessorWithLM(tokenizer=__lowerCAmelCase			,     feature_extractor=__lowerCAmelCase			,     decoder=__lowerCAmelCase )

       UpperCamelCase_				:   List[str]					    =							self._get_dummy_logits(shape=(10, 16)			,     seed=13 )

       UpperCamelCase_				:   Any					    =							processor.decode(__lowerCAmelCase )

       UpperCamelCase_				:   Any					    =							decoder.decode_beams(__lowerCAmelCase )[0]

       self.assertEqual(decoded_decoder[0]			,     decoded_processor.text )
       self.assertEqual("""</s> <s> </s>"""			,     decoded_processor.text )
       self.assertEqual(decoded_decoder[-2]			,     decoded_processor.logit_score )
       self.assertEqual(decoded_decoder[-1]			,     decoded_processor.lm_score )
 @parameterized.expand([[None], ["""fork"""], ["""spawn"""]] )
 def   _UpperCAmelCase       (							self			,     __lowerCAmelCase ):
       UpperCamelCase_				:   Union[str, Any]					    =							self.get_feature_extractor()
       UpperCamelCase_				:   str					    =							self.get_tokenizer()
       UpperCamelCase_				:   List[Any]					    =							self.get_decoder()

       UpperCamelCase_				:   Optional[Any]					    =							WavaVecaProcessorWithLM(tokenizer=__lowerCAmelCase			,     feature_extractor=__lowerCAmelCase			,     decoder=__lowerCAmelCase )

       UpperCamelCase_				:   Union[str, Any]					    =							self._get_dummy_logits()

       # note: pool should be instantiated *after* Wav2Vec2ProcessorWithLM.
       #       otherwise, the LM won't be available to the pool's sub-processes.
       # manual logic used to allow parameterized test for both pool=None and pool=Pool(...)
       if pool_context is None:
             UpperCamelCase_				:   List[Any]					    =							processor.batch_decode(__lowerCAmelCase )
       else:
             with get_context(__lowerCAmelCase ).Pool() as pool:
                   UpperCamelCase_				:   Any					    =							processor.batch_decode(__lowerCAmelCase			,     __lowerCAmelCase )

       UpperCamelCase_				:   Tuple					    =							list(__lowerCAmelCase )

       with get_context("""fork""" ).Pool() as p:
             UpperCamelCase_				:   Optional[int]					    =							decoder.decode_beams_batch(__lowerCAmelCase			,     __lowerCAmelCase )

       UpperCamelCase_    ,			UpperCamelCase_    ,			UpperCamelCase_				:   Union[str, Any]					    =							[], [], []
       for beams in decoded_beams:
             texts_decoder.append(beams[0][0] )
             logit_scores_decoder.append(beams[0][-2] )
             lm_scores_decoder.append(beams[0][-1] )

       self.assertListEqual(__lowerCAmelCase			,     decoded_processor.text )
       self.assertListEqual(["""<s> <s> </s>""", """<s> <s> <s>"""]			,     decoded_processor.text )
       self.assertListEqual(__lowerCAmelCase			,     decoded_processor.logit_score )
       self.assertListEqual(__lowerCAmelCase			,     decoded_processor.lm_score )
 def   _UpperCAmelCase       (							self ):
       UpperCamelCase_				:   Dict					    =							self.get_feature_extractor()
       UpperCamelCase_				:   Tuple					    =							self.get_tokenizer()
       UpperCamelCase_				:   Tuple					    =							self.get_decoder()

       UpperCamelCase_				:   Optional[Any]					    =							WavaVecaProcessorWithLM(tokenizer=__lowerCAmelCase			,     feature_extractor=__lowerCAmelCase			,     decoder=__lowerCAmelCase )

       UpperCamelCase_				:   List[str]					    =							self._get_dummy_logits()

       UpperCamelCase_				:   Dict					    =							15
       UpperCamelCase_				:   str					    =							-20.0
       UpperCamelCase_				:   Dict					    =							-4.0

       UpperCamelCase_				:   Union[str, Any]					    =							processor.batch_decode(
           __lowerCAmelCase			,     beam_width=__lowerCAmelCase			,     beam_prune_logp=__lowerCAmelCase			,     token_min_logp=__lowerCAmelCase			,     )
       UpperCamelCase_				:   Any					    =							decoded_processor_out.text

       UpperCamelCase_				:   Tuple					    =							list(__lowerCAmelCase )

       with get_context("""fork""" ).Pool() as pool:
             UpperCamelCase_				:   str					    =							decoder.decode_beams_batch(
                 __lowerCAmelCase			,     __lowerCAmelCase			,     beam_width=__lowerCAmelCase			,     beam_prune_logp=__lowerCAmelCase			,     token_min_logp=__lowerCAmelCase			,     )

       UpperCamelCase_				:   str					    =							[d[0][0] for d in decoded_decoder_out]
       UpperCamelCase_				:   List[str]					    =							[d[0][2] for d in decoded_decoder_out]
       UpperCamelCase_				:   Union[str, Any]					    =							[d[0][3] for d in decoded_decoder_out]

       self.assertListEqual(__lowerCAmelCase			,     __lowerCAmelCase )
       self.assertListEqual(["""</s> <s> <s>""", """<s> <s> <s>"""]			,     __lowerCAmelCase )

       self.assertTrue(np.array_equal(__lowerCAmelCase			,     decoded_processor_out.logit_score ) )
       self.assertTrue(np.allclose([-20.0_54, -18.4_47]			,     __lowerCAmelCase			,     atol=1E-3 ) )

       self.assertTrue(np.array_equal(__lowerCAmelCase			,     decoded_processor_out.lm_score ) )
       self.assertTrue(np.allclose([-15.5_54, -13.94_74]			,     __lowerCAmelCase			,     atol=1E-3 ) )
 def   _UpperCAmelCase       (							self ):
       UpperCamelCase_				:   Optional[int]					    =							self.get_feature_extractor()
       UpperCamelCase_				:   Union[str, Any]					    =							self.get_tokenizer()
       UpperCamelCase_				:   int					    =							self.get_decoder()

       UpperCamelCase_				:   Dict					    =							WavaVecaProcessorWithLM(tokenizer=__lowerCAmelCase			,     feature_extractor=__lowerCAmelCase			,     decoder=__lowerCAmelCase )

       UpperCamelCase_				:   str					    =							self._get_dummy_logits()

       UpperCamelCase_				:   Optional[int]					    =							2.0
       UpperCamelCase_				:   List[str]					    =							5.0
       UpperCamelCase_				:   Optional[Any]					    =							-20.0
       UpperCamelCase_				:   Optional[Any]					    =							True

       UpperCamelCase_				:   Union[str, Any]					    =							processor.batch_decode(
           __lowerCAmelCase			,     alpha=__lowerCAmelCase			,     beta=__lowerCAmelCase			,     unk_score_offset=__lowerCAmelCase			,     lm_score_boundary=__lowerCAmelCase			,     )
       UpperCamelCase_				:   List[str]					    =							decoded_processor_out.text

       UpperCamelCase_				:   List[str]					    =							list(__lowerCAmelCase )
       decoder.reset_params(
           alpha=__lowerCAmelCase			,     beta=__lowerCAmelCase			,     unk_score_offset=__lowerCAmelCase			,     lm_score_boundary=__lowerCAmelCase			,     )

       with get_context("""fork""" ).Pool() as pool:
             UpperCamelCase_				:   int					    =							decoder.decode_beams_batch(
                 __lowerCAmelCase			,     __lowerCAmelCase			,     )

       UpperCamelCase_				:   Any					    =							[d[0][0] for d in decoded_decoder_out]

       self.assertListEqual(__lowerCAmelCase			,     __lowerCAmelCase )
       self.assertListEqual(["""<s> </s> <s> </s> </s>""", """</s> </s> <s> </s> </s>"""]			,     __lowerCAmelCase )
       UpperCamelCase_				:   Union[str, Any]					    =							processor.decoder.model_container[processor.decoder._model_key]
       self.assertEqual(lm_model.alpha			,     2.0 )
       self.assertEqual(lm_model.beta			,     5.0 )
       self.assertEqual(lm_model.unk_score_offset			,     -20.0 )
       self.assertEqual(lm_model.score_boundary			,     __lowerCAmelCase )
 def   _UpperCAmelCase       (							self ):
       UpperCamelCase_				:   Dict					    =							WavaVecaProcessorWithLM.from_pretrained("""hf-internal-testing/processor_with_lm""" )

       UpperCamelCase_				:   Optional[int]					    =							processor.decoder.model_container[processor.decoder._model_key]
       UpperCamelCase_				:   int					    =							Path(language_model._kenlm_model.path.decode("""utf-8""" ) ).parent.parent.absolute()

       UpperCamelCase_				:   Any					    =							os.listdir(__lowerCAmelCase )
       UpperCamelCase_				:   Optional[Any]					    =							["""alphabet.json""", """language_model"""]

       downloaded_decoder_files.sort()
       expected_decoder_files.sort()

       # test that only decoder relevant files from
       # https://huggingface.co/hf-internal-testing/processor_with_lm/tree/main
       # are downloaded and none of the rest (e.g. README.md, ...)
       self.assertListEqual(__lowerCAmelCase			,     __lowerCAmelCase )
 def   _UpperCAmelCase       (							self ):
       UpperCamelCase_				:   Union[str, Any]					    =							snapshot_download("""hf-internal-testing/processor_with_lm""" )

       UpperCamelCase_				:   Union[str, Any]					    =							WavaVecaProcessorWithLM.from_pretrained(__lowerCAmelCase )

       UpperCamelCase_				:   Union[str, Any]					    =							processor.decoder.model_container[processor.decoder._model_key]
       UpperCamelCase_				:   Tuple					    =							Path(language_model._kenlm_model.path.decode("""utf-8""" ) ).parent.parent.absolute()

       UpperCamelCase_				:   Union[str, Any]					    =							os.listdir(__lowerCAmelCase )
       UpperCamelCase_				:   str					    =							os.listdir(__lowerCAmelCase )

       local_decoder_files.sort()
       expected_decoder_files.sort()

       # test that both decoder form hub and local files in cache are the same
       self.assertListEqual(__lowerCAmelCase			,     __lowerCAmelCase )
 def   _UpperCAmelCase       (							self ):
       UpperCamelCase_				:   Any					    =							WavaVecaProcessorWithLM.from_pretrained("""hf-internal-testing/processor_with_lm""" )
       UpperCamelCase_				:   Tuple					    =							AutoProcessor.from_pretrained("""hf-internal-testing/processor_with_lm""" )

       UpperCamelCase_				:   Dict					    =							floats_list((3, 10_00) )

       UpperCamelCase_				:   List[Any]					    =							processor_wavaveca(__lowerCAmelCase			,     return_tensors="""np""" )
       UpperCamelCase_				:   Tuple					    =							processor_auto(__lowerCAmelCase			,     return_tensors="""np""" )

       for key in input_wavaveca.keys():
             self.assertAlmostEqual(input_wavaveca[key].sum()			,     input_auto[key].sum()			,     delta=1E-2 )

       UpperCamelCase_				:   Optional[int]					    =							self._get_dummy_logits()

       UpperCamelCase_				:   Dict					    =							processor_wavaveca.batch_decode(__lowerCAmelCase )
       UpperCamelCase_				:   Any					    =							processor_auto.batch_decode(__lowerCAmelCase )

       self.assertListEqual(decoded_wavaveca.text			,     decoded_auto.text )
 def   _UpperCAmelCase       (							self ):
       UpperCamelCase_				:   Tuple					    =							self.get_feature_extractor()
       UpperCamelCase_				:   int					    =							self.get_tokenizer()
       UpperCamelCase_				:   List[Any]					    =							self.get_decoder()

       UpperCamelCase_				:   List[str]					    =							WavaVecaProcessorWithLM(tokenizer=__lowerCAmelCase			,     feature_extractor=__lowerCAmelCase			,     decoder=__lowerCAmelCase )

       self.assertListEqual(
           processor.model_input_names			,     feature_extractor.model_input_names			,     msg="""`processor` and `feature_extractor` model input names do not match"""			,     )
 @staticmethod
 def   _UpperCAmelCase       (							__lowerCAmelCase			,     __lowerCAmelCase ):
       UpperCamelCase_				:   Union[str, Any]					    =							[d[key] for d in offsets]
       return retrieved_list
 def   _UpperCAmelCase       (							self ):
       UpperCamelCase_				:   Dict					    =							WavaVecaProcessorWithLM.from_pretrained("""hf-internal-testing/processor_with_lm""" )
       UpperCamelCase_				:   int					    =							self._get_dummy_logits()[0]

       UpperCamelCase_				:   List[Any]					    =							processor.decode(__lowerCAmelCase			,     output_word_offsets=__lowerCAmelCase )
       # check Wav2Vec2CTCTokenizerOutput keys for word
       self.assertEqual(len(outputs.keys() )			,     4 )
       self.assertTrue("""text""" in outputs )
       self.assertTrue("""word_offsets""" in outputs )
       self.assertTrue(isinstance(__lowerCAmelCase			,     __lowerCAmelCase ) )

       self.assertEqual(""" """.join(self.get_from_offsets(outputs["""word_offsets"""]			,     """word""" ) )			,     outputs.text )
       self.assertListEqual(self.get_from_offsets(outputs["""word_offsets"""]			,     """word""" )			,     ["""<s>""", """<s>""", """</s>"""] )
       self.assertListEqual(self.get_from_offsets(outputs["""word_offsets"""]			,     """start_offset""" )			,     [0, 2, 4] )
       self.assertListEqual(self.get_from_offsets(outputs["""word_offsets"""]			,     """end_offset""" )			,     [1, 3, 5] )
 def   _UpperCAmelCase       (							self ):
       UpperCamelCase_				:   Any					    =							WavaVecaProcessorWithLM.from_pretrained("""hf-internal-testing/processor_with_lm""" )
       UpperCamelCase_				:   Union[str, Any]					    =							self._get_dummy_logits()

       UpperCamelCase_				:   Dict					    =							processor.batch_decode(__lowerCAmelCase			,     output_word_offsets=__lowerCAmelCase )

       # check Wav2Vec2CTCTokenizerOutput keys for word
       self.assertEqual(len(outputs.keys() )			,     4 )
       self.assertTrue("""text""" in outputs )
       self.assertTrue("""word_offsets""" in outputs )
       self.assertTrue(isinstance(__lowerCAmelCase			,     __lowerCAmelCase ) )

       self.assertListEqual(
           [""" """.join(self.get_from_offsets(__lowerCAmelCase			,     """word""" ) ) for o in outputs["""word_offsets"""]]			,     outputs.text )
       self.assertListEqual(self.get_from_offsets(outputs["""word_offsets"""][0]			,     """word""" )			,     ["""<s>""", """<s>""", """</s>"""] )
       self.assertListEqual(self.get_from_offsets(outputs["""word_offsets"""][0]			,     """start_offset""" )			,     [0, 2, 4] )
       self.assertListEqual(self.get_from_offsets(outputs["""word_offsets"""][0]			,     """end_offset""" )			,     [1, 3, 5] )





 @slow
 @require_torch
 @require_torchaudio
 def   _UpperCAmelCase       (							self ):
       import torch

       UpperCamelCase_				:   str					    =							load_dataset("""common_voice"""			,     """en"""			,     split="""train"""			,     streaming=__lowerCAmelCase )
       UpperCamelCase_				:   Union[str, Any]					    =							ds.cast_column("""audio"""			,     datasets.Audio(sampling_rate=1_60_00 ) )
       UpperCamelCase_				:   Union[str, Any]					    =							iter(__lowerCAmelCase )
       UpperCamelCase_				:   int					    =							next(__lowerCAmelCase )

       UpperCamelCase_				:   List[str]					    =							AutoProcessor.from_pretrained("""patrickvonplaten/wav2vec2-base-100h-with-lm""" )
       UpperCamelCase_				:   Tuple					    =							WavaVecaForCTC.from_pretrained("""patrickvonplaten/wav2vec2-base-100h-with-lm""" )

       # compare to filename `common_voice_en_100038.mp3` of dataset viewer on https://huggingface.co/datasets/common_voice/viewer/en/train
       UpperCamelCase_				:   Dict					    =							processor(sample["""audio"""]["""array"""]			,     return_tensors="""pt""" ).input_values

       with torch.no_grad():
             UpperCamelCase_				:   List[Any]					    =							model(__lowerCAmelCase ).logits.cpu().numpy()

       UpperCamelCase_				:   Tuple					    =							processor.decode(logits[0]			,     output_word_offsets=__lowerCAmelCase )

       UpperCamelCase_				:   List[Any]					    =							model.config.inputs_to_logits_ratio / processor.feature_extractor.sampling_rate
       UpperCamelCase_				:   Dict					    =							[
           {
               """start_time""": d["""start_offset"""] * time_offset,
               """end_time""": d["""end_offset"""] * time_offset,
               """word""": d["""word"""],
           }
           for d in output["""word_offsets"""]
       ]

       UpperCamelCase_				:   Optional[Any]					    =							"""WHY DOES MILISANDRA LOOK LIKE SHE WANTS TO CONSUME JOHN SNOW ON THE RIVER AT THE WALL"""

       # output words
       self.assertEqual(""" """.join(self.get_from_offsets(__lowerCAmelCase			,     """word""" ) )			,     __lowerCAmelCase )
       self.assertEqual(""" """.join(self.get_from_offsets(__lowerCAmelCase			,     """word""" ) )			,     output.text )

       # output times
       UpperCamelCase_				:   str					    =							torch.tensor(self.get_from_offsets(__lowerCAmelCase			,     """start_time""" ) )
       UpperCamelCase_				:   Union[str, Any]					    =							torch.tensor(self.get_from_offsets(__lowerCAmelCase			,     """end_time""" ) )

       # fmt: off
       UpperCamelCase_				:   Union[str, Any]					    =							torch.tensor([1.41_99, 1.65_99, 2.25_99, 3.0, 3.24, 3.59_99, 3.79_99, 4.09_99, 4.26, 4.94, 5.28, 5.65_99, 5.78, 5.94, 6.32, 6.53_99, 6.65_99] )
       UpperCamelCase_				:   Union[str, Any]					    =							torch.tensor([1.53_99, 1.89_99, 2.9, 3.16, 3.53_99, 3.72, 4.01_99, 4.17_99, 4.76, 5.15_99, 5.55_99, 5.69_99, 5.86, 6.19_99, 6.38, 6.61_99, 6.94] )
       # fmt: on

       self.assertTrue(torch.allclose(__lowerCAmelCase			,     __lowerCAmelCase			,     atol=0.01 ) )
       self.assertTrue(torch.allclose(__lowerCAmelCase			,     __lowerCAmelCase			,     atol=0.01 ) )






