
import warnings
from contextlib import contextmanager

from ...processing_utils import ProcessorMixin
from .feature_extraction_wavaveca import WavaVecaFeatureExtractor
from .tokenization_wavaveca import WavaVecaCTCTokenizer




class 						__lowercase		(UpperCamelCase__							):




      """simple docstring"""





      _snake_case												=						"""Wav2Vec2FeatureExtractor"""
      _snake_case												=						"""AutoTokenizer"""

      def __init__(				self    ,				A    ,				A     )		->    List[str]:
         super().__init__(A    ,				A     )
         snake_case				:    Union[str, Any]           =	self.feature_extractor
         snake_case				:    Optional[Any]           =	False

      @classmethod
      def      UpperCAmelCase				(				cls    ,				A    ,				**A     )		->    Tuple:
         try:
            return super().from_pretrained(A    ,				**A     )
         except OSError:
            warnings.warn(
                f"""Loading a tokenizer inside {cls.__name__} from a config that does not"""
                """ include a `tokenizer_class` attribute is deprecated and will be """
                """removed in v5. Please add `'tokenizer_class': 'Wav2Vec2CTCTokenizer'`"""
                """ attribute to either your `config.json` or `tokenizer_config.json` """
                """file to suppress this warning: """    ,				A    ,				)

            snake_case				:    str           =	WavaVecaFeatureExtractor.from_pretrained(A    ,				**A     )
            snake_case				:    Optional[int]           =	WavaVecaCTCTokenizer.from_pretrained(A    ,				**A     )

            return cls(feature_extractor=A    ,				tokenizer=A     )

      def __call__(				self    ,				*A    ,				**A     )		->    List[str]:
         # For backward compatibility
         if self._in_target_context_manager:
            return self.current_processor(*A    ,				**A     )

         if "raw_speech" in kwargs:
            warnings.warn("""Using `raw_speech` as a keyword argument is deprecated. Use `audio` instead."""     )
            snake_case				:    int           =	kwargs.pop("""raw_speech"""     )
         else:
            snake_case				:    str           =	kwargs.pop("""audio"""    ,				A     )
         snake_case				:    str           =	kwargs.pop("""sampling_rate"""    ,				A     )
         snake_case				:    Dict           =	kwargs.pop("""text"""    ,				A     )
         if len(A     ) > 0:
            snake_case				:    List[str]           =	args[0]
            snake_case				:    Any           =	args[1:]

         if audio is None and text is None:
            raise ValueError("""You need to specify either an `audio` or `text` input to process."""     )

         if audio is not None:
            snake_case				:    str           =	self.feature_extractor(A    ,				*A    ,				sampling_rate=A    ,				**A     )
         if text is not None:
            snake_case				:    str           =	self.tokenizer(A    ,				**A     )

         if text is None:
            return inputs
         elif audio is None:
            return encodings
         else:
            snake_case				:    List[Any]           =	encodings["""input_ids"""]
            return inputs

      def      UpperCAmelCase				(				self    ,				*A    ,				**A     )		->    Optional[Any]:
         # For backward compatibility
         if self._in_target_context_manager:
            return self.current_processor.pad(*A    ,				**A     )

         snake_case				:    Any           =	kwargs.pop("""input_features"""    ,				A     )
         snake_case				:    Optional[Any]           =	kwargs.pop("""labels"""    ,				A     )
         if len(A     ) > 0:
            snake_case				:    Optional[Any]           =	args[0]
            snake_case				:    str           =	args[1:]

         if input_features is not None:
            snake_case				:    Dict           =	self.feature_extractor.pad(A    ,				*A    ,				**A     )
         if labels is not None:
            snake_case				:    List[str]           =	self.tokenizer.pad(A    ,				**A     )

         if labels is None:
            return input_features
         elif input_features is None:
            return labels
         else:
            snake_case				:    Union[str, Any]           =	labels["""input_ids"""]
            return input_features

      def      UpperCAmelCase				(				self    ,				*A    ,				**A     )		->    Tuple:
         return self.tokenizer.batch_decode(*A    ,				**A     )

      def      UpperCAmelCase				(				self    ,				*A    ,				**A     )		->    Optional[int]:
         return self.tokenizer.decode(*A    ,				**A     )

      @contextmanager
      def      UpperCAmelCase				(				self     )		->    Optional[Any]:
         warnings.warn(
             """`as_target_processor` is deprecated and will be removed in v5 of Transformers. You can process your """
             """labels by using the argument `text` of the regular `__call__` method (either in the same call as """
             """your audio inputs, or in a separate call."""     )
         snake_case				:    Union[str, Any]           =	True
         snake_case				:    Union[str, Any]           =	self.tokenizer
         yield
         snake_case				:    Any           =	self.feature_extractor
         snake_case				:    Union[str, Any]           =	False











from io import BytesIO
from typing import List, Union

import requests

from ..utils import add_end_docstrings, is_decord_available, is_torch_available, logging, requires_backends
from .base import PIPELINE_INIT_ARGS, Pipeline


if is_decord_available():
       import numpy as np
       from decord import VideoReader


if is_torch_available():
       from ..models.auto.modeling_auto import MODEL_FOR_VIDEO_CLASSIFICATION_MAPPING

lowerCamelCase	:	int				=   logging.get_logger(__name__)




@add_end_docstrings(UpperCamelCase__							)
class 						__lowercase		(UpperCamelCase__							):




      """simple docstring"""





      def __init__(				self    ,				*A    ,				**A     )		->    Optional[int]:
         super().__init__(*A    ,				**A     )
         requires_backends(self    ,				"""decord"""     )
         self.check_model_type(A     )

      def      UpperCAmelCase				(				self    ,				A=None    ,				A=None    ,				A=None     )		->    int:
         snake_case				:    Any           =	{}
         if frame_sampling_rate is not None:
            snake_case				:    int           =	frame_sampling_rate
         if num_frames is not None:
            snake_case				:    Union[str, Any]           =	num_frames

         snake_case				:    str           =	{}
         if top_k is not None:
            snake_case				:    Optional[int]           =	top_k
         return preprocess_params, {}, postprocess_params

      def __call__(				self    ,				A    ,				**A     )		->    Dict:
         return super().__call__(A    ,				**A     )

      def      UpperCAmelCase				(				self    ,				A    ,				A=None    ,				A=1     )		->    Tuple:
         if num_frames is None:
            snake_case				:    Tuple           =	self.model.config.num_frames

         if video.startswith("""http://"""     ) or video.startswith("""https://"""     ):
            snake_case				:    Optional[int]           =	BytesIO(requests.get(A     ).content     )

         snake_case				:    Optional[Any]           =	VideoReader(A     )
         videoreader.seek(0     )

         snake_case				:    Optional[Any]           =	0
         snake_case				:    Optional[Any]           =	num_frames * frame_sampling_rate - 1
         snake_case				:    Any           =	np.linspace(A    ,				A    ,				num=A    ,				dtype=np.intaa     )

         snake_case				:    int           =	videoreader.get_batch(A     ).asnumpy()
         snake_case				:    List[Any]           =	list(A     )

         snake_case				:    Optional[Any]           =	self.image_processor(A    ,				return_tensors=self.framework     )
         return model_inputs

      def      UpperCAmelCase				(				self    ,				A     )		->    List[str]:
         snake_case				:    Dict           =	self.model(**A     )
         return model_outputs

      def      UpperCAmelCase				(				self    ,				A    ,				A=5     )		->    int:
         if top_k > self.model.config.num_labels:
            snake_case				:    str           =	self.model.config.num_labels

         if self.framework == "pt":
            snake_case				:    List[Any]           =	model_outputs.logits.softmax(-1     )[0]
            snake_case     , snake_case				:    Tuple           =	probs.topk(A     )
         else:
            raise ValueError(f"""Unsupported framework: {self.framework}"""     )

         snake_case				:    List[Any]           =	scores.tolist()
         snake_case				:    str           =	ids.tolist()
         return [{"score": score, "label": self.model.config.idalabel[_id]} for score, _id in zip(A    ,				A     )]





