



'''simple docstring'''



from typing import List, Union

import numpy as np

from ..tokenization_utils import TruncationStrategy
from ..utils import add_end_docstrings, logging
from .base import PIPELINE_INIT_ARGS, ArgumentHandler, ChunkPipeline


a__    :      Optional[Any]      				=      logging.get_logger(__name__)
class 							UpperCAmelCase__	(     SCREAMING_SNAKE_CASE_):



 def  __lowerCamelCase					(	self		,						lowercase      )	->				List[Any]:
  if isinstance(__snake_case		,						__snake_case      ):
   __UpperCamelCase					   =	[label.strip() for label in labels.split(""","""      ) if label.strip()]
  return labels




 def __call__(	self		,						lowercase		,						lowercase		,						lowercase      )	->				int:
  if len(__snake_case      ) == 0 or len(__snake_case      ) == 0:
   raise ValueError("""You must include at least one label and at least one sequence."""      )
  if hypothesis_template.format(labels[0]      ) == hypothesis_template:
   raise ValueError(
       (
           """The provided hypothesis_template \"{}\" was not able to be formatted with the target labels. """
           """Make sure the passed template includes formatting syntax such as {{}} where the label should go."""
       ).format(__snake_case      )      )

  if isinstance(__snake_case		,						__snake_case      ):
   __UpperCamelCase					   =	[sequences]

  __UpperCamelCase					   =	[]
  for sequence in sequences:
   sequence_pairs.extend([[sequence, hypothesis_template.format(__snake_case      )] for label in labels]      )

  return sequence_pairs, sequences


@add_end_docstrings(SCREAMING_SNAKE_CASE_)
class 							UpperCAmelCase__	(     SCREAMING_SNAKE_CASE_):



 def __init__(	self		,						lowercase=ZeroShotClassificationArgumentHandler()		,						*lowercase		,						**lowercase      )	->				int:
  __UpperCamelCase					   =	args_parser
  super().__init__(*__snake_case		,						**__snake_case      )
  if self.entailment_id == -1:
   logger.warning(
       """Failed to determine 'entailment' label id from the label2id mapping in the model config. Setting to """
       """-1. Define a descriptive label2id mapping in the model config to ensure correct outputs."""      )



 @property
 def  __lowerCamelCase					(	self      )	->				str:
  for label, ind in self.model.config.labelaid.items():
   if label.lower().startswith("""entail"""      ):
    return ind
  return -1



 def  __lowerCamelCase					(	self		,						lowercase		,						lowercase=True		,						lowercase=True		,						lowercase=TruncationStrategy.ONLY_FIRST		,						**lowercase      )	->				int:
  __UpperCamelCase					   =	self.framework
  if self.tokenizer.pad_token is None:
   # Override for tokenizers not supporting padding
   logger.error(
       """Tokenizer was not supporting padding necessary for zero-shot, attempting to use """
       """ `pad_token=eos_token`"""      )
   __UpperCamelCase					   =	self.tokenizer.eos_token
  try:
   __UpperCamelCase					   =	self.tokenizer(
       __snake_case		,						add_special_tokens=__snake_case		,						return_tensors=__snake_case		,						padding=__snake_case		,						truncation=__snake_case		,						)
  except Exception as e:
   if "too short" in str(__snake_case      ):
    # tokenizers might yell that we want to truncate
    # to a value that is not even reached by the input.
    # In that case we don't want to truncate.
    # It seems there's not a really better way to catch that
    # exception.

    __UpperCamelCase					   =	self.tokenizer(
        __snake_case		,						add_special_tokens=__snake_case		,						return_tensors=__snake_case		,						padding=__snake_case		,						truncation=TruncationStrategy.DO_NOT_TRUNCATE		,						)
   else:
    raise e

  return inputs



 def  __lowerCamelCase					(	self		,						**lowercase      )	->				int:
  if kwargs.get("""multi_class"""		,						__snake_case      ) is not None:
   __UpperCamelCase					   =	kwargs["""multi_class"""]
   logger.warning(
       """The `multi_class` argument has been deprecated and renamed to `multi_label`. """
       """`multi_class` will be removed in a future version of Transformers."""      )
  __UpperCamelCase					   =	{}
  if "candidate_labels" in kwargs:
   __UpperCamelCase					   =	self._args_parser._parse_labels(kwargs["""candidate_labels"""]      )
  if "hypothesis_template" in kwargs:
   __UpperCamelCase					   =	kwargs["""hypothesis_template"""]

  __UpperCamelCase					   =	{}
  if "multi_label" in kwargs:
   __UpperCamelCase					   =	kwargs["""multi_label"""]
  return preprocess_params, {}, postprocess_params



 def __call__(	self		,						lowercase		,						*lowercase		,						**lowercase		,						)	->				List[str]:
  if len(__snake_case      ) == 0:
   pass
  elif len(__snake_case      ) == 1 and "candidate_labels" not in kwargs:
   __UpperCamelCase					   =	args[0]
  else:
   raise ValueError(f"Unable to understand extra arguments {args}"      )

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



 def  __lowerCamelCase					(	self		,						lowercase		,						lowercase=None		,						lowercase="This example is {}."      )	->				List[Any]:
  __UpperCamelCase					   =	self._args_parser(__snake_case		,						__snake_case		,						__snake_case      )

  for i, (candidate_label, sequence_pair) in enumerate(zip(__snake_case		,						__snake_case      )      ):
   __UpperCamelCase					   =	self._parse_and_tokenize([sequence_pair]      )

   yield {
       "candidate_label": candidate_label,
       "sequence": sequences[0],
       "is_last": i == len(__snake_case      ) - 1,
       **model_input,
   }



 def  __lowerCamelCase					(	self		,						lowercase      )	->				Tuple:
  __UpperCamelCase					   =	inputs["""candidate_label"""]
  __UpperCamelCase					   =	inputs["""sequence"""]
  __UpperCamelCase					   =	{k: inputs[k] for k in self.tokenizer.model_input_names}
  __UpperCamelCase					   =	self.model(**__snake_case      )

  __UpperCamelCase					   =	{
      """candidate_label""": candidate_label,
      """sequence""": sequence,
      """is_last""": inputs["""is_last"""],
      **outputs,
  }
  return model_outputs




 def  __lowerCamelCase					(	self		,						lowercase		,						lowercase=False      )	->				Any:
  __UpperCamelCase					   =	[outputs["""candidate_label"""] for outputs in model_outputs]
  __UpperCamelCase					   =	[outputs["""sequence"""] for outputs in model_outputs]
  __UpperCamelCase					   =	np.concatenate([output["""logits"""].numpy() for output in model_outputs]      )
  __UpperCamelCase					   =	logits.shape[0]
  __UpperCamelCase					   =	len(__snake_case      )
  __UpperCamelCase					   =	N // n
  __UpperCamelCase					   =	logits.reshape((num_sequences, n, -1)      )

  if multi_label or len(__snake_case      ) == 1:
   # softmax over the entailment vs. contradiction dim for each label independently
   __UpperCamelCase					   =	self.entailment_id
   __UpperCamelCase					   =	-1 if entailment_id == 0 else 0
   __UpperCamelCase					   =	reshaped_outputs[..., [contradiction_id, entailment_id]]
   __UpperCamelCase					   =	np.exp(__snake_case      ) / np.exp(__snake_case      ).sum(-1		,						keepdims=__snake_case      )
   __UpperCamelCase					   =	scores[..., 1]
  else:
   # softmax the "entailment" logits over all candidate labels
   __UpperCamelCase					   =	reshaped_outputs[..., self.entailment_id]
   __UpperCamelCase					   =	np.exp(__snake_case      ) / np.exp(__snake_case      ).sum(-1		,						keepdims=__snake_case      )

  __UpperCamelCase					   =	list(reversed(scores[0].argsort()      )      )
  return {
      "sequence": sequences[0],
      "labels": [candidate_labels[i] for i in top_inds],
      "scores": scores[0, top_inds].tolist(),
  }













"""simple docstring"""





import datetime
import platform
import subprocess
from typing import Optional, Tuple, Union

import numpy as np



def   a    (     __UpperCAmelCase      :    bytes      ,   __UpperCAmelCase      :    int  )    ->							np.array:
						__magic_name__:	Optional[int]							=       f'{sampling_rate}'
						__magic_name__:	Union[str, Any]							=       """1"""
						__magic_name__:	Tuple							=       """f32le"""
						__magic_name__:	Optional[Any]							=       [
						    """ffmpeg""",
						    """-i""",
						    """pipe:0""",
						    """-ac""",
						    ac,
						    """-ar""",
						    ar,
						    """-f""",
						    format_for_conversion,
						    """-hide_banner""",
						    """-loglevel""",
						    """quiet""",
						    """pipe:1""",
						]

						try:
												with subprocess.Popen(__UpperCAmelCase      ,   stdin=subprocess.PIPE      ,   stdout=subprocess.PIPE  ) as ffmpeg_process:
																		__magic_name__:	Any							=       ffmpeg_process.communicate(__UpperCAmelCase  )
						except FileNotFoundError as error:
												raise ValueError("""ffmpeg was not found but is required to load audio files from filename"""  ) from error
						__magic_name__:	Optional[Any]							=       output_stream[0]
						__magic_name__:	Optional[int]							=       np.frombuffer(__UpperCAmelCase      ,   np.floataa  )
						if audio.shape[0] == 0:
												raise ValueError("""Malformed soundfile"""  )
						return audio



def   a    (     __UpperCAmelCase      :    int      ,   __UpperCAmelCase      :    float      ,   __UpperCAmelCase      :    str = "f32le"      ,   )    ->							Optional[int]:
						__magic_name__:	List[Any]							=       f'{sampling_rate}'
						__magic_name__:	Optional[int]							=       """1"""
						if format_for_conversion == "s16le":
												__magic_name__:	Dict							=       2
						elif format_for_conversion == "f32le":
												__magic_name__:	List[str]							=       4
						else:
												raise ValueError(f'Unhandled format `{format_for_conversion}`. Please use `s16le` or `f32le`'  )

						__magic_name__:	str							=       platform.system()
						if system == "Linux":
												__magic_name__:	Optional[Any]							=       """alsa"""
												__magic_name__:	Optional[Any]							=       """default"""
						elif system == "Darwin":
												__magic_name__:	Union[str, Any]							=       """avfoundation"""
												__magic_name__:	Dict							=       """:0"""
						elif system == "Windows":
												__magic_name__:	Any							=       """dshow"""
												__magic_name__:	Union[str, Any]							=       """default"""

						__magic_name__:	Union[str, Any]							=       [
						    """ffmpeg""",
						    """-f""",
						    format_,
						    """-i""",
						    input_,
						    """-ac""",
						    ac,
						    """-ar""",
						    ar,
						    """-f""",
						    format_for_conversion,
						    """-fflags""",
						    """nobuffer""",
						    """-hide_banner""",
						    """-loglevel""",
						    """quiet""",
						    """pipe:1""",
						]
						__magic_name__:	Tuple							=       int(round(sampling_rate * chunk_length_s  )  ) * size_of_sample
						__magic_name__:	Union[str, Any]							=       _ffmpeg_stream(__UpperCAmelCase      ,   __UpperCAmelCase  )
						for item in iterator:
												yield item



def   a    (     __UpperCAmelCase      :    int      ,   __UpperCAmelCase      :    float      ,   __UpperCAmelCase      :    Optional[int] = None      ,   __UpperCAmelCase      :    Optional[Union[Tuple[float, float], float]] = None      ,   __UpperCAmelCase      :    str = "f32le"      ,   )    ->							List[Any]:
						if stream_chunk_s is not None:
												__magic_name__:	Optional[int]							=       stream_chunk_s
						else:
												__magic_name__:	List[str]							=       chunk_length_s

						__magic_name__:	Optional[int]							=       ffmpeg_microphone(__UpperCAmelCase      ,   __UpperCAmelCase      ,   format_for_conversion=__UpperCAmelCase  )
						if format_for_conversion == "s16le":
												__magic_name__:	str							=       np.intaa
												__magic_name__:	str							=       2
						elif format_for_conversion == "f32le":
												__magic_name__:	Optional[Any]							=       np.floataa
												__magic_name__:	Dict							=       4
						else:
												raise ValueError(f'Unhandled format `{format_for_conversion}`. Please use `s16le` or `f32le`'  )

						if stride_length_s is None:
												__magic_name__:	List[Any]							=       chunk_length_s / 6
						__magic_name__:	Optional[Any]							=       int(round(sampling_rate * chunk_length_s  )  ) * size_of_sample
						if isinstance(__UpperCAmelCase      ,   (int, float)  ):
												__magic_name__:	Union[str, Any]							=       [stride_length_s, stride_length_s]

						__magic_name__:	List[str]							=       int(round(sampling_rate * stride_length_s[0]  )  ) * size_of_sample
						__magic_name__:	List[Any]							=       int(round(sampling_rate * stride_length_s[1]  )  ) * size_of_sample
						__magic_name__:	List[str]							=       datetime.datetime.now()
						__magic_name__:	Optional[Any]							=       datetime.timedelta(seconds=__UpperCAmelCase  )
						for item in chunk_bytes_iter(__UpperCAmelCase      ,   __UpperCAmelCase      ,   stride=(stride_left, stride_right)      ,   stream=__UpperCAmelCase  ):
												# Put everything back in numpy scale
												__magic_name__:	int							=       np.frombuffer(item["""raw"""]      ,   dtype=__UpperCAmelCase  )
												__magic_name__:	int							=       (
												    item["""stride"""][0] // size_of_sample,
												    item["""stride"""][1] // size_of_sample,
												)
												__magic_name__:	Union[str, Any]							=       sampling_rate
												audio_time += delta
												if datetime.datetime.now() > audio_time + 1_0 * delta:
																		# We're late !! SKIP
																		continue
												yield item



def   a    (     __UpperCAmelCase      :    int      ,   __UpperCAmelCase      :    int      ,   __UpperCAmelCase      :    Tuple[int, int]      ,   __UpperCAmelCase      :    bool = False  )    ->							Union[str, Any]:
						__magic_name__:	Tuple							=       B""""""
						__magic_name__,			__magic_name__:	int							=       stride
						if stride_left + stride_right >= chunk_len:
												raise ValueError(
												    f'Stride needs to be strictly smaller than chunk_len: ({stride_left}, {stride_right}) vs {chunk_len}'  )
						__magic_name__:	int							=       0
						for raw in iterator:
												acc += raw
												if stream and len(__UpperCAmelCase  ) < chunk_len:
																		__magic_name__:	Optional[int]							=       (_stride_left, 0)
																		yield {"raw": acc[:chunk_len], "stride": stride, "partial": True}
												else:
																		while len(__UpperCAmelCase  ) >= chunk_len:
																								# We are flushing the accumulator
																								__magic_name__:	Union[str, Any]							=       (_stride_left, stride_right)
																								__magic_name__:	Any							=       {"""raw""": acc[:chunk_len], """stride""": stride}
																								if stream:
																														__magic_name__:	str							=       False
																								yield item
																								__magic_name__:	str							=       stride_left
																								__magic_name__:	Optional[Any]							=       acc[chunk_len - stride_left - stride_right :]
    # Last chunk
						if len(__UpperCAmelCase  ) > stride_left:
												__magic_name__:	List[Any]							=       {"""raw""": acc, """stride""": (_stride_left, 0)}
												if stream:
																		__magic_name__:	Union[str, Any]							=       False
												yield item



def   a    (     __UpperCAmelCase      :    Tuple      ,   __UpperCAmelCase      :    int  )    ->							List[Any]:
						__magic_name__:	str							=       2**2_4  # 16Mo
						try:
												with subprocess.Popen(__UpperCAmelCase      ,   stdout=subprocess.PIPE      ,   bufsize=__UpperCAmelCase  ) as ffmpeg_process:
																		while True:
																								__magic_name__:	Dict							=       ffmpeg_process.stdout.read(__UpperCAmelCase  )
																								if raw == b"":
																														break
																								yield raw
						except FileNotFoundError as error:
												raise ValueError("""ffmpeg was not found but is required to stream audio files from filename"""  ) from error



