






'''simple docstring'''







import unicodedata
from dataclasses import dataclass
from typing import Optional, Union

import numpy as np

from transformers.data.data_collator import DataCollatorMixin
from transformers.file_utils import PaddingStrategy
from transformers.tokenization_utils_base import PreTrainedTokenizerBase


def 			SCREAMING_SNAKE_CASE_	(UpperCamelCase    ,      UpperCamelCase    ,      UpperCamelCase    ,      UpperCamelCase				)				->	List[str]:
							if isinstance(__lowerCAmelCase    ,      __lowerCAmelCase				):
														lowerCamelCase__ :     Any	  =	np.full((len(__lowerCAmelCase				), sequence_length, 2)    ,      __lowerCAmelCase				)
							else:
														lowerCamelCase__ :     Optional[int]	  =	np.full((len(__lowerCAmelCase				), sequence_length)    ,      __lowerCAmelCase				)

							for i, tensor in enumerate(__lowerCAmelCase				):
														if padding_side == "right":
																					if isinstance(__lowerCAmelCase    ,      __lowerCAmelCase				):
																												lowerCamelCase__ :     Tuple	  =	tensor[:sequence_length]
																					else:
																												lowerCamelCase__ :     Any	  =	tensor[:sequence_length]
														else:
																					if isinstance(__lowerCAmelCase    ,      __lowerCAmelCase				):
																												lowerCamelCase__ :     List[Any]	  =	tensor[:sequence_length]
																					else:
																												lowerCamelCase__ :     List[Any]	  =	tensor[:sequence_length]

							return out_tensor.tolist()


def 			SCREAMING_SNAKE_CASE_	(UpperCamelCase				)				->	Dict:
							lowerCamelCase__ :     Any	  =	ord(__lowerCAmelCase				)
							if (cp >= 33 and cp <= 47) or (cp >= 58 and cp <= 64) or (cp >= 91 and cp <= 96) or (cp >= 123 and cp <= 126):
														return True
							lowerCamelCase__ :     List[Any]	  =	unicodedata.category(__lowerCAmelCase				)
							if cat.startswith("""P"""				):
														return True
							return False



@dataclass
class _lowercase     (    UpperCamelCase_		):
							a    				=	42
							a    				=	True
							a    				=	None
							a    				=	None
							a    				=	-100
							a    				=	"pt"




							def 		lowerCamelCase_    (							self:    Optional[int]      ,							UpperCamelCase__:    str					):
														import torch

														lowerCamelCase__ :     Tuple	  =	"""label""" if """label""" in features[0].keys() else """labels"""
														lowerCamelCase__ :     int	  =	[feature[label_name] for feature in features] if label_name in features[0].keys() else None
														lowerCamelCase__ :     List[str]	  =	self.tokenizer.pad(
														    _a      ,							padding=self.padding      ,							max_length=self.max_length      ,							pad_to_multiple_of=self.pad_to_multiple_of      ,							return_tensors="""pt""" if labels is None else None      ,							)

														if labels is None:
																					return batch

														lowerCamelCase__ :     List[Any]	  =	torch.tensor(batch["""entity_ids"""]					).shape[1]
														lowerCamelCase__ :     Dict	  =	self.tokenizer.padding_side
														if padding_side == "right":
																					lowerCamelCase__ :     Any	  =	[
																					    list(_a					) + [self.label_pad_token_id] * (sequence_length - len(_a					)) for label in labels
																					]
														else:
																					lowerCamelCase__ :     List[str]	  =	[
																					    [self.label_pad_token_id] * (sequence_length - len(_a					)) + list(_a					) for label in labels
																					]

														lowerCamelCase__ :     List[str]	  =	[feature["""ner_tags"""] for feature in features]
														lowerCamelCase__ :     List[str]	  =	padding_tensor(_a      ,							-1      ,							_a      ,							_a					)
														lowerCamelCase__ :     Tuple	  =	[feature["""original_entity_spans"""] for feature in features]
														lowerCamelCase__ :     Tuple	  =	padding_tensor(_a      ,							(-1, -1)      ,							_a      ,							_a					)
														lowerCamelCase__ :     Optional[Any]	  =	{k: torch.tensor(_a      ,							dtype=torch.intaa					) for k, v in batch.items()}

														return batch









'''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     :      Optional[Any]   					=logging.get_logger(__name__)

@add_end_docstrings(_lowercase		)
class _lowercase     (    _lowercase		):




							def __init__(							self:    Union[str, Any]      ,							**UpperCamelCase__:    str					):
														super().__init__(**UpperCamelCase__					)

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




							def __call__(							self:    Optional[Any]      ,							UpperCamelCase__:    Union[np.ndarray, bytes, str]      ,							**UpperCamelCase__:    int					):
														return super().__call__(UpperCamelCase__      ,							**UpperCamelCase__					)




							def 		lowerCamelCase_    (							self:    Optional[Any]      ,							**UpperCamelCase__:    int					):
														lowerCamelCase__ :     Optional[Any]	  =	{}
														if "candidate_labels" in kwargs:
																					lowerCamelCase__ :     str	  =	kwargs["""candidate_labels"""]
														if "hypothesis_template" in kwargs:
																					lowerCamelCase__ :     int	  =	kwargs["""hypothesis_template"""]

														return preprocess_params, {}, {}




							def 		lowerCamelCase_    (							self:    Optional[Any]      ,							UpperCamelCase__:    int      ,							UpperCamelCase__:    List[str]=None      ,							UpperCamelCase__:    Dict="This is a sound of {}."					):
														if isinstance(UpperCamelCase__      ,							UpperCamelCase__					):
																					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__ :     int	  =	requests.get(UpperCamelCase__					).content
																					else:
																												with open(UpperCamelCase__      ,							"""rb"""					) as f:
																																			lowerCamelCase__ :     Dict	  =	f.read()

														if isinstance(UpperCamelCase__      ,							UpperCamelCase__					):
																					lowerCamelCase__ :     str	  =	ffmpeg_read(UpperCamelCase__      ,							self.feature_extractor.sampling_rate					)

														if not isinstance(UpperCamelCase__      ,							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__ :     Optional[Any]	  =	self.feature_extractor(
														    [audio]      ,							sampling_rate=self.feature_extractor.sampling_rate      ,							return_tensors="""pt"""					)
														lowerCamelCase__ :     Any	  =	candidate_labels
														lowerCamelCase__ :     Any	  =	[hypothesis_template.format(UpperCamelCase__					) for x in candidate_labels]
														lowerCamelCase__ :     int	  =	self.tokenizer(UpperCamelCase__      ,							return_tensors=self.framework      ,							padding=UpperCamelCase__					)
														lowerCamelCase__ :     Tuple	  =	[text_inputs]
														return inputs




							def 		lowerCamelCase_    (							self:    Tuple      ,							UpperCamelCase__:    Union[str, Any]					):
														lowerCamelCase__ :     Any	  =	model_inputs.pop("""candidate_labels"""					)
														lowerCamelCase__ :     Union[str, Any]	  =	model_inputs.pop("""text_inputs"""					)
														if isinstance(text_inputs[0]      ,							UpperCamelCase__					):
																					lowerCamelCase__ :     Optional[int]	  =	text_inputs[0]
														else:
																					# Batching case.
																					lowerCamelCase__ :     Tuple	  =	text_inputs[0][0]

														lowerCamelCase__ :     Tuple	  =	self.model(**UpperCamelCase__      ,							**UpperCamelCase__					)

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




							def 		lowerCamelCase_    (							self:    List[str]      ,							UpperCamelCase__:    Union[str, Any]					):
														lowerCamelCase__ :     List[str]	  =	model_outputs.pop("""candidate_labels"""					)
														lowerCamelCase__ :     int	  =	model_outputs["""logits"""][0]

														if self.framework == "pt":
																					lowerCamelCase__ :     Optional[int]	  =	logits.softmax(dim=0					)
																					lowerCamelCase__ :     Any	  =	probs.tolist()
														else:
																					raise ValueError("""`tf` framework not supported."""					)

														lowerCamelCase__ :     Tuple	  =	[
														    {"""score""": score, """label""": candidate_label}
														    for score, candidate_label in sorted(zip(UpperCamelCase__      ,							UpperCamelCase__					)      ,							key=lambda UpperCamelCase__					: -x[0]					)
														]
														return result

