"""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[int]			       =  logging.get_logger(__name__)
@add_end_docstrings(SCREAMING_SNAKE_CASE_						)
class 		__A				(					SCREAMING_SNAKE_CASE_						):






     def __init__(						self							,	**a__    ):
       super().__init__(**_lowerCAmelCase    )

       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__    ):
       return super().__call__(_lowerCAmelCase							,	**_lowerCAmelCase    )






     def        __A    (						self							,	**a__    ):
       _lowerCAmelCase		:			List[str]         =	{}
       if "candidate_labels" in kwargs:
         _lowerCAmelCase		:			Optional[int]         =	kwargs["""candidate_labels"""]
       if "hypothesis_template" in kwargs:
         _lowerCAmelCase		:			Dict         =	kwargs["""hypothesis_template"""]

       return preprocess_params, {}, {}






     def        __A    (						self							,	a__							,	a__=None							,	a__="This is a sound of {}."    ):
       if isinstance(_lowerCAmelCase							,	_lowerCAmelCase    ):
         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		:			Tuple         =	requests.get(_lowerCAmelCase    ).content
         else:
           with open(_lowerCAmelCase							,	"""rb"""    ) as f:
             _lowerCAmelCase		:			Dict         =	f.read()

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

       if not isinstance(_lowerCAmelCase							,	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		:			List[Any]         =	candidate_labels
       _lowerCAmelCase		:			List[Any]         =	[hypothesis_template.format(_lowerCAmelCase    ) for x in candidate_labels]
       _lowerCAmelCase		:			str         =	self.tokenizer(_lowerCAmelCase							,	return_tensors=self.framework							,	padding=_lowerCAmelCase    )
       _lowerCAmelCase		:			List[Any]         =	[text_inputs]
       return inputs






     def        __A    (						self							,	a__    ):
       _lowerCAmelCase		:			int         =	model_inputs.pop("""candidate_labels"""    )
       _lowerCAmelCase		:			Tuple         =	model_inputs.pop("""text_inputs"""    )
       if isinstance(text_inputs[0]							,	_lowerCAmelCase    ):
         _lowerCAmelCase		:			str         =	text_inputs[0]
       else:
         # Batching case.
         _lowerCAmelCase		:			List[Any]         =	text_inputs[0][0]

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

       _lowerCAmelCase		:			Optional[int]         =	{
           """candidate_labels""": candidate_labels,
           """logits""": outputs.logits_per_audio,
       }
       return model_outputs






     def        __A    (						self							,	a__    ):
       _lowerCAmelCase		:			Tuple         =	model_outputs.pop("""candidate_labels"""    )
       _lowerCAmelCase		:			Tuple         =	model_outputs["""logits"""][0]

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

       _lowerCAmelCase		:			Union[str, Any]         =	[
           {"""score""": score, """label""": candidate_label}
           for score, candidate_label in sorted(zip(_lowerCAmelCase							,	_lowerCAmelCase    )							,	key=lambda a__    : -x[0]    )
       ]
       return result








import logging
import os

from .state import PartialState


class      __lowerCAmelCase  (			logging.LoggerAdapter       ):






  """simple docstring"""





  @staticmethod
  def        lowerCAmelCase__	(  _lowerCAmelCase							:							int	) ->			Tuple:




     """simple docstring"""
     snake_case_            =       PartialState()
     return not main_process_only or (main_process_only and state.is_main_process)



  def        lowerCAmelCase__	(  self							:							Union[str, Any]     ,					_lowerCAmelCase							:							List[Any]     ,					_lowerCAmelCase							:							Tuple     ,					*_lowerCAmelCase							:							List[Any]     ,					**_lowerCAmelCase							:							Any	) ->			Optional[int]:




     """simple docstring"""
     if PartialState._shared_state == {}:
        raise RuntimeError(
            "You must initialize the accelerate state by calling either `PartialState()` or `Accelerator()` before using the logging utility."	)
     snake_case_            =       kwargs.pop("main_process_only"     ,					_lowerCAmelCase	)
     snake_case_            =       kwargs.pop("in_order"     ,					_lowerCAmelCase	)

     if self.isEnabledFor(_lowerCAmelCase	):
        if self._should_log(_lowerCAmelCase	):
           snake_case_     ,     snake_case_            =       self.process(_lowerCAmelCase     ,					_lowerCAmelCase	)
           self.logger.log(_lowerCAmelCase     ,					_lowerCAmelCase     ,					*_lowerCAmelCase     ,					**_lowerCAmelCase	)

        elif in_order:
           snake_case_            =       PartialState()
           for i in range(state.num_processes	):
              if i == state.process_index:
                 snake_case_     ,     snake_case_            =       self.process(_lowerCAmelCase     ,					_lowerCAmelCase	)
                 self.logger.log(_lowerCAmelCase     ,					_lowerCAmelCase     ,					*_lowerCAmelCase     ,					**_lowerCAmelCase	)
              state.wait_for_everyone()






def       _lowerCAmelCase    (				lowerCAmelCase_				:str		,    lowerCAmelCase_				:str = None				)->Any:
   '''simple docstring'''






   if log_level is None:
      snake_case_            =       os.environ.get("ACCELERATE_LOG_LEVEL"		,    lowerCAmelCase_				)
   snake_case_            =       logging.getLogger(lowerCAmelCase_				)
   if log_level is not None:
      logger.setLevel(log_level.upper()				)
      logger.root.setLevel(log_level.upper()				)
   return MultiProcessAdapter(lowerCAmelCase_		,    {}				)




