





import warnings
from contextlib import contextmanager

from ...processing_utils import ProcessorMixin
from .feature_extraction_wavaveca import WavaVecaFeatureExtractor
from .tokenization_wavaveca import WavaVecaCTCTokenizer







class   __lowerCAmelCase			(					lowercase__  ):
   _UpperCamelCase :						Any     		=				"""Wav2Vec2FeatureExtractor"""
   _UpperCamelCase :						Dict     		=				"""AutoTokenizer"""





   def __init__( self					,			snake_case					,			snake_case      )				->  str:
      """simple docstring"""





      super().__init__(__lowerCamelCase					,			__lowerCamelCase      )
      a__	:						Any      		=  self.feature_extractor
      a__	:						List[Any]      		=  False





   @classmethod
   def 	_snake_case ( cls					,			snake_case					,			**snake_case      )				->  List[str]:
      """simple docstring"""





      try:
         return super().from_pretrained(__lowerCamelCase					,			**__lowerCamelCase      )
      except OSError:
         warnings.warn(
             F"""Loading a tokenizer inside {cls.__name__} from a config that does not"""
             " include a `tokenizer_class` attribute is deprecated and will be "
             "removed in v5. Please add `'tokenizer_class': 'Wav2Vec2CTCTokenizer'`"
             " attribute to either your `config.json` or `tokenizer_config.json` "
             "file to suppress this warning: "					,			__lowerCamelCase					,			)

         a__	:						List[Any]      		=  WavaVecaFeatureExtractor.from_pretrained(__lowerCamelCase					,			**__lowerCamelCase      )
         a__	:						Tuple      		=  WavaVecaCTCTokenizer.from_pretrained(__lowerCamelCase					,			**__lowerCamelCase      )

         return cls(feature_extractor=__lowerCamelCase					,			tokenizer=__lowerCamelCase      )





   def __call__( self					,			*snake_case					,			**snake_case      )				->  Any:
      """simple docstring"""





      if self._in_target_context_manager:
         return self.current_processor(*__lowerCamelCase					,			**__lowerCamelCase      )

      if "raw_speech" in kwargs:
         warnings.warn("Using `raw_speech` as a keyword argument is deprecated. Use `audio` instead."      )
         a__	:						Union[str, Any]      		=  kwargs.pop("raw_speech"      )
      else:
         a__	:						List[Any]      		=  kwargs.pop("audio"					,			__lowerCamelCase      )
      a__	:						Tuple      		=  kwargs.pop("sampling_rate"					,			__lowerCamelCase      )
      a__	:						Optional[Any]      		=  kwargs.pop("text"					,			__lowerCamelCase      )
      if len(__lowerCamelCase      ) > 0:
         a__	:						Any      		=  args[0]
         a__	:						List[str]      		=  args[1:]

      if audio is None and text is None:
         raise ValueError("You need to specify either an `audio` or `text` input to process."      )

      if audio is not None:
         a__	:						List[Any]      		=  self.feature_extractor(__lowerCamelCase					,			*__lowerCamelCase					,			sampling_rate=__lowerCamelCase					,			**__lowerCamelCase      )
      if text is not None:
         a__	:						Tuple      		=  self.tokenizer(__lowerCamelCase					,			**__lowerCamelCase      )

      if text is None:
         return inputs
      elif audio is None:
         return encodings
      else:
         a__	:						Any      		=  encodings["input_ids"]
         return inputs





   def 	_snake_case ( self					,			*snake_case					,			**snake_case      )				->  List[str]:
      """simple docstring"""





      if self._in_target_context_manager:
         return self.current_processor.pad(*__lowerCamelCase					,			**__lowerCamelCase      )

      a__	:						Optional[int]      		=  kwargs.pop("input_features"					,			__lowerCamelCase      )
      a__	:						List[str]      		=  kwargs.pop("labels"					,			__lowerCamelCase      )
      if len(__lowerCamelCase      ) > 0:
         a__	:						Dict      		=  args[0]
         a__	:						List[str]      		=  args[1:]

      if input_features is not None:
         a__	:						Tuple      		=  self.feature_extractor.pad(__lowerCamelCase					,			*__lowerCamelCase					,			**__lowerCamelCase      )
      if labels is not None:
         a__	:						Dict      		=  self.tokenizer.pad(__lowerCamelCase					,			**__lowerCamelCase      )

      if labels is None:
         return input_features
      elif input_features is None:
         return labels
      else:
         a__	:						Any      		=  labels["input_ids"]
         return input_features





   def 	_snake_case ( self					,			*snake_case					,			**snake_case      )				->  List[Any]:
      """simple docstring"""





      return self.tokenizer.batch_decode(*__lowerCamelCase					,			**__lowerCamelCase      )





   def 	_snake_case ( self					,			*snake_case					,			**snake_case      )				->  List[Any]:
      """simple docstring"""





      return self.tokenizer.decode(*__lowerCamelCase					,			**__lowerCamelCase      )





   @contextmanager
   def 	_snake_case ( self      )				->  Dict:
      """simple docstring"""





      warnings.warn(
          "`as_target_processor` is deprecated and will be removed in v5 of Transformers. You can process your "
          "labels by using the argument `text` of the regular `__call__` method (either in the same call as "
          "your audio inputs, or in a separate call."      )
      a__	:						Tuple      		=  True
      a__	:						Optional[int]      		=  self.tokenizer
      yield
      a__	:						Tuple      		=  self.feature_extractor
      a__	:						Optional[int]      		=  False








import subprocess
import sys

from transformers import BertConfig, BertModel, BertTokenizer, pipeline
from transformers.testing_utils import TestCasePlus, require_torch







class   __lowerCAmelCase			(					_UpperCamelCase  ):





   @require_torch
   def 	_snake_case ( self      )				->  str:
      """simple docstring"""





      a__	:						Tuple      		=  "\nfrom transformers import BertConfig, BertModel, BertTokenizer, pipeline\n        "

      a__	:						str      		=  "\nmname = \"hf-internal-testing/tiny-random-bert\"\nBertConfig.from_pretrained(mname)\nBertModel.from_pretrained(mname)\nBertTokenizer.from_pretrained(mname)\npipe = pipeline(task=\"fill-mask\", model=mname)\nprint(\"success\")\n        "

      a__	:						Optional[Any]      		=  "\nimport socket\ndef offline_socket(*args, **kwargs): raise RuntimeError(\"Offline mode is enabled, we shouldn't access internet\")\nsocket.socket = offline_socket\n        "

      # Force fetching the files so that we can use the cache
      a__	:						Tuple      		=  "hf-internal-testing/tiny-random-bert"
      BertConfig.from_pretrained(snake_case      )
      BertModel.from_pretrained(snake_case      )
      BertTokenizer.from_pretrained(snake_case      )
      pipeline(task="fill-mask"					,			model=snake_case      )

      # baseline - just load from_pretrained with normal network
      a__	:						Optional[int]      		=  [sys.executable, "-c", "\n".join([load, run, mock]      )]

      # should succeed
      a__	:						Dict      		=  self.get_env()
      # should succeed as TRANSFORMERS_OFFLINE=1 tells it to use local files
      a__	:						Tuple      		=  "1"
      a__	:						List[Any]      		=  subprocess.run(snake_case					,			env=snake_case					,			check=snake_case					,			capture_output=snake_case      )
      self.assertEqual(result.returncode					,			0					,			result.stderr      )
      self.assertIn("success"					,			result.stdout.decode()      )





   @require_torch
   def 	_snake_case ( self      )				->  List[Any]:
      """simple docstring"""





      a__	:						Tuple      		=  "\nfrom transformers import BertConfig, BertModel, BertTokenizer, pipeline\n        "

      a__	:						Optional[Any]      		=  "\nmname = \"hf-internal-testing/tiny-random-bert\"\nBertConfig.from_pretrained(mname)\nBertModel.from_pretrained(mname)\nBertTokenizer.from_pretrained(mname)\npipe = pipeline(task=\"fill-mask\", model=mname)\nprint(\"success\")\n        "

      a__	:						Tuple      		=  "\nimport socket\ndef offline_socket(*args, **kwargs): raise socket.error(\"Faking flaky internet\")\nsocket.socket = offline_socket\n        "

      # Force fetching the files so that we can use the cache
      a__	:						List[Any]      		=  "hf-internal-testing/tiny-random-bert"
      BertConfig.from_pretrained(snake_case      )
      BertModel.from_pretrained(snake_case      )
      BertTokenizer.from_pretrained(snake_case      )
      pipeline(task="fill-mask"					,			model=snake_case      )

      # baseline - just load from_pretrained with normal network
      a__	:						Tuple      		=  [sys.executable, "-c", "\n".join([load, run, mock]      )]

      # should succeed
      a__	:						Any      		=  self.get_env()
      a__	:						int      		=  subprocess.run(snake_case					,			env=snake_case					,			check=snake_case					,			capture_output=snake_case      )
      self.assertEqual(result.returncode					,			0					,			result.stderr      )
      self.assertIn("success"					,			result.stdout.decode()      )





   @require_torch
   def 	_snake_case ( self      )				->  Tuple:
      """simple docstring"""





      a__	:						List[Any]      		=  "\nfrom transformers import BertConfig, BertModel, BertTokenizer\n        "

      a__	:						Dict      		=  "\nmname = \"hf-internal-testing/tiny-random-bert-sharded\"\nBertConfig.from_pretrained(mname)\nBertModel.from_pretrained(mname)\nprint(\"success\")\n        "

      a__	:						int      		=  "\nimport socket\ndef offline_socket(*args, **kwargs): raise ValueError(\"Offline mode is enabled\")\nsocket.socket = offline_socket\n        "

      # baseline - just load from_pretrained with normal network
      a__	:						List[Any]      		=  [sys.executable, "-c", "\n".join([load, run]      )]

      # should succeed
      a__	:						str      		=  self.get_env()
      a__	:						List[str]      		=  subprocess.run(snake_case					,			env=snake_case					,			check=snake_case					,			capture_output=snake_case      )
      self.assertEqual(result.returncode					,			0					,			result.stderr      )
      self.assertIn("success"					,			result.stdout.decode()      )

      # next emulate no network
      a__	:						List[Any]      		=  [sys.executable, "-c", "\n".join([load, mock, run]      )]

      # Doesn't fail anymore since the model is in the cache due to other tests, so commenting this.
      # env["TRANSFORMERS_OFFLINE"] = "0"
      # result = subprocess.run(cmd, env=env, check=False, capture_output=True)
      # self.assertEqual(result.returncode, 1, result.stderr)

      # should succeed as TRANSFORMERS_OFFLINE=1 tells it to use local files
      a__	:						Union[str, Any]      		=  "1"
      a__	:						Tuple      		=  subprocess.run(snake_case					,			env=snake_case					,			check=snake_case					,			capture_output=snake_case      )
      self.assertEqual(result.returncode					,			0					,			result.stderr      )
      self.assertIn("success"					,			result.stdout.decode()      )





   @require_torch
   def 	_snake_case ( self      )				->  Tuple:
      """simple docstring"""





      a__	:						Optional[Any]      		=  "\nfrom transformers import pipeline\n        "
      a__	:						int      		=  "\nmname = \"hf-internal-testing/tiny-random-bert\"\npipe = pipeline(model=mname)\n        "

      a__	:						Dict      		=  "\nimport socket\ndef offline_socket(*args, **kwargs): raise socket.error(\"Offline mode is enabled\")\nsocket.socket = offline_socket\n        "
      a__	:						List[str]      		=  self.get_env()
      a__	:						Union[str, Any]      		=  "1"
      a__	:						List[str]      		=  [sys.executable, "-c", "\n".join([load, mock, run]      )]
      a__	:						Any      		=  subprocess.run(snake_case					,			env=snake_case					,			check=snake_case					,			capture_output=snake_case      )
      self.assertEqual(result.returncode					,			1					,			result.stderr      )
      self.assertIn(
          "You cannot infer task automatically within `pipeline` when using offline mode"					,			result.stderr.decode().replace("\n"					,			""      )					,			)





   @require_torch
   def 	_snake_case ( self      )				->  Union[str, Any]:
      """simple docstring"""





      a__	:						Any      		=  "\nfrom transformers import AutoModel\n        "
      a__	:						Any      		=  "\nmname = \"hf-internal-testing/test_dynamic_model\"\nAutoModel.from_pretrained(mname, trust_remote_code=True)\nprint(\"success\")\n        "

      # baseline - just load from_pretrained with normal network
      a__	:						List[str]      		=  [sys.executable, "-c", "\n".join([load, run]      )]

      # should succeed
      a__	:						Optional[Any]      		=  self.get_env()
      a__	:						Union[str, Any]      		=  subprocess.run(snake_case					,			env=snake_case					,			check=snake_case					,			capture_output=snake_case      )
      self.assertEqual(result.returncode					,			0					,			result.stderr      )
      self.assertIn("success"					,			result.stdout.decode()      )

      # should succeed as TRANSFORMERS_OFFLINE=1 tells it to use local files
      a__	:						Dict      		=  "1"
      a__	:						Union[str, Any]      		=  subprocess.run(snake_case					,			env=snake_case					,			check=snake_case					,			capture_output=snake_case      )
      self.assertEqual(result.returncode					,			0					,			result.stderr      )
      self.assertIn("success"					,			result.stdout.decode()      )

