




import random
import shutil
import tempfile
import unittest

import numpy as np

from transformers import TaTokenizer, TaTokenizerFast
from transformers.testing_utils import require_sentencepiece, require_torch
from transformers.utils.import_utils import is_speech_available, is_torch_available


if is_torch_available():
       pass

if is_speech_available():
       from transformers import EncodecFeatureExtractor, MusicgenProcessor


SCREAMING_SNAKE_CASE__ :   Optional[Any]						=     random.Random()

def   A   (   _SCREAMING_SNAKE_CASE				,_SCREAMING_SNAKE_CASE=1.0				,_SCREAMING_SNAKE_CASE=None				,_SCREAMING_SNAKE_CASE=None					)			->				List[str]:

       '''simple docstring'''





       if rng is None:
              lowerCamelCase     :   Any    				=      global_rng

       lowerCamelCase     :   Optional[int]    				=      []
       for batch_idx in range(shape[0]					):
              values.append([]					)
              for _ in range(shape[1]					):
                     values[-1].append(rng.random() * scale					)

       return values




@require_torch
@require_sentencepiece
class 	UpperCamelCase__			(unittest.TestCase  ):

    '''simple docstring'''



    def 					_lowercase  (					self  )  ->   Union[str, Any]:
           lowerCamelCase     :   str    				=      "facebook/musicgen-small"
           lowerCamelCase     :   str    				=      tempfile.mkdtemp()







    def 					_lowercase  (					self  ,						**UpperCamelCase__  )  ->   int:
           return TaTokenizer.from_pretrained(self.checkpoint  ,						**UpperCamelCase__  )







    def 					_lowercase  (					self  ,						**UpperCamelCase__  )  ->   List[Any]:
           return EncodecFeatureExtractor.from_pretrained(self.checkpoint  ,						**UpperCamelCase__  )







    def 					_lowercase  (					self  )  ->   Any:
           shutil.rmtree(self.tmpdirname  )







    def 					_lowercase  (					self  )  ->   str:
           lowerCamelCase     :   Optional[int]    				=      self.get_tokenizer()
           lowerCamelCase     :   int    				=      self.get_feature_extractor()

           lowerCamelCase     :   List[Any]    				=      MusicgenProcessor(tokenizer=UpperCamelCase__  ,						feature_extractor=UpperCamelCase__  )

           processor.save_pretrained(self.tmpdirname  )
           lowerCamelCase     :   Any    				=      MusicgenProcessor.from_pretrained(self.tmpdirname  )

           self.assertEqual(processor.tokenizer.get_vocab()  ,						tokenizer.get_vocab()  )
           self.assertIsInstance(processor.tokenizer  ,						UpperCamelCase__  )

           self.assertEqual(processor.feature_extractor.to_json_string()  ,						feature_extractor.to_json_string()  )
           self.assertIsInstance(processor.feature_extractor  ,						UpperCamelCase__  )







    def 					_lowercase  (					self  )  ->   Union[str, Any]:
           lowerCamelCase     :   int    				=      MusicgenProcessor(tokenizer=self.get_tokenizer()  ,						feature_extractor=self.get_feature_extractor()  )
           processor.save_pretrained(self.tmpdirname  )

           lowerCamelCase     :   Any    				=      self.get_tokenizer(bos_token="(BOS)"  ,						eos_token="(EOS)"  )
           lowerCamelCase     :   Tuple    				=      self.get_feature_extractor(do_normalize=UpperCamelCase__  ,						padding_value=1.0  )

           lowerCamelCase     :   int    				=      MusicgenProcessor.from_pretrained(
               self.tmpdirname  ,						bos_token="(BOS)"  ,						eos_token="(EOS)"  ,						do_normalize=UpperCamelCase__  ,						padding_value=1.0  )

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

           self.assertEqual(processor.feature_extractor.to_json_string()  ,						feature_extractor_add_kwargs.to_json_string()  )
           self.assertIsInstance(processor.feature_extractor  ,						UpperCamelCase__  )







    def 					_lowercase  (					self  )  ->   List[str]:
           lowerCamelCase     :   str    				=      self.get_feature_extractor()
           lowerCamelCase     :   Optional[Any]    				=      self.get_tokenizer()

           lowerCamelCase     :   List[str]    				=      MusicgenProcessor(tokenizer=UpperCamelCase__  ,						feature_extractor=UpperCamelCase__  )

           lowerCamelCase     :   str    				=      floats_list((3, 1000)  )

           lowerCamelCase     :   List[Any]    				=      feature_extractor(UpperCamelCase__  ,						return_tensors="np"  )
           lowerCamelCase     :   Union[str, Any]    				=      processor(UpperCamelCase__  ,						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 					_lowercase  (					self  )  ->   int:
           lowerCamelCase     :   Any    				=      self.get_feature_extractor()
           lowerCamelCase     :   List[str]    				=      self.get_tokenizer()

           lowerCamelCase     :   int    				=      MusicgenProcessor(tokenizer=UpperCamelCase__  ,						feature_extractor=UpperCamelCase__  )

           lowerCamelCase     :   Optional[int]    				=      "This is a test string"

           lowerCamelCase     :   List[Any]    				=      processor(text=UpperCamelCase__  )

           lowerCamelCase     :   Tuple    				=      tokenizer(UpperCamelCase__  )

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







    def 					_lowercase  (					self  )  ->   Union[str, Any]:
           lowerCamelCase     :   List[str]    				=      self.get_feature_extractor()
           lowerCamelCase     :   Optional[int]    				=      self.get_tokenizer()

           lowerCamelCase     :   Union[str, Any]    				=      MusicgenProcessor(tokenizer=UpperCamelCase__  ,						feature_extractor=UpperCamelCase__  )

           lowerCamelCase     :   List[Any]    				=      [[1, 4, 5, 8, 1, 0, 8], [3, 4, 3, 1, 1, 8, 9]]

           lowerCamelCase     :   Tuple    				=      processor.batch_decode(sequences=UpperCamelCase__  )
           lowerCamelCase     :   List[str]    				=      tokenizer.batch_decode(UpperCamelCase__  )

           self.assertListEqual(UpperCamelCase__  ,						UpperCamelCase__  )







    def 					_lowercase  (					self  )  ->   Tuple:
           lowerCamelCase     :   Tuple    				=      self.get_feature_extractor()
           lowerCamelCase     :   List[Any]    				=      self.get_tokenizer()

           lowerCamelCase     :   Tuple    				=      MusicgenProcessor(tokenizer=UpperCamelCase__  ,						feature_extractor=UpperCamelCase__  )

           self.assertListEqual(
               processor.model_input_names  ,						feature_extractor.model_input_names  ,						msg="`processor` and `feature_extractor` model input names do not match"  ,						)







    def 					_lowercase  (					self  )  ->   Union[str, Any]:
           lowerCamelCase     :   Optional[int]    				=      self.get_feature_extractor(padding_side="left"  )
           lowerCamelCase     :   Optional[Any]    				=      self.get_tokenizer()

           lowerCamelCase     :   Any    				=      MusicgenProcessor(tokenizer=UpperCamelCase__  ,						feature_extractor=UpperCamelCase__  )

           lowerCamelCase     :   Tuple    				=      [floats_list((1, x)  )[0] for x in range(5  ,						20  ,						5  )]
           lowerCamelCase     :   Union[str, Any]    				=      processor(UpperCamelCase__  ).padding_mask

           lowerCamelCase     :   List[Any]    				=      np.asarray(floats_list((3, 20)  )  )[:, None, :]
           lowerCamelCase     :   str    				=      processor.batch_decode(UpperCamelCase__  ,						padding_mask=UpperCamelCase__  )

           self.assertIsInstance(UpperCamelCase__  ,						UpperCamelCase__  )

           for audio in decoded_audios:
                  self.assertIsInstance(UpperCamelCase__  ,						np.ndarray  )

           self.assertTrue(decoded_audios[0].shape == (1, 10)  )
           self.assertTrue(decoded_audios[1].shape == (1, 15)  )
           self.assertTrue(decoded_audios[2].shape == (1, 20)  )












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






