



"""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__		:    List[str]				   =     logging.get_logger(__name__)







@add_end_docstrings(__SCREAMING_SNAKE_CASE  )
class 				UpperCamelCase__       (     __SCREAMING_SNAKE_CASE  ):




      """simple docstring"""




      def __init__(				self    :				int   ,      **SCREAMING_SNAKE_CASE_    :				Tuple	):
        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    :				Any   ,      SCREAMING_SNAKE_CASE_    :				Tuple   ,      **SCREAMING_SNAKE_CASE_    :				str	):
        return super().__call__(_a   ,      **_a	)



      def 					SCREAMING_SNAKE_CASE__			(				self    :				Union[str, Any]   ,      **SCREAMING_SNAKE_CASE_    :				Any	):
        lowerCAmelCase_       :    Tuple								=			{}
        if "candidate_labels" in kwargs:
          lowerCAmelCase_       :    Dict								=			kwargs['candidate_labels']
        if "hypothesis_template" in kwargs:
          lowerCAmelCase_       :    Dict								=			kwargs['hypothesis_template']

        return preprocess_params, {}, {}



      def 					SCREAMING_SNAKE_CASE__			(				self    :				Optional[int]   ,      SCREAMING_SNAKE_CASE_    :				Optional[int]   ,      SCREAMING_SNAKE_CASE_    :				Any=None   ,      SCREAMING_SNAKE_CASE_    :				str="This is a sound of {}."	):
        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
            lowerCAmelCase_       :    Optional[int]								=			requests.get(_a	).content
          else:
            with open(_a   ,      'rb'	) as f:
              lowerCAmelCase_       :    Union[str, Any]								=			f.read()

        if isinstance(_a   ,      _a	):
          lowerCAmelCase_       :    int								=			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'	)

        lowerCAmelCase_       :    Optional[int]								=			self.feature_extractor(
            [audio]   ,      sampling_rate=self.feature_extractor.sampling_rate   ,      return_tensors='pt'	)
        lowerCAmelCase_       :    Dict								=			candidate_labels
        lowerCAmelCase_       :    Any								=			[hypothesis_template.format(_a	) for x in candidate_labels]
        lowerCAmelCase_       :    List[Any]								=			self.tokenizer(_a   ,      return_tensors=self.framework   ,      padding=_a	)
        lowerCAmelCase_       :    str								=			[text_inputs]
        return inputs



      def 					SCREAMING_SNAKE_CASE__			(				self    :				List[Any]   ,      SCREAMING_SNAKE_CASE_    :				Tuple	):
        lowerCAmelCase_       :    str								=			model_inputs.pop('candidate_labels'	)
        lowerCAmelCase_       :    Optional[int]								=			model_inputs.pop('text_inputs'	)
        if isinstance(text_inputs[0]   ,      _a	):
          lowerCAmelCase_       :    Dict								=			text_inputs[0]
        else:
          # Batching case.
          lowerCAmelCase_       :    Any								=			text_inputs[0][0]

        lowerCAmelCase_       :    Optional[int]								=			self.model(**_a   ,      **_a	)

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





      def 					SCREAMING_SNAKE_CASE__			(				self    :				int   ,      SCREAMING_SNAKE_CASE_    :				Optional[Any]	):
        lowerCAmelCase_       :    Any								=			model_outputs.pop('candidate_labels'	)
        lowerCAmelCase_       :    Tuple								=			model_outputs['logits'][0]

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

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










"""simple docstring"""







from __future__ import annotations

def     UpperCamelCase_   (   lowerCAmelCase__				:    int						)						->       list[int]:

  """simple docstring"""




  lowerCAmelCase_       :    Any								=			2
  lowerCAmelCase_       :    List[Any]								=			[]
  while i * i <= n:
    if n % i:
      i += 1
    else:
      n //= i
      factors.append(lowerCAmelCase__						)
  if n > 1:
    factors.append(lowerCAmelCase__						)
  return factors


if __name__ == "__main__":
     import doctest

     doctest.testmod()





