



"""simple docstring"""
from typing import List

import numpy as np

def  _snake_case		(		UpperCamelCase		:						Optional[Any]		):
 UpperCAmelCase      :   Any      =      {key: len(UpperCAmelCase_		) for key, value in gen_kwargs.items() if isinstance(UpperCAmelCase_     ,						UpperCAmelCase_		)}
 if len(set(lists_lengths.values()		)		) > 1:
  raise RuntimeError(
      (
          """Sharding is ambiguous for this dataset: """
          + """we found several data sources lists of different lengths, and we don\'t know over which list we should parallelize:\n"""
          + """\n""".join(F"\t- key {key} has length {length}" for key, length in lists_lengths.items()		)
          + """\nTo fix this, check the \'gen_kwargs\' and make sure to use lists only for data sources, """
          + """and use tuples otherwise. In the end there should only be one single list, or several lists with the same length."""
      )		)
 UpperCAmelCase      :   Tuple      =      max(lists_lengths.values()     ,						default=0		)
 return max(1     ,						UpperCAmelCase_		)

def  _snake_case		(		UpperCamelCase		:						Dict     ,						UpperCamelCase		:						int		):
 UpperCAmelCase      :   Dict      =      []
 for group_idx in range(UpperCAmelCase_		):
  UpperCAmelCase      :   int      =      num_shards // max_num_jobs + (group_idx < (num_shards % max_num_jobs))
  if num_shards_to_add == 0:
   break
  UpperCAmelCase      :   Tuple      =      shards_indices_per_group[-1].stop if shards_indices_per_group else 0
  UpperCAmelCase      :   Tuple      =      range(UpperCAmelCase_     ,						start + num_shards_to_add		)
  shards_indices_per_group.append(UpperCAmelCase_		)
 return shards_indices_per_group

def  _snake_case		(		UpperCamelCase		:						int     ,						UpperCamelCase		:						List[str]		):
 UpperCAmelCase      :   Tuple      =      _number_of_shards_in_gen_kwargs(UpperCAmelCase_		)
 if num_shards == 1:
  return [dict(UpperCAmelCase_		)]
 else:
  UpperCAmelCase      :   str      =      _distribute_shards(num_shards=UpperCAmelCase_     ,						max_num_jobs=UpperCAmelCase_		)
  return [
      {
          key: [value[shard_idx] for shard_idx in shard_indices_per_group[group_idx]]
          if isinstance(UpperCAmelCase_     ,						UpperCAmelCase_		)
          else value
          for key, value in gen_kwargs.items()
      }
      for group_idx in range(len(UpperCAmelCase_		)		)
  ]

def  _snake_case		(		UpperCamelCase		:						str		):
 return {
     key: [value for gen_kwargs in gen_kwargs_list for value in gen_kwargs[key]]
     if isinstance(gen_kwargs_list[0][key]     ,						UpperCAmelCase_		)
     else gen_kwargs_list[0][key]
     for key in gen_kwargs_list[0]
 }

def  _snake_case		(		UpperCamelCase		:						List[Any]     ,						UpperCamelCase		:						Union[str, Any]		):
 UpperCAmelCase      :   int      =      {len(UpperCAmelCase_		) for value in gen_kwargs.values() if isinstance(UpperCAmelCase_     ,						UpperCAmelCase_		)}
 UpperCAmelCase      :   Union[str, Any]      =      {}
 for size in list_sizes:
  UpperCAmelCase      :   str      =      list(range(UpperCAmelCase_		)		)
  rng.shuffle(indices_per_size[size]		)
 # Now let's copy the gen_kwargs and shuffle the lists based on their sizes
 UpperCAmelCase      :   Union[str, Any]      =      dict(UpperCAmelCase_		)
 for key, value in shuffled_kwargs.items():
  if isinstance(UpperCAmelCase_     ,						UpperCAmelCase_		):
   UpperCAmelCase      :   Dict      =      [value[i] for i in indices_per_size[len(UpperCAmelCase_		)]]
 return shuffled_kwargs





'''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


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






@add_end_docstrings(lowercase  )
class 				lowercase__      (				lowercase  ):





	def __init__(					self   :							Tuple					,**lowerCamelCase__   :							Union[str, Any]					):







							'''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   :							Dict					,lowerCamelCase__   :							Union[np.ndarray, bytes, str]					,**lowerCamelCase__   :							Any					):







							'''simple docstring'''
							return super().__call__(lowerCamelCase__					,**lowerCamelCase__					)





	def       UpperCamelCase_    (					self   :							int					,**lowerCamelCase__   :							Optional[int]					):







							'''simple docstring'''
							_UpperCamelCase   :						List[Any]												=  {}
							if "candidate_labels" in kwargs:
													_UpperCamelCase   :						Any												=  kwargs['candidate_labels']
							if "hypothesis_template" in kwargs:
													_UpperCamelCase   :						Any												=  kwargs['hypothesis_template']

							return preprocess_params, {}, {}





	def       UpperCamelCase_    (					self   :							Any					,lowerCamelCase__   :							List[str]					,lowerCamelCase__   :							str=None					,lowerCamelCase__   :							List[Any]="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
																			_UpperCamelCase   :						Optional[Any]												=  requests.get(lowerCamelCase__					).content
													else:
																			with open(lowerCamelCase__					,'rb'					) as f:
																									_UpperCamelCase   :						List[str]												=  f.read()

							if isinstance(lowerCamelCase__					,lowerCamelCase__					):
													_UpperCamelCase   :						int												=  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'					)

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





	def       UpperCamelCase_    (					self   :							str					,lowerCamelCase__   :							Tuple					):







							'''simple docstring'''
							_UpperCamelCase   :						Any												=  model_inputs.pop('candidate_labels'					)
							_UpperCamelCase   :						str												=  model_inputs.pop('text_inputs'					)
							if isinstance(text_inputs[0]					,lowerCamelCase__					):
													_UpperCamelCase   :						Union[str, Any]												=  text_inputs[0]
							else:
													# Batching case.
													_UpperCamelCase   :						int												=  text_inputs[0][0]

							_UpperCamelCase   :						List[str]												=  self.model(**lowerCamelCase__					,**lowerCamelCase__					)

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





	def       UpperCamelCase_    (					self   :							int					,lowerCamelCase__   :							Optional[Any]					):







							'''simple docstring'''
							_UpperCamelCase   :						Optional[Any]												=  model_outputs.pop('candidate_labels'					)
							_UpperCamelCase   :						int												=  model_outputs['logits'][0]

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

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


