






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


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





@add_end_docstrings(_lowerCamelCase       )
class       snake_case_       (    _lowerCamelCase       ):



	"""simple docstring"""

	def __init__(      self  ,					**__a						):


					"""simple docstring"""

					super().__init__(**__a						)

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


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


					"""simple docstring"""

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


	def       _UpperCAmelCase						(      self  ,					**__a						):


					"""simple docstring"""

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

					return preprocess_params, {}, {}


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


					"""simple docstring"""

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

					if isinstance(__a  ,					__a						):
									A__							 =       ffmpeg_read(__a  ,					self.feature_extractor.sampling_rate						)

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

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


	def       _UpperCAmelCase						(      self  ,					__a						):


					"""simple docstring"""

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

					A__							 =       self.model(**__a  ,					**__a						)

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






	def       _UpperCAmelCase						(      self  ,					__a						):


					"""simple docstring"""

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

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

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










"""simple docstring"""






from math import ceil



def  __lowerCamelCase					(					lowerCAmelCase__       ,lowerCAmelCase__				):
				A__							 =       list(range(0       ,lowerCAmelCase__				)				)

				A__							 =       [item for sublist in list(device_map.values()				) for item in sublist]

				# Duplicate check
				A__							 =       []
				for i in device_map_blocks:
								if device_map_blocks.count(lowerCAmelCase__				) > 1 and i not in duplicate_blocks:
												duplicate_blocks.append(lowerCAmelCase__				)
    # Missing blocks
				A__							 =       [i for i in blocks if i not in device_map_blocks]
				A__							 =       [i for i in device_map_blocks if i not in blocks]

				if len(lowerCAmelCase__				) != 0:
								raise ValueError(
								    'Duplicate attention blocks specified in device_map. Attention blocks must be specified to one device.'
								    ' These attention blocks were specified more than once: ' + str(lowerCAmelCase__				)				)
				if len(lowerCAmelCase__				) != 0:
								raise ValueError(
								    'There are attention blocks for this model that are not specified in the device_map. Add these attention '
								    'blocks to a device on the device_map: ' + str(lowerCAmelCase__				)				)
				if len(lowerCAmelCase__				) != 0:
								raise ValueError(
								    'The device_map contains more attention blocks than this model has. Remove these from the device_map:'
								    + str(lowerCAmelCase__				)				)







def  __lowerCamelCase					(					lowerCAmelCase__       ,lowerCAmelCase__				):
				A__							 =       list(range(lowerCAmelCase__				)				)
				A__							 =       int(ceil(n_layers / len(lowerCAmelCase__				)				)				)
				A__							 =       [layers[i : i + n_blocks] for i in range(0       ,lowerCAmelCase__       ,lowerCAmelCase__				)]

				return dict(zip(lowerCAmelCase__       ,lowerCAmelCase__				)				)


