
'''simple docstring'''



from typing import List

import numpy as np


def A (__lowerCamelCase :dict						):
						_lowerCAmelCase    		=					{key: len(__lowerCamelCase						) for key, value in gen_kwargs.items() if isinstance(__lowerCamelCase					,      __lowerCamelCase						)}
						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."""
												    )						)
						_lowerCAmelCase    		=					max(lists_lengths.values()					,      default=0						)
						return max(1					,      __lowerCamelCase						)


def A (__lowerCamelCase :int					,      __lowerCamelCase :int						):
						_lowerCAmelCase    		=					[]
						for group_idx in range(__lowerCamelCase						):
												_lowerCAmelCase    		=					num_shards // max_num_jobs + (group_idx < (num_shards % max_num_jobs))
												if num_shards_to_add == 0:
																		break
												_lowerCAmelCase    		=					shards_indices_per_group[-1].stop if shards_indices_per_group else 0
												_lowerCAmelCase    		=					range(__lowerCamelCase					,      start + num_shards_to_add						)
												shards_indices_per_group.append(__lowerCamelCase						)
						return shards_indices_per_group


def A (__lowerCamelCase :dict					,      __lowerCamelCase :int						):
						_lowerCAmelCase    		=					_number_of_shards_in_gen_kwargs(__lowerCamelCase						)
						if num_shards == 1:
												return [dict(__lowerCamelCase						)]
						else:
												_lowerCAmelCase    		=					_distribute_shards(num_shards=__lowerCamelCase					,      max_num_jobs=__lowerCamelCase						)
												return [
												    {
												        key: [value[shard_idx] for shard_idx in shard_indices_per_group[group_idx]]
												        if isinstance(__lowerCamelCase					,      __lowerCamelCase						)
												        else value
												        for key, value in gen_kwargs.items()
												    }
												    for group_idx in range(len(__lowerCamelCase						)						)
												]


def A (__lowerCamelCase :List[dict]						):
						return {
						    key: [value for gen_kwargs in gen_kwargs_list for value in gen_kwargs[key]]
						    if isinstance(gen_kwargs_list[0][key]					,      __lowerCamelCase						)
						    else gen_kwargs_list[0][key]
						    for key in gen_kwargs_list[0]
						}





def A (__lowerCamelCase :np.random.Generator					,      __lowerCamelCase :dict						):
						_lowerCAmelCase    		=					{len(__lowerCamelCase						) for value in gen_kwargs.values() if isinstance(__lowerCamelCase					,      __lowerCamelCase						)}
						_lowerCAmelCase    		=					{}
						for size in list_sizes:
												_lowerCAmelCase    		=					list(range(__lowerCamelCase						)						)
												rng.shuffle(indices_per_size[size]						)
						# Now let's copy the gen_kwargs and shuffle the lists based on their sizes
						_lowerCAmelCase    		=					dict(__lowerCamelCase						)
						for key, value in shuffled_kwargs.items():
												if isinstance(__lowerCamelCase					,      __lowerCamelCase						):
																		_lowerCAmelCase    		=					[value[i] for i in indices_per_size[len(__lowerCamelCase						)]]
						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


_lowercase          =						logging.get_logger(__name__)




@add_end_docstrings(_SCREAMING_SNAKE_CASE					)
class     UpperCAmelCase_	(						_SCREAMING_SNAKE_CASE					):






						'''simple docstring'''






						def __init__(   self       ,  **_lowercase ):

												"""simple docstring"""



												super().__init__(**_lowercase )

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




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

												"""simple docstring"""



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




						def    _lowercase      (   self       ,  **_lowercase ):

												"""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       ,  _lowercase       ,  _lowercase=None       ,  _lowercase="This is a sound of {}." ):

												"""simple docstring"""



												if isinstance(_lowercase       ,  _lowercase ):
																		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(_lowercase ).content
																		else:
																								with open(_lowercase       ,  """rb""" ) as f:
																														_lowerCAmelCase    		=					f.read()

												if isinstance(_lowercase       ,  _lowercase ):
																		_lowerCAmelCase    		=					ffmpeg_read(_lowercase       ,  self.feature_extractor.sampling_rate )

												if not isinstance(_lowercase       ,  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(_lowercase ) for x in candidate_labels]
												_lowerCAmelCase    		=					self.tokenizer(_lowercase       ,  return_tensors=self.framework       ,  padding=_lowercase )
												_lowerCAmelCase    		=					[text_inputs]
												return inputs




						def    _lowercase      (   self       ,  _lowercase ):

												"""simple docstring"""



												_lowerCAmelCase    		=					model_inputs.pop("""candidate_labels""" )
												_lowerCAmelCase    		=					model_inputs.pop("""text_inputs""" )
												if isinstance(text_inputs[0]       ,  _lowercase ):
																		_lowerCAmelCase    		=					text_inputs[0]
												else:
																		# Batching case.
																		_lowerCAmelCase    		=					text_inputs[0][0]

												_lowerCAmelCase    		=					self.model(**_lowercase       ,  **_lowercase )

												_lowerCAmelCase    		=					{
												    """candidate_labels""": candidate_labels,
												    """logits""": outputs.logits_per_audio,
												}
												return model_outputs




						def    _lowercase      (   self       ,  _lowercase ):

												"""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(_lowercase       ,  _lowercase )       ,  key=lambda _lowercase : -x[0] )
												]
												return result
