




import unittest

from transformers import (
    MODEL_FOR_IMAGE_CLASSIFICATION_MAPPING,
    TF_MODEL_FOR_IMAGE_CLASSIFICATION_MAPPING,
    PreTrainedTokenizer,
    is_vision_available,
)
from transformers.pipelines import ImageClassificationPipeline, pipeline
from transformers.testing_utils import (
    is_pipeline_test,
    nested_simplify,
    require_tf,
    require_torch,
    require_torch_or_tf,
    require_vision,
    slow,
)

from .test_pipelines_common import ANY


if is_vision_available():
       from PIL import Image
else:




       class 	UpperCamelCase__			:

           '''simple docstring'''



           @staticmethod
           def 					_lowercase  (					*UpperCamelCase__  ,						**UpperCamelCase__  )  ->   int:
                  pass




@is_pipeline_test
@require_torch_or_tf
@require_vision
class 	UpperCamelCase__			(unittest.TestCase  ):

    '''simple docstring'''



    lowerCamelCase_   :			Optional[int]			  =   MODEL_FOR_IMAGE_CLASSIFICATION_MAPPING
    lowerCamelCase_   :			Dict			  =   TF_MODEL_FOR_IMAGE_CLASSIFICATION_MAPPING







    def 					_lowercase  (					self  ,						UpperCamelCase__  ,						UpperCamelCase__  ,						UpperCamelCase__  )  ->   Tuple:
           lowerCamelCase     :   Dict    				=      ImageClassificationPipeline(model=__snake_case  ,						image_processor=__snake_case  ,						top_k=2  )
           lowerCamelCase     :   List[Any]    				=      [
               Image.open("./tests/fixtures/tests_samples/COCO/000000039769.png"  ),
               "http://images.cocodataset.org/val2017/000000039769.jpg",
           ]
           return image_classifier, examples







    def 					_lowercase  (					self  ,						UpperCamelCase__  ,						UpperCamelCase__  )  ->   List[str]:
           lowerCamelCase     :   Tuple    				=      image_classifier("./tests/fixtures/tests_samples/COCO/000000039769.png"  )

           self.assertEqual(
               __snake_case  ,						[
                   {"score": ANY(__snake_case  ), "label": ANY(__snake_case  )},
                   {"score": ANY(__snake_case  ), "label": ANY(__snake_case  )},
               ]  ,						)

           import datasets

           lowerCamelCase     :   Any    				=      datasets.load_dataset("hf-internal-testing/fixtures_image_utils"  ,						"image"  ,						split="test"  )

           # Accepts URL + PIL.Image + lists
           lowerCamelCase     :   Optional[int]    				=      image_classifier(
               [
                   Image.open("./tests/fixtures/tests_samples/COCO/000000039769.png"  ),
                   "http://images.cocodataset.org/val2017/000000039769.jpg",
                   # RGBA
                   dataset[0]["file"],
                   # LA
                   dataset[1]["file"],
                   # L
                   dataset[2]["file"],
               ]  )
           self.assertEqual(
               __snake_case  ,						[
                   [
                       {"score": ANY(__snake_case  ), "label": ANY(__snake_case  )},
                       {"score": ANY(__snake_case  ), "label": ANY(__snake_case  )},
                   ],
                   [
                       {"score": ANY(__snake_case  ), "label": ANY(__snake_case  )},
                       {"score": ANY(__snake_case  ), "label": ANY(__snake_case  )},
                   ],
                   [
                       {"score": ANY(__snake_case  ), "label": ANY(__snake_case  )},
                       {"score": ANY(__snake_case  ), "label": ANY(__snake_case  )},
                   ],
                   [
                       {"score": ANY(__snake_case  ), "label": ANY(__snake_case  )},
                       {"score": ANY(__snake_case  ), "label": ANY(__snake_case  )},
                   ],
                   [
                       {"score": ANY(__snake_case  ), "label": ANY(__snake_case  )},
                       {"score": ANY(__snake_case  ), "label": ANY(__snake_case  )},
                   ],
               ]  ,						)







    @require_torch
    def 					_lowercase  (					self  )  ->   Tuple:
           lowerCamelCase     :   Optional[Any]    				=      "hf-internal-testing/tiny-random-vit"
           lowerCamelCase     :   List[str]    				=      pipeline("image-classification"  ,						model=__snake_case  )

           lowerCamelCase     :   Union[str, Any]    				=      image_classifier("http://images.cocodataset.org/val2017/000000039769.jpg"  )
           self.assertEqual(
               nested_simplify(__snake_case  ,						decimals=4  )  ,						[{"label": "LABEL_1", "score": 0.574}, {"label": "LABEL_0", "score": 0.426}]  ,						)

           lowerCamelCase     :   int    				=      image_classifier(
               [
                   "http://images.cocodataset.org/val2017/000000039769.jpg",
                   "http://images.cocodataset.org/val2017/000000039769.jpg",
               ]  ,						top_k=2  ,						)
           self.assertEqual(
               nested_simplify(__snake_case  ,						decimals=4  )  ,						[
                   [{"label": "LABEL_1", "score": 0.574}, {"label": "LABEL_0", "score": 0.426}],
                   [{"label": "LABEL_1", "score": 0.574}, {"label": "LABEL_0", "score": 0.426}],
               ]  ,						)







    @require_tf
    def 					_lowercase  (					self  )  ->   int:
           lowerCamelCase     :   Union[str, Any]    				=      "hf-internal-testing/tiny-random-vit"
           lowerCamelCase     :   Tuple    				=      pipeline("image-classification"  ,						model=__snake_case  ,						framework="tf"  )

           lowerCamelCase     :   List[Any]    				=      image_classifier("http://images.cocodataset.org/val2017/000000039769.jpg"  )
           self.assertEqual(
               nested_simplify(__snake_case  ,						decimals=4  )  ,						[{"label": "LABEL_1", "score": 0.574}, {"label": "LABEL_0", "score": 0.426}]  ,						)

           lowerCamelCase     :   List[str]    				=      image_classifier(
               [
                   "http://images.cocodataset.org/val2017/000000039769.jpg",
                   "http://images.cocodataset.org/val2017/000000039769.jpg",
               ]  ,						top_k=2  ,						)
           self.assertEqual(
               nested_simplify(__snake_case  ,						decimals=4  )  ,						[
                   [{"label": "LABEL_1", "score": 0.574}, {"label": "LABEL_0", "score": 0.426}],
                   [{"label": "LABEL_1", "score": 0.574}, {"label": "LABEL_0", "score": 0.426}],
               ]  ,						)







    def 					_lowercase  (					self  )  ->   Any:
           lowerCamelCase     :   Any    				=      PreTrainedTokenizer()

           # Assert that the pipeline can be initialized with a feature extractor that is not in any mapping
           lowerCamelCase     :   Any    				=      pipeline(
               "image-classification"  ,						model="hf-internal-testing/tiny-random-vit"  ,						tokenizer=__snake_case  )

           self.assertIs(image_classifier.tokenizer  ,						__snake_case  )







    @slow
    @require_torch
    def 					_lowercase  (					self  )  ->   List[Any]:
           # Perceiver is not tested by `run_pipeline_test` properly.
           # That is because the type of feature_extractor and model preprocessor need to be kept
           # in sync, which is not the case in the current design
           lowerCamelCase     :   str    				=      pipeline("image-classification"  ,						model="deepmind/vision-perceiver-conv"  )
           lowerCamelCase     :   Union[str, Any]    				=      image_classifier("http://images.cocodataset.org/val2017/000000039769.jpg"  )
           self.assertEqual(
               nested_simplify(__snake_case  ,						decimals=4  )  ,						[
                   {"score": 0.4385, "label": "tabby, tabby cat"},
                   {"score": 0.321, "label": "tiger cat"},
                   {"score": 0.0502, "label": "Egyptian cat"},
                   {"score": 0.0137, "label": "crib, cot"},
                   {"score": 0.007, "label": "radiator"},
               ]  ,						)

           lowerCamelCase     :   Tuple    				=      pipeline("image-classification"  ,						model="deepmind/vision-perceiver-fourier"  )
           lowerCamelCase     :   Tuple    				=      image_classifier("http://images.cocodataset.org/val2017/000000039769.jpg"  )
           self.assertEqual(
               nested_simplify(__snake_case  ,						decimals=4  )  ,						[
                   {"score": 0.5658, "label": "tabby, tabby cat"},
                   {"score": 0.1309, "label": "tiger cat"},
                   {"score": 0.0722, "label": "Egyptian cat"},
                   {"score": 0.0707, "label": "remote control, remote"},
                   {"score": 0.0082, "label": "computer keyboard, keypad"},
               ]  ,						)

           lowerCamelCase     :   int    				=      pipeline("image-classification"  ,						model="deepmind/vision-perceiver-learned"  )
           lowerCamelCase     :   Any    				=      image_classifier("http://images.cocodataset.org/val2017/000000039769.jpg"  )
           self.assertEqual(
               nested_simplify(__snake_case  ,						decimals=4  )  ,						[
                   {"score": 0.3022, "label": "tabby, tabby cat"},
                   {"score": 0.2362, "label": "Egyptian cat"},
                   {"score": 0.1856, "label": "tiger cat"},
                   {"score": 0.0324, "label": "remote control, remote"},
                   {"score": 0.0096, "label": "quilt, comforter, comfort, puff"},
               ]  ,						)












import subprocess
from typing import Union

import numpy as np
import requests

from ..utils import add_end_docstrings, is_torch_available, is_torchaudio_available, logging
from .base import PIPELINE_INIT_ARGS, Pipeline


if is_torch_available():
       from ..models.auto.modeling_auto import MODEL_FOR_AUDIO_CLASSIFICATION_MAPPING

SCREAMING_SNAKE_CASE__ :   Union[str, Any]						=     logging.get_logger(__name__)

def   A   (   _SCREAMING_SNAKE_CASE				,_SCREAMING_SNAKE_CASE					)			->				np.array:
       lowerCamelCase     :   Optional[Any]    				=      f'''{sampling_rate}'''
       lowerCamelCase     :   Optional[Any]    				=      "1"
       lowerCamelCase     :   Tuple    				=      "f32le"
       lowerCamelCase     :   Any    				=      [
           "ffmpeg",
           "-i",
           "pipe:0",
           "-ac",
           ac,
           "-ar",
           ar,
           "-f",
           format_for_conversion,
           "-hide_banner",
           "-loglevel",
           "quiet",
           "pipe:1",
       ]

       try:
              lowerCamelCase     :   Optional[int]    				=      subprocess.Popen(_SCREAMING_SNAKE_CASE				,stdin=subprocess.PIPE				,stdout=subprocess.PIPE					)
       except FileNotFoundError:
              raise ValueError("ffmpeg was not found but is required to load audio files from filename"					)
       lowerCamelCase     :   Optional[int]    				=      ffmpeg_process.communicate(_SCREAMING_SNAKE_CASE					)
       lowerCamelCase     :   Union[str, Any]    				=      output_stream[0]

       lowerCamelCase     :   Dict    				=      np.frombuffer(_SCREAMING_SNAKE_CASE				,np.floataa					)
       if audio.shape[0] == 0:
              raise ValueError("Malformed soundfile"					)
       return audio




@add_end_docstrings(lowerCAmelCase__  )
class 	UpperCamelCase__			(lowerCAmelCase__  ):

    '''simple docstring'''



    def __init__(					self  ,						*UpperCamelCase__  ,						**UpperCamelCase__  )  ->   Dict:
           # Default, might be overriden by the model.config.
           lowerCamelCase     :   Union[str, Any]    				=      5
           super().__init__(*UpperCamelCase__  ,						**UpperCamelCase__  )

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

           self.check_model_type(UpperCamelCase__  )







    def __call__(					self  ,						UpperCamelCase__  ,						**UpperCamelCase__  ,						)  ->   Optional[int]:
           return super().__call__(UpperCamelCase__  ,						**UpperCamelCase__  )







    def 					_lowercase  (					self  ,						UpperCamelCase__=None  ,						**UpperCamelCase__  )  ->   str:
           # No parameters on this pipeline right now
           lowerCamelCase     :   int    				=      {}
           if top_k is not None:
                  if top_k > self.model.config.num_labels:
                         lowerCamelCase     :   Any    				=      self.model.config.num_labels
                  lowerCamelCase     :   Any    				=      top_k
           return {}, {}, postprocess_params







    def 					_lowercase  (					self  ,						UpperCamelCase__  )  ->   Dict:
           if isinstance(UpperCamelCase__  ,						UpperCamelCase__  ):
                  if inputs.startswith("http://"  ) or inputs.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
                         lowerCamelCase     :   str    				=      requests.get(UpperCamelCase__  ).content
                  else:
                         with open(UpperCamelCase__  ,						"rb"  ) as f:
                                lowerCamelCase     :   int    				=      f.read()

           if isinstance(UpperCamelCase__  ,						UpperCamelCase__  ):
                  lowerCamelCase     :   str    				=      ffmpeg_read(UpperCamelCase__  ,						self.feature_extractor.sampling_rate  )

           if isinstance(UpperCamelCase__  ,						UpperCamelCase__  ):
                  # Accepting `"array"` which is the key defined in `datasets` for
                  # better integration
                  if not ("sampling_rate" in inputs and ("raw" in inputs or "array" in inputs)):
                         raise ValueError(
                             "When passing a dictionary to AudioClassificationPipeline, the dict needs to contain a "
                             "\"raw\" key containing the numpy array representing the audio and a \"sampling_rate\" key, "
                             "containing the sampling_rate associated with that array"  )

                  lowerCamelCase     :   Optional[int]    				=      inputs.pop("raw"  ,						UpperCamelCase__  )
                  if _inputs is None:
                         # Remove path which will not be used from `datasets`.
                         inputs.pop("path"  ,						UpperCamelCase__  )
                         lowerCamelCase     :   List[str]    				=      inputs.pop("array"  ,						UpperCamelCase__  )
                  lowerCamelCase     :   str    				=      inputs.pop("sampling_rate"  )
                  lowerCamelCase     :   Union[str, Any]    				=      _inputs
                  if in_sampling_rate != self.feature_extractor.sampling_rate:
                         import torch

                         if is_torchaudio_available():
                                from torchaudio import functional as F
                         else:
                                raise ImportError(
                                    "torchaudio is required to resample audio samples in AudioClassificationPipeline. "
                                    "The torchaudio package can be installed through: `pip install torchaudio`."  )

                         lowerCamelCase     :   Tuple    				=      F.resample(
                             torch.from_numpy(UpperCamelCase__  )  ,						UpperCamelCase__  ,						self.feature_extractor.sampling_rate  ).numpy()

           if not isinstance(UpperCamelCase__  ,						np.ndarray  ):
                  raise ValueError("We expect a numpy ndarray as input"  )
           if len(inputs.shape  ) != 1:
                  raise ValueError("We expect a single channel audio input for AudioClassificationPipeline"  )

           lowerCamelCase     :   Optional[int]    				=      self.feature_extractor(
               UpperCamelCase__  ,						sampling_rate=self.feature_extractor.sampling_rate  ,						return_tensors="pt"  )
           return processed







    def 					_lowercase  (					self  ,						UpperCamelCase__  )  ->   str:
           lowerCamelCase     :   Optional[Any]    				=      self.model(**UpperCamelCase__  )
           return model_outputs







    def 					_lowercase  (					self  ,						UpperCamelCase__  ,						UpperCamelCase__=5  )  ->   Tuple:
           lowerCamelCase     :   Tuple    				=      model_outputs.logits[0].softmax(-1  )
           lowerCamelCase				,   lowerCamelCase     :   str    				=      probs.topk(UpperCamelCase__  )

           lowerCamelCase     :   str    				=      scores.tolist()
           lowerCamelCase     :   Optional[int]    				=      ids.tolist()

           lowerCamelCase     :   Optional[int]    				=      [{"score": score, "label": self.model.config.idalabel[_id]} for score, _id in zip(UpperCamelCase__  ,						UpperCamelCase__  )]

           return labels






