



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

lowercase_:  Dict     						=					logging.get_logger(__name__)



@add_end_docstrings(__snake_case						)
class 						lowercase__      (__snake_case						):

     """simple docstring"""






     def __init__( self :	Tuple   ,			*__a :	Union[str, Any]   ,			**__a :	List[str] ):
      super().__init__(*__a   ,			**__a )
      requires_backends(self   ,			"""decord""" )
      self.check_model_type(__a )


     def 		lowercase   ( self :	str   ,			__a :	int=None   ,			__a :	Optional[int]=None   ,			__a :	Optional[int]=None ):
      snake_case__				:							Optional[int]      							= {}
      if frame_sampling_rate is not None:
       snake_case__				:							Optional[int]      							= frame_sampling_rate
      if num_frames is not None:
       snake_case__				:							Dict      							= num_frames

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


     def __call__( self :	Dict   ,			__a :	Union[str, Any]   ,			**__a :	int ):
      return super().__call__(__a   ,			**__a )


     def 		lowercase   ( self :	Optional[Any]   ,			__a :	List[str]   ,			__a :	List[Any]=None   ,			__a :	Any=1 ):
      if num_frames is None:
       snake_case__				:							Union[str, Any]      							= 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__				:							Union[str, Any]      							= num_frames * frame_sampling_rate - 1
      snake_case__				:							Optional[Any]      							= np.linspace(__a   ,			__a   ,			num=__a   ,			dtype=np.intaa )

      snake_case__				:							str      							= videoreader.get_batch(__a ).asnumpy()
      snake_case__				:							str      							= list(__a )

      snake_case__				:							Dict      							= self.image_processor(__a   ,			return_tensors=self.framework )
      return model_inputs


     def 		lowercase   ( self :	Dict   ,			__a :	List[str] ):
      snake_case__				:							Optional[Any]      							= self.model(**__a )
      return model_outputs


     def 		lowercase   ( self :	Any   ,			__a :	Union[str, Any]   ,			__a :	int=5 ):
      if top_k > self.model.config.num_labels:
       snake_case__				:							Optional[Any]      							= self.model.config.num_labels

      if self.framework == "pt":
       snake_case__				:							str      							= model_outputs.logits.softmax(-1 )[0]
       snake_case__				:							Tuple      							= probs.topk(__a )
      else:
       raise ValueError(f'Unsupported framework: {self.framework}' )

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








'''simple docstring'''






from typing import List, Union

from ..utils import (
    add_end_docstrings,
    is_tf_available,
    is_torch_available,
    is_vision_available,
    logging,
    requires_backends,
)
from .base import PIPELINE_INIT_ARGS, Pipeline


if is_vision_available():
 from PIL import Image

 from ..image_utils import load_image

if is_tf_available():
 import tensorflow as tf

 from ..models.auto.modeling_tf_auto import TF_MODEL_FOR_IMAGE_CLASSIFICATION_MAPPING
 from ..tf_utils import stable_softmax

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

lowercase				=logging.get_logger(__name__)






@add_end_docstrings(lowerCAmelCase							)
class    __magic_name__				(						lowerCAmelCase							):






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




          '''simple docstring'''
          super().__init__(*snake_case    ,				**snake_case)
          requires_backends(self    ,				'vision')
          self.check_model_type(
              TF_MODEL_FOR_IMAGE_CLASSIFICATION_MAPPING
              if self.framework == 'tf'
              else MODEL_FOR_IMAGE_CLASSIFICATION_MAPPING)






       def      lowerCAmelCase						(     self    ,				snake_case=None)		->    str:




          '''simple docstring'''
          _UpperCAmelCase  :  Any				       ={}
          if top_k is not None:
             _UpperCAmelCase  :  Optional[int]				       =top_k
          return {}, {}, postprocess_params






       def __call__(     self    ,				snake_case    ,				**snake_case)		->    List[str]:




          '''simple docstring'''
          return super().__call__(snake_case    ,				**snake_case)






       def      lowerCAmelCase						(     self    ,				snake_case)		->    List[str]:




          '''simple docstring'''
          _UpperCAmelCase  :  int				       =load_image(snake_case)
          _UpperCAmelCase  :  Tuple				       =self.image_processor(images=snake_case    ,				return_tensors=self.framework)
          return model_inputs






       def      lowerCAmelCase						(     self    ,				snake_case)		->    Optional[Any]:




          '''simple docstring'''
          _UpperCAmelCase  :  List[str]				       =self.model(**snake_case)
          return model_outputs






       def      lowerCAmelCase						(     self    ,				snake_case    ,				snake_case=5)		->    Any:




          '''simple docstring'''
          if top_k > self.model.config.num_labels:
             _UpperCAmelCase  :  Optional[Any]				       =self.model.config.num_labels

          if self.framework == "pt":
             _UpperCAmelCase  :  List[Any]				       =model_outputs.logits.softmax(-1)[0]
             _UpperCAmelCase       ,			_UpperCAmelCase  :  Optional[Any]				       =probs.topk(snake_case)
          elif self.framework == "tf":
             _UpperCAmelCase  :  int				       =stable_softmax(model_outputs.logits    ,				axis=-1)[0]
             _UpperCAmelCase  :  List[Any]				       =tf.math.top_k(snake_case    ,				k=snake_case)
             _UpperCAmelCase       ,			_UpperCAmelCase  :  List[str]				       =topk.values.numpy(), topk.indices.numpy()
          else:
             raise ValueError(f"Unsupported framework: {self.framework}")

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

