
'''simple docstring'''



from typing import Dict, List, Optional, Union

import numpy as np

from .feature_extraction_utils import BatchFeature, FeatureExtractionMixin
from .utils import PaddingStrategy, TensorType, is_tf_tensor, is_torch_tensor, logging, to_numpy


UpperCamelCase__   :		List[Any]											=	logging.get_logger(__name__)

class  _a					(_lowerCamelCase):





       """simple docstring"""




       def __init__(				self							,				A__							,				A__							,				A__							,				**A__   )     ->      int:
          _SCREAMING_SNAKE_CASE  			=					feature_size
          _SCREAMING_SNAKE_CASE  			=					sampling_rate
          _SCREAMING_SNAKE_CASE  			=					padding_value

          _SCREAMING_SNAKE_CASE  			=					kwargs.pop("""padding_side"""							,				"""right"""   )
          _SCREAMING_SNAKE_CASE  			=					kwargs.pop("""return_attention_mask"""							,				a_   )

          super().__init__(**a_   )






       def 							UpperCamelCase					(				self							,				A__							,				A__ = True							,				A__ = None							,				A__ = False							,				A__ = None							,				A__ = None							,				A__ = None							,				)     ->      BatchFeature:
          # If we have a list of dicts, let's convert it in a dict of lists
          # We do this to allow using this method as a collate_fn function in PyTorch Dataloader
          if isinstance(a_							,				(list, tuple)   ) and isinstance(processed_features[0]							,				(dict, BatchFeature)   ):
             _SCREAMING_SNAKE_CASE  			=					{
                 key: [example[key] for example in processed_features] for key in processed_features[0].keys()
             }

          # The model's main input name, usually `input_values`, has be passed for padding
          if self.model_input_names[0] not in processed_features:
             raise ValueError(
                 """You should supply an instance of `transformers.BatchFeature` or list of `transformers.BatchFeature`"""
                 F" to this method that includes {self.model_input_names[0]}, but you provided"
                 F" {list(processed_features.keys()   )}"   )

          _SCREAMING_SNAKE_CASE  			=					processed_features[self.model_input_names[0]]
          _SCREAMING_SNAKE_CASE  			=					(
              return_attention_mask if return_attention_mask is not None else self.return_attention_mask
          )

          if len(a_   ) == 0:
             if return_attention_mask:
                _SCREAMING_SNAKE_CASE  			=					[]
             return processed_features

          # If we have PyTorch/TF tensors or lists as inputs, we cast them as Numpy arrays
          # and rebuild them afterwards if no return_tensors is specified
          # Note that we lose the specific device the tensor may be on for PyTorch

          _SCREAMING_SNAKE_CASE  			=					required_input[0]
          if isinstance(a_							,				(list, tuple)   ):
             # first_element might be an empty list/tuple in some edge cases so we grab the first non empty element.
             _SCREAMING_SNAKE_CASE  			=					0
             while len(required_input[index]   ) == 0:
                index += 1
             if index < len(a_   ):
                _SCREAMING_SNAKE_CASE  			=					required_input[index][0]

          if return_tensors is None:
             if is_tf_tensor(a_   ):
                _SCREAMING_SNAKE_CASE  			=					"""tf"""
             elif is_torch_tensor(a_   ):
                _SCREAMING_SNAKE_CASE  			=					"""pt"""
             elif isinstance(a_							,				(int, float, list, tuple, np.ndarray)   ):
                _SCREAMING_SNAKE_CASE  			=					"""np"""
             else:
                raise ValueError(
                    F"type of {first_element} unknown: {type(a_   )}. "
                    """Should be one of a python, numpy, pytorch or tensorflow object."""   )

          for key, value in processed_features.items():
             if isinstance(value[0]							,				(int, float)   ):
                _SCREAMING_SNAKE_CASE  			=					to_numpy(a_   )
             else:
                _SCREAMING_SNAKE_CASE  			=					[to_numpy(a_   ) for v in value]

                # Convert padding_strategy in PaddingStrategy
          _SCREAMING_SNAKE_CASE  			=					self._get_padding_strategies(padding=a_							,				max_length=a_   )

          _SCREAMING_SNAKE_CASE  			=					processed_features[self.model_input_names[0]]

          _SCREAMING_SNAKE_CASE  			=					len(a_   )
          if not all(len(a_   ) == batch_size for v in processed_features.values()   ):
             raise ValueError("""Some items in the output dictionary have a different batch size than others."""   )

          _SCREAMING_SNAKE_CASE  			=					[]
          for i in range(a_   ):
             _SCREAMING_SNAKE_CASE  			=					{k: v[i] for k, v in processed_features.items()}
             # truncation
             _SCREAMING_SNAKE_CASE  			=					self._truncate(
                 a_							,				max_length=a_							,				pad_to_multiple_of=a_							,				truncation=a_							,				)
             truncated_inputs.append(a_   )

          if padding_strategy == PaddingStrategy.LONGEST:
             # make sure that `max_length` cannot be longer than the longest truncated length
             _SCREAMING_SNAKE_CASE  			=					max(len(input_slice[self.model_input_names[0]]   ) for input_slice in truncated_inputs   )
             _SCREAMING_SNAKE_CASE  			=					PaddingStrategy.MAX_LENGTH

          _SCREAMING_SNAKE_CASE  			=					{}
          for i in range(a_   ):
             # padding
             _SCREAMING_SNAKE_CASE  			=					self._pad(
                 truncated_inputs[i]							,				max_length=a_							,				padding_strategy=a_							,				pad_to_multiple_of=a_							,				return_attention_mask=a_							,				)

             for key, value in outputs.items():
                if key not in batch_outputs:
                   _SCREAMING_SNAKE_CASE  			=					[]
                if value.dtype is np.dtype(np.floataa   ):
                   _SCREAMING_SNAKE_CASE  			=					value.astype(np.floataa   )
                batch_outputs[key].append(a_   )

          return BatchFeature(a_							,				tensor_type=a_   )






       def 							UpperCamelCase					(				self							,				A__							,				A__ = None							,				A__ = PaddingStrategy.DO_NOT_PAD							,				A__ = None							,				A__ = None							,				)     ->      dict:
          _SCREAMING_SNAKE_CASE  			=					processed_features[self.model_input_names[0]]

          if padding_strategy == PaddingStrategy.LONGEST:
             _SCREAMING_SNAKE_CASE  			=					len(a_   )

          if max_length is not None and pad_to_multiple_of is not None and (max_length % pad_to_multiple_of != 0):
             _SCREAMING_SNAKE_CASE  			=					((max_length // pad_to_multiple_of) + 1) * pad_to_multiple_of

          _SCREAMING_SNAKE_CASE  			=					padding_strategy != PaddingStrategy.DO_NOT_PAD and len(a_   ) < max_length

          if return_attention_mask and "attention_mask" not in processed_features:
             _SCREAMING_SNAKE_CASE  			=					np.ones(len(a_   )							,				dtype=np.intaa   )

          if needs_to_be_padded:
             _SCREAMING_SNAKE_CASE  			=					max_length - len(a_   )
             if self.padding_side == "right":
                if return_attention_mask:
                   _SCREAMING_SNAKE_CASE  			=					np.pad(
                       processed_features["""attention_mask"""]							,				(0, difference)   )
                _SCREAMING_SNAKE_CASE  			=					((0, difference), (0, 0)) if self.feature_size > 1 else (0, difference)
                _SCREAMING_SNAKE_CASE  			=					np.pad(
                    a_							,				a_							,				"""constant"""							,				constant_values=self.padding_value   )
             elif self.padding_side == "left":
                if return_attention_mask:
                   _SCREAMING_SNAKE_CASE  			=					np.pad(
                       processed_features["""attention_mask"""]							,				(difference, 0)   )
                _SCREAMING_SNAKE_CASE  			=					((difference, 0), (0, 0)) if self.feature_size > 1 else (difference, 0)
                _SCREAMING_SNAKE_CASE  			=					np.pad(
                    a_							,				a_							,				"""constant"""							,				constant_values=self.padding_value   )
             else:
                raise ValueError("""Invalid padding strategy:""" + str(self.padding_side   )   )

          return processed_features






       def 							UpperCamelCase					(				self							,				A__							,				A__ = None							,				A__ = None							,				A__ = None							,				)     ->      Tuple:
          if not truncation:
             return processed_features
          elif truncation and max_length is None:
             raise ValueError("""When setting ``truncation=True``, make sure that ``max_length`` is defined."""   )

          _SCREAMING_SNAKE_CASE  			=					processed_features[self.model_input_names[0]]

          # find `max_length` that fits `pad_to_multiple_of`
          if max_length is not None and pad_to_multiple_of is not None and (max_length % pad_to_multiple_of != 0):
             _SCREAMING_SNAKE_CASE  			=					((max_length // pad_to_multiple_of) + 1) * pad_to_multiple_of

          _SCREAMING_SNAKE_CASE  			=					len(a_   ) > max_length

          if needs_to_be_truncated:
             _SCREAMING_SNAKE_CASE  			=					processed_features[self.model_input_names[0]][:max_length]
             if "attention_mask" in processed_features:
                _SCREAMING_SNAKE_CASE  			=					processed_features["""attention_mask"""][:max_length]

          return processed_features






       def 							UpperCamelCase					(				self							,				A__=False							,				A__=None   )     ->      int:

          # Get padding strategy
          if padding is not False:
             if padding is True:
                _SCREAMING_SNAKE_CASE  			=					PaddingStrategy.LONGEST  # Default to pad to the longest sequence in the batch
             elif not isinstance(a_							,				a_   ):
                _SCREAMING_SNAKE_CASE  			=					PaddingStrategy(a_   )
             elif isinstance(a_							,				a_   ):
                _SCREAMING_SNAKE_CASE  			=					padding
          else:
             _SCREAMING_SNAKE_CASE  			=					PaddingStrategy.DO_NOT_PAD

          # Set max length if needed
          if max_length is None:
             if padding_strategy == PaddingStrategy.MAX_LENGTH:
                raise ValueError(
                    F"When setting ``padding={PaddingStrategy.MAX_LENGTH}``, make sure that max_length is defined"   )

                # Test if we have a padding value
          if padding_strategy != PaddingStrategy.DO_NOT_PAD and (self.padding_value is None):
             raise ValueError(
                 """Asking to pad but the feature_extractor does not have a padding value. Please select a value to use"""
                 """ as `padding_value`. For example: `feature_extractor.padding_value = 0.0`."""   )

          return padding_strategy









'''simple docstring'''




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

_a					:			List[Any]							=    logging.get_logger(__name__)



@add_end_docstrings(a			)
class 							lowercase_						(		a			):

  '''simple docstring'''





  def __init__(  self ,			*a_ ,			**a_  )				->					str:


    """simple docstring"""
    super().__init__(*a_ ,			**a_  )
    requires_backends(self ,			'decord'  )
    self.check_model_type(a_  )
  def     snake_case_      (  self ,			a_=None ,			a_=None ,			a_=None  )				->					int:


    """simple docstring"""
    UpperCAmelCase    				=						{}
    if frame_sampling_rate is not None:
      UpperCAmelCase    				=						frame_sampling_rate
    if num_frames is not None:
      UpperCAmelCase    				=						num_frames

    UpperCAmelCase    				=						{}
    if top_k is not None:
      UpperCAmelCase    				=						top_k
    return preprocess_params, {}, postprocess_params
  def __call__(  self ,			a_ ,			**a_  )				->					Union[str, Any]:


    """simple docstring"""
    return super().__call__(a_ ,			**a_  )
  def     snake_case_      (  self ,			a_ ,			a_=None ,			a_=1  )				->					Tuple:


    """simple docstring"""
    if num_frames is None:
      UpperCAmelCase    				=						self.model.config.num_frames

    if video.startswith('http://'  ) or video.startswith('https://'  ):
      UpperCAmelCase    				=						BytesIO(requests.get(a_  ).content  )

    UpperCAmelCase    				=						VideoReader(a_  )
    videoreader.seek(0  )

    UpperCAmelCase    				=						0
    UpperCAmelCase    				=						num_frames * frame_sampling_rate - 1
    UpperCAmelCase    				=						np.linspace(a_ ,			a_ ,			num=a_ ,			dtype=np.intaa  )

    UpperCAmelCase    				=						videoreader.get_batch(a_  ).asnumpy()
    UpperCAmelCase    				=						list(a_  )

    UpperCAmelCase    				=						self.image_processor(a_ ,			return_tensors=self.framework  )
    return model_inputs
  def     snake_case_      (  self ,			a_  )				->					Union[str, Any]:


    """simple docstring"""
    UpperCAmelCase    				=						self.model(**a_  )
    return model_outputs





  def     snake_case_      (  self ,			a_ ,			a_=5  )				->					Union[str, Any]:


    """simple docstring"""
    if top_k > self.model.config.num_labels:
      UpperCAmelCase    				=						self.model.config.num_labels

    if self.framework == "pt":
      UpperCAmelCase    				=						model_outputs.logits.softmax(-1  )[0]
      UpperCAmelCase				,   UpperCAmelCase    				=						probs.topk(a_  )
    else:
      raise ValueError(F'''Unsupported framework: {self.framework}'''  )

    UpperCAmelCase    				=						scores.tolist()
    UpperCAmelCase    				=						ids.tolist()
    return [{"score": score, "label": self.model.config.idalabel[_id]} for score, _id in zip(a_ ,			a_  )]


