




'''simple docstring'''




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_      :		Any							      =       logging.get_logger(__name__)


@add_end_docstrings(lowerCamelCase__				)
class   __UpperCamelCase       ( lowerCamelCase__				):






       def __init__(							self,		**lowerCAmelCase  ):



              """simple docstring"""







              super().__init__(**lowerCAmelCase  )

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






       def __call__(							self,		lowerCAmelCase,		**lowerCAmelCase  ):



              """simple docstring"""







              return super().__call__(lowerCAmelCase,		**lowerCAmelCase  )






       def      lowercase__     (							self,		**lowerCAmelCase  ):



              """simple docstring"""







              lowerCamelCase_							={}
              if "candidate_labels" in kwargs:
                     lowerCamelCase_							=kwargs['''candidate_labels''']
              if "hypothesis_template" in kwargs:
                     lowerCamelCase_							=kwargs['''hypothesis_template''']

              return preprocess_params, {}, {}






       def      lowercase__     (							self,		lowerCAmelCase,		lowerCAmelCase=None,		lowerCAmelCase="This is a sound of {}."  ):



              """simple docstring"""







              if isinstance(lowerCAmelCase,		lowerCAmelCase  ):
                     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
                            lowerCamelCase_							=requests.get(lowerCAmelCase  ).content
                     else:
                            with open(lowerCAmelCase,		'''rb'''  ) as f:
                                   lowerCamelCase_							=f.read()

              if isinstance(lowerCAmelCase,		lowerCAmelCase  ):
                     lowerCamelCase_							=ffmpeg_read(lowerCAmelCase,		self.feature_extractor.sampling_rate  )

              if not isinstance(lowerCAmelCase,		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'''  )

              lowerCamelCase_							=self.feature_extractor(
                  [audio],		sampling_rate=self.feature_extractor.sampling_rate,		return_tensors='''pt'''  )
              lowerCamelCase_							=candidate_labels
              lowerCamelCase_							=[hypothesis_template.format(lowerCAmelCase  ) for x in candidate_labels]
              lowerCamelCase_							=self.tokenizer(lowerCAmelCase,		return_tensors=self.framework,		padding=lowerCAmelCase  )
              lowerCamelCase_							=[text_inputs]
              return inputs






       def      lowercase__     (							self,		lowerCAmelCase  ):



              """simple docstring"""







              lowerCamelCase_							=model_inputs.pop('''candidate_labels'''  )
              lowerCamelCase_							=model_inputs.pop('''text_inputs'''  )
              if isinstance(text_inputs[0],		lowerCAmelCase  ):
                     lowerCamelCase_							=text_inputs[0]
              else:
                     # Batching case.
                     lowerCamelCase_							=text_inputs[0][0]

              lowerCamelCase_							=self.model(**lowerCAmelCase,		**lowerCAmelCase  )

              lowerCamelCase_							={
                  '''candidate_labels''': candidate_labels,
                  '''logits''': outputs.logits_per_audio,
              }
              return model_outputs






       def      lowercase__     (							self,		lowerCAmelCase  ):



              """simple docstring"""







              lowerCamelCase_							=model_outputs.pop('''candidate_labels'''  )
              lowerCamelCase_							=model_outputs['''logits'''][0]

              if self.framework == "pt":
                     lowerCamelCase_							=logits.softmax(dim=0  )
                     lowerCamelCase_							=probs.tolist()
              else:
                     raise ValueError('''`tf` framework not supported.'''  )

              lowerCamelCase_							=[
                  {'''score''': score, '''label''': candidate_label}
                  for score, candidate_label in sorted(zip(lowerCAmelCase,		lowerCAmelCase  ),		key=lambda lowerCAmelCase  : -x[0]  )
              ]
              return result





import pytest

from datasets.splits import SplitDict, SplitInfo
from datasets.utils.py_utils import asdict
@pytest.mark.parametrize(
    '''split_dict'''		,  [
        SplitDict(),
        SplitDict({'''train''': SplitInfo(name='''train'''		,  num_bytes=1_337		,  num_examples=42		,  dataset_name='''my_dataset'''						)}						),
        SplitDict({'''train''': SplitInfo(name='''train'''		,  num_bytes=1_337		,  num_examples=42						)}						),
        SplitDict({'''train''': SplitInfo()}						),
    ]		,  )
def       A	(    _lowercase						):
						SCREAMING_SNAKE_CASE							:    Tuple							=							split_dict._to_yaml_list()
						assert len(_lowercase						) == len(_lowercase						)
						SCREAMING_SNAKE_CASE							:    Tuple							=							SplitDict._from_yaml_list(_lowercase						)
						for split_name, split_info in split_dict.items():
												# dataset_name field is deprecated, and is therefore not part of the YAML dump
												SCREAMING_SNAKE_CASE							:    Any							=							None
												# the split name of split_dict takes over the name of the split info object
												SCREAMING_SNAKE_CASE							:    Optional[Any]							=							split_name
						assert split_dict == reloaded



@pytest.mark.parametrize(
    '''split_info'''		,  [SplitInfo(), SplitInfo(dataset_name=_lowercase						), SplitInfo(dataset_name='''my_dataset'''						)]						)
def       A	(    _lowercase						):
						# For backward compatibility, we need asdict(split_dict) to return split info dictrionaries with the "dataset_name"
						# field even if it's deprecated. This way old versionso of `datasets` can still reload dataset_infos.json files
						SCREAMING_SNAKE_CASE							:    List[Any]							=							asdict(SplitDict({'''train''': split_info}						)						)
						assert "dataset_name" in split_dict_asdict["train"]
						assert split_dict_asdict["train"]["dataset_name"] == split_info.dataset_name





