





import unittest
from pathlib import Path
from tempfile import NamedTemporaryFile, TemporaryDirectory

from transformers import BertConfig, BertTokenizerFast, FeatureExtractionPipeline
from transformers.convert_graph_to_onnx import (
    convert,
    ensure_valid_input,
    generate_identified_filename,
    infer_shapes,
    quantize,
)
from transformers.testing_utils import require_tf, require_tokenizers, require_torch, slow







class 				__lowerCAmelCase       :







      def    _lowercase  (    self		,      lowerCAmelCase__		,      lowerCAmelCase__		,      lowerCAmelCase__							)   ->						List[str]:



        '''simple docstring'''




        return None







class 				__lowerCAmelCase       :







      def    _lowercase  (    self		,      lowerCAmelCase__		,      lowerCAmelCase__		,      lowerCAmelCase__		,      lowerCAmelCase__							)   ->						Optional[Any]:



        '''simple docstring'''




        return None







class 				__lowerCAmelCase       (    unittest.TestCase):
      _lowercase				:					Any						=			[
          # (model_name, model_kwargs)
          ("""bert-base-cased""", {}),
          ("""gpt2""", {"""use_cache""": False}),  # We don't support exporting GPT2 past keys anymore
      ]







      @require_tf
      @slow
      def    _lowercase  (    self							)   ->						Dict:



        '''simple docstring'''




        for model, model_kwargs in OnnxExportTestCase.MODEL_TO_TEST:
          self._test_export(lowerCAmelCase__		,      "tf"		,      1_2		,      **lowerCAmelCase__							)







      @require_torch
      @slow
      def    _lowercase  (    self							)   ->						str:



        '''simple docstring'''




        for model, model_kwargs in OnnxExportTestCase.MODEL_TO_TEST:
          self._test_export(lowerCAmelCase__		,      "pt"		,      1_2		,      **lowerCAmelCase__							)







      @require_torch
      @slow
      def    _lowercase  (    self							)   ->						List[Any]:



        '''simple docstring'''




        from transformers import BertModel

        a__	:					Union[str, Any]										=["[UNK]", "[SEP]", "[CLS]", "[PAD]", "[MASK]", "some", "other", "words"]
        with NamedTemporaryFile(mode="w+t"							) as vocab_file:
          vocab_file.write("\n".join(lowerCAmelCase__							)							)
          vocab_file.flush()
          a__	:					Union[str, Any]										=BertTokenizerFast(vocab_file.name							)

        with TemporaryDirectory() as bert_save_dir:
          a__	:					Any										=BertModel(BertConfig(vocab_size=len(lowerCAmelCase__							)							)							)
          model.save_pretrained(lowerCAmelCase__							)
          self._test_export(lowerCAmelCase__		,      "pt"		,      1_2		,      lowerCAmelCase__							)







      @require_tf
      @slow
      def    _lowercase  (    self							)   ->						Optional[int]:



        '''simple docstring'''




        for model, model_kwargs in OnnxExportTestCase.MODEL_TO_TEST:
          a__	:					Optional[int]										=self._test_export(lowerCAmelCase__		,      "tf"		,      1_2		,      **lowerCAmelCase__							)
          a__	:					List[str]										=quantize(Path(lowerCAmelCase__							)							)

          # Ensure the actual quantized model is not bigger than the original one
          if quantized_path.stat().st_size >= Path(lowerCAmelCase__							).stat().st_size:
            self.fail("Quantized model is bigger than initial ONNX model"							)







      @require_torch
      @slow
      def    _lowercase  (    self							)   ->						Any:



        '''simple docstring'''




        for model, model_kwargs in OnnxExportTestCase.MODEL_TO_TEST:
          a__	:					List[str]										=self._test_export(lowerCAmelCase__		,      "pt"		,      1_2		,      **lowerCAmelCase__							)
          a__	:					List[Any]										=quantize(lowerCAmelCase__							)

          # Ensure the actual quantized model is not bigger than the original one
          if quantized_path.stat().st_size >= Path(lowerCAmelCase__							).stat().st_size:
            self.fail("Quantized model is bigger than initial ONNX model"							)







      def    _lowercase  (    self		,      lowerCAmelCase__		,      lowerCAmelCase__		,      lowerCAmelCase__		,      lowerCAmelCase__=None		,      **lowerCAmelCase__							)   ->						str:



        '''simple docstring'''




        try:
          # Compute path
          with TemporaryDirectory() as tempdir:
            a__	:					Any										=Path(lowerCAmelCase__							).joinpath("model.onnx"							)

          # Remove folder if exists
          if path.parent.exists():
            path.parent.rmdir()

          # Export
          convert(lowerCAmelCase__		,      lowerCAmelCase__		,      lowerCAmelCase__		,      lowerCAmelCase__		,      lowerCAmelCase__		,      **lowerCAmelCase__							)

          return path
        except Exception as e:
          self.fail(lowerCAmelCase__							)







      @require_torch
      @require_tokenizers
      @slow
      def    _lowercase  (    self							)   ->						int:



        '''simple docstring'''




        from transformers import BertModel

        a__	:					List[Any]										=BertModel(BertConfig.from_pretrained("lysandre/tiny-bert-random"							)							)
        a__	:					int										=BertTokenizerFast.from_pretrained("lysandre/tiny-bert-random"							)
        self._test_infer_dynamic_axis(lowerCAmelCase__		,      lowerCAmelCase__		,      "pt"							)







      @require_tf
      @require_tokenizers
      @slow
      def    _lowercase  (    self							)   ->						Optional[Any]:



        '''simple docstring'''




        from transformers import TFBertModel

        a__	:					Any										=TFBertModel(BertConfig.from_pretrained("lysandre/tiny-bert-random"							)							)
        a__	:					str										=BertTokenizerFast.from_pretrained("lysandre/tiny-bert-random"							)
        self._test_infer_dynamic_axis(lowerCAmelCase__		,      lowerCAmelCase__		,      "tf"							)







      def    _lowercase  (    self		,      lowerCAmelCase__		,      lowerCAmelCase__		,      lowerCAmelCase__							)   ->						Tuple:



        '''simple docstring'''




        a__	:					int										=FeatureExtractionPipeline(lowerCAmelCase__		,      lowerCAmelCase__							)

        a__	:					int										=["input_ids", "token_type_ids", "attention_mask", "output_0", "output_1"]
        a__					,							a__					,							a__					,							a__	:					str										=infer_shapes(lowerCAmelCase__		,      lowerCAmelCase__							)

        # Assert all variables are present
        self.assertEqual(len(lowerCAmelCase__							)		,      len(lowerCAmelCase__							)							)
        self.assertTrue(all(var_name in shapes for var_name in variable_names							)							)
        self.assertSequenceEqual(variable_names[:3]		,      lowerCAmelCase__							)
        self.assertSequenceEqual(variable_names[3:]		,      lowerCAmelCase__							)

        # Assert inputs are {0: batch, 1: sequence}
        for var_name in ["input_ids", "token_type_ids", "attention_mask"]:
          self.assertDictEqual(shapes[var_name]		,      {0: "batch", 1: "sequence"}							)

        # Assert outputs are {0: batch, 1: sequence} and {0: batch}
        self.assertDictEqual(shapes["output_0"]		,      {0: "batch", 1: "sequence"}							)
        self.assertDictEqual(shapes["output_1"]		,      {0: "batch"}							)







      def    _lowercase  (    self							)   ->						Dict:



        '''simple docstring'''




        a__	:					int										=["input_ids", "attention_mask", "token_type_ids"]
        a__	:					Any										={"input_ids": [1, 2, 3, 4], "attention_mask": [0, 0, 0, 0], "token_type_ids": [1, 1, 1, 1]}
        a__					,							a__	:					Tuple										=ensure_valid_input(FuncContiguousArgs()		,      lowerCAmelCase__		,      lowerCAmelCase__							)

        # Should have exactly the same number of args (all are valid)
        self.assertEqual(len(lowerCAmelCase__							)		,      3							)

        # Should have exactly the same input names
        self.assertEqual(set(lowerCAmelCase__							)		,      set(lowerCAmelCase__							)							)

        # Parameter should be reordered according to their respective place in the function:
        # (input_ids, token_type_ids, attention_mask)
        self.assertEqual(lowerCAmelCase__		,      (tokens["input_ids"], tokens["token_type_ids"], tokens["attention_mask"])							)

        # Generated args are interleaved with another args (for instance parameter "past" in GPT2)
        a__					,							a__	:					Any										=ensure_valid_input(FuncNonContiguousArgs()		,      lowerCAmelCase__		,      lowerCAmelCase__							)

        # Should have exactly the one arg (all before the one not provided "some_other_args")
        self.assertEqual(len(lowerCAmelCase__							)		,      1							)
        self.assertEqual(len(lowerCAmelCase__							)		,      1							)

        # Should have only "input_ids"
        self.assertEqual(inputs_args[0]		,      tokens["input_ids"]							)
        self.assertEqual(ordered_input_names[0]		,      "input_ids"							)







      def    _lowercase  (    self							)   ->						Tuple:



        '''simple docstring'''




        a__	:					List[Any]										=generate_identified_filename(Path("/home/something/my_fake_model.onnx"							)		,      "-test"							)
        self.assertEqual("/home/something/my_fake_model-test.onnx"		,      generated.as_posix()							)











from collections import UserDict
from typing import Union

import numpy as np
import requests

from ..utils import (
    add_end_docstrings,
    logging,
)
from .audio_classification import ffmpeg_read
from .base import PIPELINE_INIT_ARGS, Pipeline


a_   :List[Any]        =      logging.get_logger(__name__)




@add_end_docstrings(lowerCAmelCase_     )
class 					snake_case__						(       lowerCAmelCase_     ):






     """simple docstring"""


     def __init__(       self  :  Optional[Any], **_snake_case  :  str					)   ->Dict:
            super().__init__(**_snake_case					)

            if self.framework != "pt":
                   raise ValueError(F'''The {self.__class__} is only available in PyTorch.'''					)
            # No specific FOR_XXX available yet






     def __call__(       self  :  Union[str, Any], _snake_case  :  Union[np.ndarray, bytes, str], **_snake_case  :  Tuple					)   ->Dict:
            return super().__call__(_snake_case, **_snake_case					)






     def    lowercase_   (       self  :  Tuple, **_snake_case  :  Any					)   ->Union[str, Any]:
            snake_case__							:	str   	=	{}
            if "candidate_labels" in kwargs:
                   snake_case__							:	str   	=	kwargs['candidate_labels']
            if "hypothesis_template" in kwargs:
                   snake_case__							:	str   	=	kwargs['hypothesis_template']

            return preprocess_params, {}, {}






     def    lowercase_   (       self  :  Dict, _snake_case  :  str, _snake_case  :  Optional[int]=None, _snake_case  :  List[str]="This is a sound of {}."					)   ->int:
            if isinstance(_snake_case, _snake_case					):
                   if audio.startswith('http://'					) or audio.startswith('https://'					):
                          # We need to actually check for a real protocol, otherwise it's impossible to use a local file
                          # like http_huggingface_co.png
                          snake_case__							:	List[Any]   	=	requests.get(_snake_case					).content
                   else:
                          with open(_snake_case, 'rb'					) as f:
                                 snake_case__							:	Union[str, Any]   	=	f.read()

            if isinstance(_snake_case, _snake_case					):
                   snake_case__							:	List[Any]   	=	ffmpeg_read(_snake_case, self.feature_extractor.sampling_rate					)

            if not isinstance(_snake_case, np.ndarray					):
                   raise ValueError('We expect a numpy ndarray as input'					)
            if len(audio.shape					) != 1:
                   raise ValueError('We expect a single channel audio input for ZeroShotAudioClassificationPipeline'					)

            snake_case__							:	Tuple   	=	self.feature_extractor(
                [audio], sampling_rate=self.feature_extractor.sampling_rate, return_tensors='pt'					)
            snake_case__							:	int   	=	candidate_labels
            snake_case__							:	int   	=	[hypothesis_template.format(_snake_case					) for x in candidate_labels]
            snake_case__							:	Optional[int]   	=	self.tokenizer(_snake_case, return_tensors=self.framework, padding=_snake_case					)
            snake_case__							:	List[Any]   	=	[text_inputs]
            return inputs






     def    lowercase_   (       self  :  Optional[int], _snake_case  :  Optional[Any]					)   ->int:
            snake_case__							:	Optional[int]   	=	model_inputs.pop('candidate_labels'					)
            snake_case__							:	str   	=	model_inputs.pop('text_inputs'					)
            if isinstance(text_inputs[0], _snake_case					):
                   snake_case__							:	Optional[Any]   	=	text_inputs[0]
            else:
                   # Batching case.
                   snake_case__							:	int   	=	text_inputs[0][0]

            snake_case__							:	Any   	=	self.model(**_snake_case, **_snake_case					)

            snake_case__							:	List[Any]   	=	{
                'candidate_labels': candidate_labels,
                'logits': outputs.logits_per_audio,
            }
            return model_outputs






     def    lowercase_   (       self  :  Union[str, Any], _snake_case  :  str					)   ->List[str]:
            snake_case__							:	int   	=	model_outputs.pop('candidate_labels'					)
            snake_case__							:	List[Any]   	=	model_outputs['logits'][0]

            if self.framework == "pt":
                   snake_case__							:	Tuple   	=	logits.softmax(dim=0					)
                   snake_case__							:	Union[str, Any]   	=	probs.tolist()
            else:
                   raise ValueError('`tf` framework not supported.'					)

            snake_case__							:	Union[str, Any]   	=	[
                {'score': score, 'label': candidate_label}
                for score, candidate_label in sorted(zip(_snake_case, _snake_case					), key=lambda _snake_case					: -x[0]					)
            ]
            return result






