


"""simple docstring"""



import os
import string
import sys


__snake_case   			=    1 << 8

__snake_case   			=    {
    """tab""": ord("""\t"""),
    """newline""": ord("""\r"""),
    """esc""": 27,
    """up""": 65 + ARROW_KEY_FLAG,
    """down""": 66 + ARROW_KEY_FLAG,
    """right""": 67 + ARROW_KEY_FLAG,
    """left""": 68 + ARROW_KEY_FLAG,
    """mod_int""": 91,
    """undefined""": sys.maxsize,
    """interrupt""": 3,
    """insert""": 50,
    """delete""": 51,
    """pg_up""": 53,
    """pg_down""": 54,
}

__snake_case   			=    KEYMAP["""up"""]
__snake_case   			=    KEYMAP["""left"""]

if sys.platform == "win32":
  __snake_case   			=    []
  __snake_case   			=    {
      B"""\xe0H""": KEYMAP["""up"""] - ARROW_KEY_FLAG,
      B"""\x00H""": KEYMAP["""up"""] - ARROW_KEY_FLAG,
      B"""\xe0P""": KEYMAP["""down"""] - ARROW_KEY_FLAG,
      B"""\x00P""": KEYMAP["""down"""] - ARROW_KEY_FLAG,
      B"""\xe0M""": KEYMAP["""right"""] - ARROW_KEY_FLAG,
      B"""\x00M""": KEYMAP["""right"""] - ARROW_KEY_FLAG,
      B"""\xe0K""": KEYMAP["""left"""] - ARROW_KEY_FLAG,
      B"""\x00K""": KEYMAP["""left"""] - ARROW_KEY_FLAG,
  }

for i in range(10):
  __snake_case   			=    ord(str(i))







def   __lowerCAmelCase						(						)	->						Optional[int]:


    """simple docstring"""





    if os.name == "nt":
        import msvcrt

        snake_case    :      Any      					=    "mbcs"
        # Flush the keyboard buffer
        while msvcrt.kbhit():
            msvcrt.getch()
        if len(_A ) == 0:
            # Read the keystroke
            snake_case    :      Optional[Any]      					=    msvcrt.getch()

            # If it is a prefix char, get second part
            if ch in (b"\x00", b"\xe0"):
                snake_case    :      Union[str, Any]      					=    ch + msvcrt.getch()
                # Translate actual Win chars to bullet char types
                try:
                    snake_case    :      Dict      					=    chr(WIN_KEYMAP[cha] )
                    WIN_CH_BUFFER.append(chr(KEYMAP["mod_int"] ) )
                    WIN_CH_BUFFER.append(_A )
                    if ord(_A ) in (
                        KEYMAP["insert"] - 1 << 9,
                        KEYMAP["delete"] - 1 << 9,
                        KEYMAP["pg_up"] - 1 << 9,
                        KEYMAP["pg_down"] - 1 << 9,
                    ):
                        WIN_CH_BUFFER.append(chr(126 ) )
                    snake_case    :      str      					=    chr(KEYMAP["esc"] )
                except KeyError:
                    snake_case    :      Optional[int]      					=    cha[1]
            else:
                snake_case    :      str      					=    ch.decode(_A )
        else:
            snake_case    :      Optional[Any]      					=    WIN_CH_BUFFER.pop(0 )
    elif os.name == "posix":
        import termios
        import tty

        snake_case    :      List[str]      					=    sys.stdin.fileno()
        snake_case    :      str      					=    termios.tcgetattr(_A )
        try:
            tty.setraw(_A )
            snake_case    :      str      					=    sys.stdin.read(1 )
        finally:
            termios.tcsetattr(_A						,							termios.TCSADRAIN						,							_A )
    return ch







def   __lowerCAmelCase						(						)	->						Optional[Any]:


    """simple docstring"""





    snake_case    :      Dict      					=    get_raw_chars()
    if ord(_A ) in [KEYMAP["interrupt"], KEYMAP["newline"]]:
        return char

    elif ord(_A ) == KEYMAP["esc"]:
        snake_case    :      Any      					=    get_raw_chars()
        if ord(_A ) == KEYMAP["mod_int"]:
            snake_case    :      Tuple      					=    get_raw_chars()
            if ord(_A ) >= KEYMAP["arrow_begin"] - ARROW_KEY_FLAG and ord(_A ) <= KEYMAP["arrow_end"] - ARROW_KEY_FLAG:
                return chr(ord(_A ) + ARROW_KEY_FLAG )
            else:
                return KEYMAP["undefined"]
        else:
            return get_raw_chars()

    else:
        if char in string.printable:
            return char
        else:
            return KEYMAP["undefined"]








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


lowerCamelCase           =					logging.get_logger(__name__)







@add_end_docstrings(lowerCamelCase__							)
class 						__magic_name__			(							lowerCamelCase__							):

   '''simple docstring'''




   def __init__(				self,    **lowercase_       )							->    Dict:






     """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_       )							->    Tuple:






     """simple docstring"""






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



   def  _UpperCAmelCase (				self,    **lowercase_       )							->    int:






     """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,    lowercase_,    lowercase_=None,    lowercase_="This is a sound of {}."       )							->    Union[str, Any]:






     """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
         a__  =requests.get(lowercase_       ).content
       else:
         with open(lowercase_,    '''rb'''       ) as f:
           a__  =f.read()

     if isinstance(lowercase_,    lowercase_       ):
       a__  =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'''       )

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



   def  _UpperCAmelCase (				self,    lowercase_       )							->    str:






     """simple docstring"""






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

     a__  =self.model(**lowercase_,    **lowercase_       )

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



   def  _UpperCAmelCase (				self,    lowercase_       )							->    Any:






     """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(lowercase_,    lowercase_       ),    key=lambda lowercase_       : -x[0]       )
     ]
     return result
