

import json
import sys
import tempfile
import unittest
from pathlib import Path

import transformers
from transformers import (
    CONFIG_MAPPING,
    FEATURE_EXTRACTOR_MAPPING,
    AutoConfig,
    AutoFeatureExtractor,
    WavaVecaConfig,
    WavaVecaFeatureExtractor,
)
from transformers.testing_utils import DUMMY_UNKNOWN_IDENTIFIER, get_tests_dir


sys.path.append(str(Path(__file__).parent.parent.parent.parent / '''utils'''))

from test_module.custom_configuration import CustomConfig  # noqa E402
from test_module.custom_feature_extraction import CustomFeatureExtractor  # noqa E402


_SCREAMING_SNAKE_CASE					:  Optional[int]             =      get_tests_dir('''fixtures''')
_SCREAMING_SNAKE_CASE					:  int             =      get_tests_dir('''fixtures/dummy_feature_extractor_config.json''')
_SCREAMING_SNAKE_CASE					:  Tuple             =      get_tests_dir('''fixtures/dummy-config.json''')






class 	UpperCAmelCase__		(	unittest.TestCase	):






  """simple docstring"""


  def 				lowercase_		(							self						:		Tuple )		->							Optional[int]:
       SCREAMING_SNAKE_CASE__   =   0






  def 				lowercase_		(							self						:		str )		->							Any:
       SCREAMING_SNAKE_CASE__   =   AutoFeatureExtractor.from_pretrained('''facebook/wav2vec2-base-960h''' )
       self.assertIsInstance(__lowerCamelCase     ,       __lowerCamelCase )






  def 				lowercase_		(							self						:		Any )		->							int:
       SCREAMING_SNAKE_CASE__   =   AutoFeatureExtractor.from_pretrained(__lowerCamelCase )
       self.assertIsInstance(__lowerCamelCase     ,       __lowerCamelCase )






  def 				lowercase_		(							self						:		Tuple )		->							int:
       with tempfile.TemporaryDirectory() as tmpdirname:
            SCREAMING_SNAKE_CASE__   =   WavaVecaConfig()

            # remove feature_extractor_type to make sure config.json alone is enough to load feature processor locally
            SCREAMING_SNAKE_CASE__   =   AutoFeatureExtractor.from_pretrained(__lowerCamelCase ).to_dict()

            config_dict.pop('''feature_extractor_type''' )
            SCREAMING_SNAKE_CASE__   =   WavaVecaFeatureExtractor(**__lowerCamelCase )

            # save in new folder
            model_config.save_pretrained(__lowerCamelCase )
            config.save_pretrained(__lowerCamelCase )

            SCREAMING_SNAKE_CASE__   =   AutoFeatureExtractor.from_pretrained(__lowerCamelCase )

            # make sure private variable is not incorrectly saved
            SCREAMING_SNAKE_CASE__   =   json.loads(config.to_json_string() )
            self.assertTrue('''_processor_class''' not in dict_as_saved )

       self.assertIsInstance(__lowerCamelCase     ,       __lowerCamelCase )






  def 				lowercase_		(							self						:		Optional[int] )		->							Union[str, Any]:
       SCREAMING_SNAKE_CASE__   =   AutoFeatureExtractor.from_pretrained(__lowerCamelCase )
       self.assertIsInstance(__lowerCamelCase     ,       __lowerCamelCase )






  def 				lowercase_		(							self						:		List[str] )		->							Tuple:
       with self.assertRaisesRegex(
           __lowerCamelCase     ,       '''bert-base is not a local folder and is not a valid model identifier''' ):
            SCREAMING_SNAKE_CASE__   =   AutoFeatureExtractor.from_pretrained('''bert-base''' )






  def 				lowercase_		(							self						:		List[str] )		->							int:
       with self.assertRaisesRegex(
           __lowerCamelCase     ,       r'''aaaaaa is not a valid git identifier \(branch name, tag name or commit id\)''' ):
            SCREAMING_SNAKE_CASE__   =   AutoFeatureExtractor.from_pretrained(__lowerCamelCase     ,       revision='''aaaaaa''' )






  def 				lowercase_		(							self						:		Dict )		->							Dict:
       with self.assertRaisesRegex(
           __lowerCamelCase     ,       '''hf-internal-testing/config-no-model does not appear to have a file named preprocessor_config.json.'''     ,       ):
            SCREAMING_SNAKE_CASE__   =   AutoFeatureExtractor.from_pretrained('''hf-internal-testing/config-no-model''' )






  def 				lowercase_		(							self						:		List[Any] )		->							List[str]:
       # If remote code is not set, we will time out when asking whether to load the model.
       with self.assertRaises(__lowerCamelCase ):
            SCREAMING_SNAKE_CASE__   =   AutoFeatureExtractor.from_pretrained(
                '''hf-internal-testing/test_dynamic_feature_extractor''' )
       # If remote code is disabled, we can't load this config.
       with self.assertRaises(__lowerCamelCase ):
            SCREAMING_SNAKE_CASE__   =   AutoFeatureExtractor.from_pretrained(
                '''hf-internal-testing/test_dynamic_feature_extractor'''     ,       trust_remote_code=__lowerCamelCase )

       SCREAMING_SNAKE_CASE__   =   AutoFeatureExtractor.from_pretrained(
           '''hf-internal-testing/test_dynamic_feature_extractor'''     ,       trust_remote_code=__lowerCamelCase )
       self.assertEqual(feature_extractor.__class__.__name__     ,       '''NewFeatureExtractor''' )

       # Test feature extractor can be reloaded.
       with tempfile.TemporaryDirectory() as tmp_dir:
            feature_extractor.save_pretrained(__lowerCamelCase )
            SCREAMING_SNAKE_CASE__   =   AutoFeatureExtractor.from_pretrained(__lowerCamelCase     ,       trust_remote_code=__lowerCamelCase )
       self.assertEqual(reloaded_feature_extractor.__class__.__name__     ,       '''NewFeatureExtractor''' )






  def 				lowercase_		(							self						:		List[Any] )		->							Tuple:
       try:
            AutoConfig.register('''custom'''     ,       __lowerCamelCase )
            AutoFeatureExtractor.register(__lowerCamelCase     ,       __lowerCamelCase )
            # Trying to register something existing in the Transformers library will raise an error
            with self.assertRaises(__lowerCamelCase ):
                 AutoFeatureExtractor.register(__lowerCamelCase     ,       __lowerCamelCase )

            # Now that the config is registered, it can be used as any other config with the auto-API
            SCREAMING_SNAKE_CASE__   =   CustomFeatureExtractor.from_pretrained(__lowerCamelCase )
            with tempfile.TemporaryDirectory() as tmp_dir:
                 feature_extractor.save_pretrained(__lowerCamelCase )
                 SCREAMING_SNAKE_CASE__   =   AutoFeatureExtractor.from_pretrained(__lowerCamelCase )
                 self.assertIsInstance(__lowerCamelCase     ,       __lowerCamelCase )

       finally:
            if "custom" in CONFIG_MAPPING._extra_content:
                 del CONFIG_MAPPING._extra_content["custom"]
            if CustomConfig in FEATURE_EXTRACTOR_MAPPING._extra_content:
                 del FEATURE_EXTRACTOR_MAPPING._extra_content[CustomConfig]







  def 				lowercase_		(							self						:		Optional[Any] )		->							int:






       class 	UpperCAmelCase__		(	A__	):






         """simple docstring"""


         a									= True

       try:
            AutoConfig.register('''custom'''     ,       __lowerCamelCase )
            AutoFeatureExtractor.register(__lowerCamelCase     ,       __lowerCamelCase )
            # If remote code is not set, the default is to use local
            SCREAMING_SNAKE_CASE__   =   AutoFeatureExtractor.from_pretrained(
                '''hf-internal-testing/test_dynamic_feature_extractor''' )
            self.assertEqual(feature_extractor.__class__.__name__     ,       '''NewFeatureExtractor''' )
            self.assertTrue(feature_extractor.is_local )

            # If remote code is disabled, we load the local one.
            SCREAMING_SNAKE_CASE__   =   AutoFeatureExtractor.from_pretrained(
                '''hf-internal-testing/test_dynamic_feature_extractor'''     ,       trust_remote_code=__lowerCamelCase )
            self.assertEqual(feature_extractor.__class__.__name__     ,       '''NewFeatureExtractor''' )
            self.assertTrue(feature_extractor.is_local )

            # If remote is enabled, we load from the Hub
            SCREAMING_SNAKE_CASE__   =   AutoFeatureExtractor.from_pretrained(
                '''hf-internal-testing/test_dynamic_feature_extractor'''     ,       trust_remote_code=__lowerCamelCase )
            self.assertEqual(feature_extractor.__class__.__name__     ,       '''NewFeatureExtractor''' )
            self.assertTrue(not hasattr(__lowerCamelCase     ,       '''is_local''' ) )

       finally:
            if "custom" in CONFIG_MAPPING._extra_content:
                 del CONFIG_MAPPING._extra_content["custom"]
            if CustomConfig in FEATURE_EXTRACTOR_MAPPING._extra_content:
                 del FEATURE_EXTRACTOR_MAPPING._extra_content[CustomConfig]





import argparse
import re
from typing import Dict

import torch
from datasets import Audio, Dataset, load_dataset, load_metric

from transformers import AutoFeatureExtractor, pipeline
def     UpperCAmelCase_       (						_A				,		_A    ):





     '''simple docstring'''




     SCREAMING_SNAKE_CASE__   =   args.log_outputs
     SCREAMING_SNAKE_CASE__   =   '''_'''.join(args.dataset.split('''/'''    ) + [args.config, args.split]    )

     # load metric
     SCREAMING_SNAKE_CASE__   =   load_metric('''wer'''    )
     SCREAMING_SNAKE_CASE__   =   load_metric('''cer'''    )

     # compute metrics
     SCREAMING_SNAKE_CASE__   =   wer.compute(references=result['''target''']				,		predictions=result['''prediction''']    )
     SCREAMING_SNAKE_CASE__   =   cer.compute(references=result['''target''']				,		predictions=result['''prediction''']    )

     # print & log results
     SCREAMING_SNAKE_CASE__   =   F'''WER: {wer_result}\nCER: {cer_result}'''
     print(_A    )

     with open(F'''{dataset_id}_eval_results.txt'''				,		'''w'''    ) as f:
          f.write(_A    )

     # log all results in text file. Possibly interesting for analysis
     if log_outputs is not None:
          SCREAMING_SNAKE_CASE__   =   F'''log_{dataset_id}_predictions.txt'''
          SCREAMING_SNAKE_CASE__   =   F'''log_{dataset_id}_targets.txt'''

          with open(_A				,		'''w'''    ) as p, open(_A				,		'''w'''    ) as t:
               # mapping function to write output
               def write_to_file(_A				,		_A    ):
                    p.write(F'''{i}''' + '''\n'''    )
                    p.write(batch['''prediction'''] + '''\n'''    )
                    t.write(F'''{i}''' + '''\n'''    )
                    t.write(batch['''target'''] + '''\n'''    )

               result.map(_A				,		with_indices=_A    )
def     UpperCAmelCase_       (						_A    ):





     '''simple docstring'''




     SCREAMING_SNAKE_CASE__   =   '''[,?.!\-\;\:"“%‘”�—’…–]'''  # noqa: W605 IMPORTANT: this should correspond to the chars that were ignored during training

     SCREAMING_SNAKE_CASE__   =   re.sub(_A				,		''''''				,		text.lower()    )

     # In addition, we can normalize the target text, e.g. removing new lines characters etc...
     # note that order is important here!
     SCREAMING_SNAKE_CASE__   =   ['''\n\n''', '''\n''', '''   ''', '''  ''']

     for t in token_sequences_to_ignore:
          SCREAMING_SNAKE_CASE__   =   ''' '''.join(text.split(_A    )    )

     return text
def     UpperCAmelCase_       (						_A    ):





     '''simple docstring'''




     SCREAMING_SNAKE_CASE__   =   load_dataset(args.dataset				,		args.config				,		split=args.split				,		use_auth_token=_A    )

     # for testing: only process the first two examples as a test
     # dataset = dataset.select(range(10))

     # load processor
     SCREAMING_SNAKE_CASE__   =   AutoFeatureExtractor.from_pretrained(args.model_id    )
     SCREAMING_SNAKE_CASE__   =   feature_extractor.sampling_rate

     # resample audio
     SCREAMING_SNAKE_CASE__   =   dataset.cast_column('''audio'''				,		Audio(sampling_rate=_A    )    )

     # load eval pipeline
     if args.device is None:
          SCREAMING_SNAKE_CASE__   =   0 if torch.cuda.is_available() else -1
     SCREAMING_SNAKE_CASE__   =   pipeline('''automatic-speech-recognition'''				,		model=args.model_id				,		device=args.device    )

     # map function to decode audio
     def map_to_pred(_A    ):
          SCREAMING_SNAKE_CASE__   =   asr(
              batch['''audio''']['''array''']				,		chunk_length_s=args.chunk_length_s				,		stride_length_s=args.stride_length_s    )

          SCREAMING_SNAKE_CASE__   =   prediction['''text''']
          SCREAMING_SNAKE_CASE__   =   normalize_text(batch['''sentence''']    )
          return batch

     # run inference on all examples
     SCREAMING_SNAKE_CASE__   =   dataset.map(_A				,		remove_columns=dataset.column_names    )

     # compute and log_results
     # do not change function below
     log_results(_A				,		_A    )


if __name__ == "__main__":
       _SCREAMING_SNAKE_CASE					:  List[str]             =      argparse.ArgumentParser()

       parser.add_argument(
           '''--model_id''', type=str, required=True, help='''Model identifier. Should be loadable with 🤗 Transformers'''
       )
       parser.add_argument(
           '''--dataset''',
           type=str,
           required=True,
           help='''Dataset name to evaluate the `model_id`. Should be loadable with 🤗 Datasets''',
       )
       parser.add_argument(
           '''--config''', type=str, required=True, help='''Config of the dataset. *E.g.* `\'en\'`  for Common Voice'''
       )
       parser.add_argument('''--split''', type=str, required=True, help='''Split of the dataset. *E.g.* `\'test\'`''')
       parser.add_argument(
           '''--chunk_length_s''', type=float, default=None, help='''Chunk length in seconds. Defaults to 5 seconds.'''
       )
       parser.add_argument(
           '''--stride_length_s''', type=float, default=None, help='''Stride of the audio chunks. Defaults to 1 second.'''
       )
       parser.add_argument(
           '''--log_outputs''', action='''store_true''', help='''If defined, write outputs to log file for analysis.'''
       )
       parser.add_argument(
           '''--device''',
           type=int,
           default=None,
           help='''The device to run the pipeline on. -1 for CPU (default), 0 for the first GPU and so on.''',
       )
       _SCREAMING_SNAKE_CASE					:  List[Any]             =      parser.parse_args()

       main(args)


