





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

UpperCamelCase__          =logging.get_logger(__name__)






def 			lowerCamelCase__	(__lowerCamelCase,				__lowerCamelCase		):
     _SCREAMING_SNAKE_CASE				:		Dict  			=							f"""{sampling_rate}"""
     _SCREAMING_SNAKE_CASE				:		List[Any]  			=							"1"
     _SCREAMING_SNAKE_CASE				:		Tuple  			=							"f32le"
     _SCREAMING_SNAKE_CASE				:		Union[str, Any]  			=							[
         "ffmpeg",
         "-i",
         "pipe:0",
         "-ac",
         ac,
         "-ar",
         ar,
         "-f",
         format_for_conversion,
         "-hide_banner",
         "-loglevel",
         "quiet",
         "pipe:1",
     ]

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

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





@add_end_docstrings(__lowercase			)
class 	lowerCAmelCase__(       __lowercase			):






       '''simple docstring'''


       def __init__(					self			,  *__lowerCamelCase			,  **__lowerCamelCase							)	->   int:
            # Default, might be overriden by the model.config.
            _SCREAMING_SNAKE_CASE				:		List[str]  			=							5
            super().__init__(*__lowerCamelCase			,  **__lowerCamelCase							)

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

            self.check_model_type(__lowerCamelCase							)




       def __call__(					self			,  __lowerCamelCase			,  **__lowerCamelCase			,  )	->   Tuple:
            return super().__call__(__lowerCamelCase			,  **__lowerCamelCase							)




       def 			UpperCamelCase_	(					self			,  __lowerCamelCase=None			,  **__lowerCamelCase							)	->   int:
            # No parameters on this pipeline right now
            _SCREAMING_SNAKE_CASE				:		Optional[int]  			=							{}
            if top_k is not None:
                 if top_k > self.model.config.num_labels:
                      _SCREAMING_SNAKE_CASE				:		Union[str, Any]  			=							self.model.config.num_labels
                 _SCREAMING_SNAKE_CASE				:		str  			=							top_k
            return {}, {}, postprocess_params




       def 			UpperCamelCase_	(					self			,  __lowerCamelCase							)	->   Union[str, Any]:
            if isinstance(__lowerCamelCase			,  __lowerCamelCase							):
                 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
                      _SCREAMING_SNAKE_CASE				:		str  			=							requests.get(__lowerCamelCase							).content
                 else:
                      with open(__lowerCamelCase			,  "rb"							) as f:
                           _SCREAMING_SNAKE_CASE				:		str  			=							f.read()

            if isinstance(__lowerCamelCase			,  __lowerCamelCase							):
                 _SCREAMING_SNAKE_CASE				:		Dict  			=							ffmpeg_read(__lowerCamelCase			,  self.feature_extractor.sampling_rate							)

            if isinstance(__lowerCamelCase			,  __lowerCamelCase							):
                 # 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"							)

                 _SCREAMING_SNAKE_CASE				:		int  			=							inputs.pop("raw"			,  __lowerCamelCase							)
                 if _inputs is None:
                      # Remove path which will not be used from `datasets`.
                      inputs.pop("path"			,  __lowerCamelCase							)
                      _SCREAMING_SNAKE_CASE				:		Union[str, Any]  			=							inputs.pop("array"			,  __lowerCamelCase							)
                 _SCREAMING_SNAKE_CASE				:		int  			=							inputs.pop("sampling_rate"							)
                 _SCREAMING_SNAKE_CASE				:		List[str]  			=							_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`."							)

                      _SCREAMING_SNAKE_CASE				:		List[str]  			=							F.resample(
                          torch.from_numpy(__lowerCamelCase							)			,  __lowerCamelCase			,  self.feature_extractor.sampling_rate							).numpy()

            if not isinstance(__lowerCamelCase			,  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"							)

            _SCREAMING_SNAKE_CASE				:		int  			=							self.feature_extractor(
                __lowerCamelCase			,  sampling_rate=self.feature_extractor.sampling_rate			,  return_tensors="pt"							)
            return processed




       def 			UpperCamelCase_	(					self			,  __lowerCamelCase							)	->   Tuple:
            _SCREAMING_SNAKE_CASE				:		Dict  			=							self.model(**__lowerCamelCase							)
            return model_outputs







       def 			UpperCamelCase_	(					self			,  __lowerCamelCase			,  __lowerCamelCase=5							)	->   Union[str, Any]:
            _SCREAMING_SNAKE_CASE				:		List[Any]  			=							model_outputs.logits[0].softmax(-1							)
            _SCREAMING_SNAKE_CASE			,	_SCREAMING_SNAKE_CASE				:		Optional[int]  			=							probs.topk(__lowerCamelCase							)

            _SCREAMING_SNAKE_CASE				:		List[Any]  			=							scores.tolist()
            _SCREAMING_SNAKE_CASE				:		Union[str, Any]  			=							ids.tolist()

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

            return labels






import json
import os
import random
import shutil
import tempfile
import unittest

import numpy as np
import pytest

from transformers import BertTokenizer, BertTokenizerFast
from transformers.models.bert.tokenization_bert import VOCAB_FILES_NAMES
from transformers.testing_utils import require_vision
from transformers.utils import IMAGE_PROCESSOR_NAME, is_vision_available


if is_vision_available():
       from PIL import Image

       from transformers import FlavaImageProcessor, FlavaProcessor
       from transformers.models.flava.image_processing_flava import (
           FLAVA_CODEBOOK_MEAN,
           FLAVA_CODEBOOK_STD,
           FLAVA_IMAGE_MEAN,
           FLAVA_IMAGE_STD,
       )





@require_vision
class 	lowerCAmelCase__(       unittest.TestCase			):






       '''simple docstring'''


       def 			UpperCamelCase_	(					self							)	->   str:
            _SCREAMING_SNAKE_CASE				:		str  			=							tempfile.mkdtemp()

            # fmt: off
            _SCREAMING_SNAKE_CASE				:		Optional[Any]  			=							["[UNK]", "[CLS]", "[SEP]", "[PAD]", "[MASK]", "want", "##want", "##ed", "wa", "un", "runn", "##ing", ",", "low", "lowest"]
            # fmt: on
            _SCREAMING_SNAKE_CASE				:		int  			=							os.path.join(self.tmpdirname			,  VOCAB_FILES_NAMES["vocab_file"]							)

            with open(self.vocab_file			,  "w"			,  encoding="utf-8"							) as fp:
                 fp.write("".join([x + "\n" for x in vocab_tokens]							)							)

            _SCREAMING_SNAKE_CASE				:		Any  			=							{
                "image_mean": FLAVA_IMAGE_MEAN,
                "image_std": FLAVA_IMAGE_STD,
                "do_normalize": True,
                "do_resize": True,
                "size": 2_2_4,
                "do_center_crop": True,
                "crop_size": 2_2_4,
                "input_size_patches": 1_4,
                "total_mask_patches": 7_5,
                "mask_group_max_patches": None,
                "mask_group_min_patches": 1_6,
                "mask_group_min_aspect_ratio": 0.3,
                "mask_group_max_aspect_ratio": None,
                "codebook_do_resize": True,
                "codebook_size": 1_1_2,
                "codebook_do_center_crop": True,
                "codebook_crop_size": 1_1_2,
                "codebook_do_map_pixels": True,
                "codebook_do_normalize": True,
                "codebook_image_mean": FLAVA_CODEBOOK_MEAN,
                "codebook_image_std": FLAVA_CODEBOOK_STD,
            }

            _SCREAMING_SNAKE_CASE				:		Union[str, Any]  			=							os.path.join(self.tmpdirname			,  __lowerCamelCase							)
            with open(self.image_processor_file			,  "w"			,  encoding="utf-8"							) as fp:
                 json.dump(__lowerCamelCase			,  __lowerCamelCase							)




       def 			UpperCamelCase_	(					self			,  **__lowerCamelCase							)	->   List[str]:
            return BertTokenizer.from_pretrained(self.tmpdirname			,  **__lowerCamelCase							)




       def 			UpperCamelCase_	(					self			,  **__lowerCamelCase							)	->   Optional[Any]:
            return BertTokenizerFast.from_pretrained(self.tmpdirname			,  **__lowerCamelCase							)




       def 			UpperCamelCase_	(					self			,  **__lowerCamelCase							)	->   Tuple:
            return FlavaImageProcessor.from_pretrained(self.tmpdirname			,  **__lowerCamelCase							)




       def 			UpperCamelCase_	(					self							)	->   int:
            shutil.rmtree(self.tmpdirname							)




       def 			UpperCamelCase_	(					self							)	->   Optional[int]:
            _SCREAMING_SNAKE_CASE				:		Optional[int]  			=							[np.random.randint(2_5_5			,  size=(3, 3_0, 4_0_0)			,  dtype=np.uinta							)]

            _SCREAMING_SNAKE_CASE				:		Union[str, Any]  			=							[Image.fromarray(np.moveaxis(__lowerCamelCase			,  0			,  -1							)							) for x in image_inputs]

            return image_inputs




       def 			UpperCamelCase_	(					self							)	->   Any:
            _SCREAMING_SNAKE_CASE				:		Dict  			=							self.get_tokenizer()
            _SCREAMING_SNAKE_CASE				:		Any  			=							self.get_rust_tokenizer()
            _SCREAMING_SNAKE_CASE				:		Dict  			=							self.get_image_processor()

            _SCREAMING_SNAKE_CASE				:		List[Any]  			=							FlavaProcessor(tokenizer=__lowerCamelCase			,  image_processor=__lowerCamelCase							)
            processor_slow.save_pretrained(self.tmpdirname							)
            _SCREAMING_SNAKE_CASE				:		Tuple  			=							FlavaProcessor.from_pretrained(self.tmpdirname			,  use_fast=__lowerCamelCase							)

            _SCREAMING_SNAKE_CASE				:		Optional[Any]  			=							FlavaProcessor(tokenizer=__lowerCamelCase			,  image_processor=__lowerCamelCase							)
            processor_fast.save_pretrained(self.tmpdirname							)
            _SCREAMING_SNAKE_CASE				:		int  			=							FlavaProcessor.from_pretrained(self.tmpdirname							)

            self.assertEqual(processor_slow.tokenizer.get_vocab()			,  tokenizer_slow.get_vocab()							)
            self.assertEqual(processor_fast.tokenizer.get_vocab()			,  tokenizer_fast.get_vocab()							)
            self.assertEqual(tokenizer_slow.get_vocab()			,  tokenizer_fast.get_vocab()							)
            self.assertIsInstance(processor_slow.tokenizer			,  __lowerCamelCase							)
            self.assertIsInstance(processor_fast.tokenizer			,  __lowerCamelCase							)

            self.assertEqual(processor_slow.image_processor.to_json_string()			,  image_processor.to_json_string()							)
            self.assertEqual(processor_fast.image_processor.to_json_string()			,  image_processor.to_json_string()							)
            self.assertIsInstance(processor_slow.image_processor			,  __lowerCamelCase							)
            self.assertIsInstance(processor_fast.image_processor			,  __lowerCamelCase							)




       def 			UpperCamelCase_	(					self							)	->   Optional[int]:
            _SCREAMING_SNAKE_CASE				:		Dict  			=							FlavaProcessor(tokenizer=self.get_tokenizer()			,  image_processor=self.get_image_processor()							)
            processor.save_pretrained(self.tmpdirname							)

            _SCREAMING_SNAKE_CASE				:		str  			=							self.get_tokenizer(bos_token="(BOS)"			,  eos_token="(EOS)"							)
            _SCREAMING_SNAKE_CASE				:		List[str]  			=							self.get_image_processor(do_normalize=__lowerCamelCase			,  padding_value=1.0							)

            _SCREAMING_SNAKE_CASE				:		Dict  			=							FlavaProcessor.from_pretrained(
                self.tmpdirname			,  bos_token="(BOS)"			,  eos_token="(EOS)"			,  do_normalize=__lowerCamelCase			,  padding_value=1.0							)

            self.assertEqual(processor.tokenizer.get_vocab()			,  tokenizer_add_kwargs.get_vocab()							)
            self.assertIsInstance(processor.tokenizer			,  __lowerCamelCase							)

            self.assertEqual(processor.image_processor.to_json_string()			,  image_processor_add_kwargs.to_json_string()							)
            self.assertIsInstance(processor.image_processor			,  __lowerCamelCase							)




       def 			UpperCamelCase_	(					self							)	->   Any:
            _SCREAMING_SNAKE_CASE				:		str  			=							self.get_image_processor()
            _SCREAMING_SNAKE_CASE				:		Union[str, Any]  			=							self.get_tokenizer()

            _SCREAMING_SNAKE_CASE				:		Any  			=							FlavaProcessor(tokenizer=__lowerCamelCase			,  image_processor=__lowerCamelCase							)

            _SCREAMING_SNAKE_CASE				:		int  			=							self.prepare_image_inputs()

            _SCREAMING_SNAKE_CASE				:		List[str]  			=							image_processor(__lowerCamelCase			,  return_tensors="np"							)
            _SCREAMING_SNAKE_CASE				:		str  			=							processor(images=__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							)

            # With rest of the args
            random.seed(1_2_3_4							)
            _SCREAMING_SNAKE_CASE				:		List[Any]  			=							image_processor(
                __lowerCamelCase			,  return_image_mask=__lowerCamelCase			,  return_codebook_pixels=__lowerCamelCase			,  return_tensors="np"							)
            random.seed(1_2_3_4							)
            _SCREAMING_SNAKE_CASE				:		Optional[Any]  			=							processor(
                images=__lowerCamelCase			,  return_image_mask=__lowerCamelCase			,  return_codebook_pixels=__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							)	->   int:
            _SCREAMING_SNAKE_CASE				:		Any  			=							self.get_image_processor()
            _SCREAMING_SNAKE_CASE				:		Union[str, Any]  			=							self.get_tokenizer()

            _SCREAMING_SNAKE_CASE				:		int  			=							FlavaProcessor(tokenizer=__lowerCamelCase			,  image_processor=__lowerCamelCase							)

            _SCREAMING_SNAKE_CASE				:		Optional[Any]  			=							"lower newer"

            _SCREAMING_SNAKE_CASE				:		Union[str, Any]  			=							processor(text=__lowerCamelCase							)

            _SCREAMING_SNAKE_CASE				:		List[str]  			=							tokenizer(__lowerCamelCase							)

            for key in encoded_tok.keys():
                 self.assertListEqual(encoded_tok[key]			,  encoded_processor[key]							)




       def 			UpperCamelCase_	(					self							)	->   Tuple:
            _SCREAMING_SNAKE_CASE				:		Union[str, Any]  			=							self.get_image_processor()
            _SCREAMING_SNAKE_CASE				:		Optional[Any]  			=							self.get_tokenizer()

            _SCREAMING_SNAKE_CASE				:		List[Any]  			=							FlavaProcessor(tokenizer=__lowerCamelCase			,  image_processor=__lowerCamelCase							)

            _SCREAMING_SNAKE_CASE				:		Any  			=							"lower newer"
            _SCREAMING_SNAKE_CASE				:		Dict  			=							self.prepare_image_inputs()

            _SCREAMING_SNAKE_CASE				:		List[Any]  			=							processor(text=__lowerCamelCase			,  images=__lowerCamelCase							)

            self.assertListEqual(list(inputs.keys()							)			,  ["input_ids", "token_type_ids", "attention_mask", "pixel_values"]							)

            # add extra args
            _SCREAMING_SNAKE_CASE				:		Tuple  			=							processor(text=__lowerCamelCase			,  images=__lowerCamelCase			,  return_codebook_pixels=__lowerCamelCase			,  return_image_mask=__lowerCamelCase							)

            self.assertListEqual(
                list(inputs.keys()							)			,  [
                    "input_ids",
                    "token_type_ids",
                    "attention_mask",
                    "pixel_values",
                    "codebook_pixel_values",
                    "bool_masked_pos",
                ]			,  )

            # test if it raises when no input is passed
            with pytest.raises(__lowerCamelCase							):
                 processor()




       def 			UpperCamelCase_	(					self							)	->   Dict:
            _SCREAMING_SNAKE_CASE				:		Optional[int]  			=							self.get_image_processor()
            _SCREAMING_SNAKE_CASE				:		str  			=							self.get_tokenizer()

            _SCREAMING_SNAKE_CASE				:		int  			=							FlavaProcessor(tokenizer=__lowerCamelCase			,  image_processor=__lowerCamelCase							)

            _SCREAMING_SNAKE_CASE				:		Optional[int]  			=							[[1, 4, 5, 8, 1, 0, 8], [3, 4, 3, 1, 1, 8, 9]]

            _SCREAMING_SNAKE_CASE				:		Dict  			=							processor.batch_decode(__lowerCamelCase							)
            _SCREAMING_SNAKE_CASE				:		Any  			=							tokenizer.batch_decode(__lowerCamelCase							)

            self.assertListEqual(__lowerCamelCase			,  __lowerCamelCase							)







       def 			UpperCamelCase_	(					self							)	->   Any:
            _SCREAMING_SNAKE_CASE				:		int  			=							self.get_image_processor()
            _SCREAMING_SNAKE_CASE				:		Union[str, Any]  			=							self.get_tokenizer()

            _SCREAMING_SNAKE_CASE				:		Dict  			=							FlavaProcessor(tokenizer=__lowerCamelCase			,  image_processor=__lowerCamelCase							)

            _SCREAMING_SNAKE_CASE				:		Optional[Any]  			=							"lower newer"
            _SCREAMING_SNAKE_CASE				:		Optional[int]  			=							self.prepare_image_inputs()

            _SCREAMING_SNAKE_CASE				:		List[Any]  			=							processor(text=__lowerCamelCase			,  images=__lowerCamelCase							)

            self.assertListEqual(list(inputs.keys()							)			,  processor.model_input_names							)