
'''simple docstring'''



from __future__ import annotations

from collections.abc import Callable
from typing import Generic, TypeVar

_lowercase          =						TypeVar("""T""")
_lowercase          =						TypeVar("""U""")




class     UpperCAmelCase_	(						Generic[T, U]					):






						'''simple docstring'''






						def __init__(   self       ,  _lowercase       ,  _lowercase ):

												"""simple docstring"""



												_lowerCAmelCase    		=					key
												_lowerCAmelCase    		=					val
												_lowerCAmelCase    		=					None
												_lowerCAmelCase    		=					None




						def __repr__(   self ):

												"""simple docstring"""



												return (
												    F'Node: key: {self.key}, val: {self.val}, '
												    F'has next: {bool(self.next )}, has prev: {bool(self.prev )}'
												)






class     UpperCAmelCase_	(						Generic[T, U]					):






						'''simple docstring'''






						def __init__(   self ):

												"""simple docstring"""



												_lowerCAmelCase    		=					DoubleLinkedListNode(_lowercase       ,  _lowercase )
												_lowerCAmelCase    		=					DoubleLinkedListNode(_lowercase       ,  _lowercase )
												_lowerCAmelCase		,       _lowerCAmelCase    		=					self.rear, self.head




						def __repr__(   self ):

												"""simple docstring"""



												_lowerCAmelCase    		=					["""DoubleLinkedList"""]
												_lowerCAmelCase    		=					self.head
												while node.next is not None:
																		rep.append(str(_lowercase ) )
																		_lowerCAmelCase    		=					node.next
												rep.append(str(self.rear ) )
												return ",\n    ".join(_lowercase )




						def    _lowercase      (   self       ,  _lowercase ):

												"""simple docstring"""



												_lowerCAmelCase    		=					self.rear.prev

												# All nodes other than self.head are guaranteed to have non-None previous
												assert previous is not None

												_lowerCAmelCase    		=					node
												_lowerCAmelCase    		=					previous
												_lowerCAmelCase    		=					node
												_lowerCAmelCase    		=					self.rear




						def    _lowercase      (   self       ,  _lowercase ):

												"""simple docstring"""



												if node.prev is None or node.next is None:
																		return None

												_lowerCAmelCase    		=					node.next
												_lowerCAmelCase    		=					node.prev
												_lowerCAmelCase    		=					None
												_lowerCAmelCase    		=					None
												return node






class     UpperCAmelCase_	(						Generic[T, U]					):






						'''simple docstring'''






						_lowercase						: dict[Callable[[T], U], LRUCache[T, U]]										=  {}




						def __init__(   self       ,  _lowercase ):

												"""simple docstring"""



												_lowerCAmelCase    		=					DoubleLinkedList()
												_lowerCAmelCase    		=					capacity
												_lowerCAmelCase    		=					0
												_lowerCAmelCase    		=					0
												_lowerCAmelCase    		=					0
												_lowerCAmelCase    		=					{}




						def __repr__(   self ):

												"""simple docstring"""



												return (
												    F'CacheInfo(hits={self.hits}, misses={self.miss}, '
												    F'capacity={self.capacity}, current size={self.num_keys})'
												)




						def __contains__(   self       ,  _lowercase ):

												"""simple docstring"""



												return key in self.cache




						def    _lowercase      (   self       ,  _lowercase ):

												"""simple docstring"""



												if key in self.cache:
																		self.hits += 1
																		_lowerCAmelCase    		=					self.cache[key]
																		_lowerCAmelCase    		=					self.list.remove(self.cache[key] )
																		assert node == value_node

																		# node is guaranteed not None because it is in self.cache
																		assert node is not None
																		self.list.add(_lowercase )
																		return node.val
												self.miss += 1
												return None




						def    _lowercase      (   self       ,  _lowercase       ,  _lowercase ):

												"""simple docstring"""



												if key not in self.cache:
																		if self.num_keys >= self.capacity:
																								# delete first node (oldest) when over capacity
																								_lowerCAmelCase    		=					self.list.head.next

																								# guaranteed to have a non-None first node when num_keys > 0
																								# explain to type checker via assertions
																								assert first_node is not None
																								assert first_node.key is not None
																								assert (
																								    self.list.remove(_lowercase ) is not None
																								)  # node guaranteed to be in list assert node.key is not None

																								del self.cache[first_node.key]
																								self.num_keys -= 1
																		_lowerCAmelCase    		=					DoubleLinkedListNode(_lowercase       ,  _lowercase )
																		self.list.add(self.cache[key] )
																		self.num_keys += 1

												else:
																		# bump node to the end of the list, update value
																		_lowerCAmelCase    		=					self.list.remove(self.cache[key] )
																		assert node is not None  # node guaranteed to be in list
																		_lowerCAmelCase    		=					value
																		self.list.add(_lowercase )




						@classmethod
						def    _lowercase      (   cls       ,  _lowercase = 128 ):

												"""simple docstring"""



												def cache_decorator_inner(_lowercase ) -> Callable[..., U]:
																		def cache_decorator_wrapper(*_lowercase ) -> U:
																								if func not in cls.decorator_function_to_instance_map:
																														_lowerCAmelCase    		=					LRUCache(_lowercase )

																								_lowerCAmelCase    		=					cls.decorator_function_to_instance_map[func].get(args[0] )
																								if result is None:
																														_lowerCAmelCase    		=					func(*_lowercase )
																														cls.decorator_function_to_instance_map[func].put(args[0]       ,  _lowercase )
																								return result

																		def cache_info() -> LRUCache[T, U]:
																								return cls.decorator_function_to_instance_map[func]

																		setattr(_lowercase       ,  """cache_info"""       ,  _lowercase )  # noqa: B010

																		return cache_decorator_wrapper

												return cache_decorator_inner


if __name__ == "__main__":
						import doctest

						doctest.testmod()


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