



from typing import List, Optional, Union

from ...configuration_utils import PretrainedConfig
from ...utils import logging


lowerCAmelCase:      Dict		 =		logging.get_logger(__name__)

lowerCAmelCase:      Any		 =		{
    "huggingface/informer-tourism-monthly": (
        "https://huggingface.co/huggingface/informer-tourism-monthly/resolve/main/config.json"
    ),
    # See all Informer models at https://huggingface.co/models?filter=informer
}

class 							a__(      lowerCamelCase__			):
    lowercase__				=					"""informer"""
    lowercase__				=					{
        """hidden_size""": """d_model""",
        """num_attention_heads""": """encoder_attention_heads""",
        """num_hidden_layers""": """encoder_layers""",
    }







    def __init__(   self				:						Dict		,    __snake_case				:						Optional[int] = None		,    __snake_case				:						Optional[int] = None		,    __snake_case				:						str = "student_t"		,    __snake_case				:						str = "nll"		,    __snake_case				:						int = 1		,    __snake_case				:						List[int] = None		,    __snake_case				:						Optional[Union[str, bool]] = "mean"		,    __snake_case				:						int = 0		,    __snake_case				:						int = 0		,    __snake_case				:						int = 0		,    __snake_case				:						int = 0		,    __snake_case				:						Optional[List[int]] = None		,    __snake_case				:						Optional[List[int]] = None		,    __snake_case				:						int = 64		,    __snake_case				:						int = 32		,    __snake_case				:						int = 32		,    __snake_case				:						int = 2		,    __snake_case				:						int = 2		,    __snake_case				:						int = 2		,    __snake_case				:						int = 2		,    __snake_case				:						bool = True		,    __snake_case				:						str = "gelu"		,    __snake_case				:						float = 0.05		,    __snake_case				:						float = 0.1		,    __snake_case				:						float = 0.1		,    __snake_case				:						float = 0.1		,    __snake_case				:						float = 0.1		,    __snake_case				:						int = 1_00		,    __snake_case				:						float = 0.02		,    __snake_case				:						List[Any]=True		,    __snake_case				:						str = "prob"		,    __snake_case				:						int = 5		,    __snake_case				:						bool = True		,    **__snake_case				:						int		,    ):
      # time series specific configuration
      a		:      str       =  prediction_length
      a		:      List[str]       =  context_length or prediction_length
      a		:      Optional[int]       =  distribution_output
      a		:      Tuple       =  loss
      a		:      List[Any]       =  input_size
      a		:      Optional[Any]       =  num_time_features
      a		:      List[Any]       =  lags_sequence if lags_sequence is not None else [1, 2, 3, 4, 5, 6, 7]
      a		:      int       =  scaling
      a		:      Any       =  num_dynamic_real_features
      a		:      Any       =  num_static_real_features
      a		:      Tuple       =  num_static_categorical_features

      # set cardinality
      if cardinality and num_static_categorical_features > 0:
        if len(UpperCAmelCase__			) != num_static_categorical_features:
          raise ValueError(
              'The cardinality should be a list of the same length as `num_static_categorical_features`'			)
        a		:      Optional[int]       =  cardinality
      else:
        a		:      int       =  [0]

      # set embedding_dimension
      if embedding_dimension and num_static_categorical_features > 0:
        if len(UpperCAmelCase__			) != num_static_categorical_features:
          raise ValueError(
              'The embedding dimension should be a list of the same length as `num_static_categorical_features`'			)
        a		:      Optional[Any]       =  embedding_dimension
      else:
        a		:      Union[str, Any]       =  [min(50		,    (cat + 1) // 2			) for cat in self.cardinality]

      a		:      Tuple       =  num_parallel_samples

      # Transformer architecture configuration
      a		:      Union[str, Any]       =  input_size * len(self.lags_sequence			) + self._number_of_features
      a		:      Dict       =  d_model
      a		:      Dict       =  encoder_attention_heads
      a		:      List[Any]       =  decoder_attention_heads
      a		:      Union[str, Any]       =  encoder_ffn_dim
      a		:      List[Any]       =  decoder_ffn_dim
      a		:      Any       =  encoder_layers
      a		:      Dict       =  decoder_layers

      a		:      Union[str, Any]       =  dropout
      a		:      Optional[Any]       =  attention_dropout
      a		:      Tuple       =  activation_dropout
      a		:      List[Any]       =  encoder_layerdrop
      a		:      int       =  decoder_layerdrop

      a		:      Union[str, Any]       =  activation_function
      a		:      Dict       =  init_std

      a		:      Dict       =  use_cache

      # Informer
      a		:      str       =  attention_type
      a		:      List[str]       =  sampling_factor
      a		:      int       =  distil

      super().__init__(is_encoder_decoder=UpperCAmelCase__		,    **UpperCAmelCase__			)







    @property
    def     lowercase_ (   self				:						Optional[int]			):
      return (
          sum(self.embedding_dimension			)
          + self.num_dynamic_real_features
          + self.num_time_features
          + self.num_static_real_features
          + self.input_size * 2  # the log1p(abs(loc)) and log(scale) features
      )




'''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


lowerCAmelCase:      Optional[Any]		 =		logging.get_logger(__name__)

@add_end_docstrings(lowerCamelCase__			)
class 							a__(      lowerCamelCase__			):







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

      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		,    __snake_case				:						Union[np.ndarray, bytes, str]		,    **__snake_case				:						Optional[Any]			):
      return super().__call__(__snake_case		,    **__snake_case			)







    def     lowercase_ (   self				:						Dict		,    **__snake_case				:						Optional[int]			):
      a		:      str       =  {}
      if "candidate_labels" in kwargs:
        a		:      Optional[int]       =  kwargs['candidate_labels']
      if "hypothesis_template" in kwargs:
        a		:      Any       =  kwargs['hypothesis_template']

      return preprocess_params, {}, {}







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

      if isinstance(__snake_case		,    __snake_case			):
        a		:      Optional[Any]       =  ffmpeg_read(__snake_case		,    self.feature_extractor.sampling_rate			)

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







    def     lowercase_ (   self				:						Optional[Any]		,    __snake_case				:						str			):
      a		:      Any       =  model_inputs.pop('candidate_labels'			)
      a		:      List[Any]       =  model_inputs.pop('text_inputs'			)
      if isinstance(text_inputs[0]		,    __snake_case			):
        a		:      str       =  text_inputs[0]
      else:
        # Batching case.
        a		:      Dict       =  text_inputs[0][0]

      a		:      List[Any]       =  self.model(**__snake_case		,    **__snake_case			)

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







    def     lowercase_ (   self				:						Union[str, Any]		,    __snake_case				:						str			):
      a		:      Optional[int]       =  model_outputs.pop('candidate_labels'			)
      a		:      List[str]       =  model_outputs['logits'][0]

      if self.framework == "pt":
        a		:      str       =  logits.softmax(dim=0			)
        a		:      List[Any]       =  probs.tolist()
      else:
        raise ValueError('`tf` framework not supported.'			)

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