

"""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_				:      Optional[int]           =			logging.get_logger(__name__)





@add_end_docstrings(A__      )
class  lowerCamelCase    (A__      ):

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

			def        SCREAMING_SNAKE_CASE					(       self					:    Union[str, Any]     ,					__UpperCAmelCase					:    Optional[int]=None     ,					__UpperCAmelCase					:    Dict=None     ,					__UpperCAmelCase					:    Any=None							) ->     List[str]:
						SCREAMING_SNAKE_CASE__										=   {}
						if frame_sampling_rate is not None:
									SCREAMING_SNAKE_CASE__										=   frame_sampling_rate
						if num_frames is not None:
									SCREAMING_SNAKE_CASE__										=   num_frames

						SCREAMING_SNAKE_CASE__										=   {}
						if top_k is not None:
									SCREAMING_SNAKE_CASE__										=   top_k
						return preprocess_params, {}, postprocess_params

			def __call__(       self					:    Optional[Any]     ,					__UpperCAmelCase					:    Union[str, List[str]]     ,					**__UpperCAmelCase					:    Tuple							) ->     Tuple:
						return super().__call__(__UpperCAmelCase     ,					**__UpperCAmelCase							)

			def        SCREAMING_SNAKE_CASE					(       self					:    List[Any]     ,					__UpperCAmelCase					:    Union[str, Any]     ,					__UpperCAmelCase					:    List[Any]=None     ,					__UpperCAmelCase					:    Union[str, Any]=1							) ->     Any:
						if num_frames is None:
									SCREAMING_SNAKE_CASE__										=   self.model.config.num_frames

						if video.startswith("""http://"""							) or video.startswith("""https://"""							):
									SCREAMING_SNAKE_CASE__										=   BytesIO(requests.get(__UpperCAmelCase							).content							)

						SCREAMING_SNAKE_CASE__										=   VideoReader(__UpperCAmelCase							)
						videoreader.seek(0							)

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

						SCREAMING_SNAKE_CASE__										=   videoreader.get_batch(__UpperCAmelCase							).asnumpy()
						SCREAMING_SNAKE_CASE__										=   list(__UpperCAmelCase							)

						SCREAMING_SNAKE_CASE__										=   self.image_processor(__UpperCAmelCase     ,					return_tensors=self.framework							)
						return model_inputs

			def        SCREAMING_SNAKE_CASE					(       self					:    Tuple     ,					__UpperCAmelCase					:    Optional[int]							) ->     Union[str, Any]:
						SCREAMING_SNAKE_CASE__										=   self.model(**__UpperCAmelCase							)
						return model_outputs





			def        SCREAMING_SNAKE_CASE					(       self					:    List[Any]     ,					__UpperCAmelCase					:    str     ,					__UpperCAmelCase					:    Optional[Any]=5							) ->     int:
						if top_k > self.model.config.num_labels:
									SCREAMING_SNAKE_CASE__										=   self.model.config.num_labels

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

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





"""simple docstring"""




from typing import List

from .keymap import KEYMAP, get_character

def 						A							(  snake_case__     ):




			'''simple docstring'''


			def decorator(snake_case__     ):
						SCREAMING_SNAKE_CASE__										=   getattr(snake_case__		,					"""handle_key"""		,					[]     )
						handle += [key]
						setattr(snake_case__		,					"""handle_key"""		,					snake_case__     )
						return func

			return decorator

def 						A							(  *snake_case__     ):




			'''simple docstring'''


			def decorator(snake_case__     ):
						SCREAMING_SNAKE_CASE__										=   getattr(snake_case__		,					"""handle_key"""		,					[]     )
						handle += keys
						setattr(snake_case__		,					"""handle_key"""		,					snake_case__     )
						return func

			return decorator





class  lowerCamelCase    (A__      ):

			def __new__(       cls					:    Dict     ,					__UpperCAmelCase					:    Tuple     ,					__UpperCAmelCase					:    Any     ,					__UpperCAmelCase					:    Optional[Any]							) ->     List[str]:
						SCREAMING_SNAKE_CASE__										=   super().__new__(cls     ,					__UpperCAmelCase     ,					__UpperCAmelCase     ,					__UpperCAmelCase							)
						if not hasattr(__UpperCAmelCase     ,					"""key_handler"""							):
									setattr(__UpperCAmelCase     ,					"""key_handler"""     ,					{}							)
						setattr(__UpperCAmelCase     ,					"""handle_input"""     ,					KeyHandler.handle_input							)

						for value in attrs.values():
									SCREAMING_SNAKE_CASE__										=   getattr(__UpperCAmelCase     ,					"""handle_key"""     ,					[]							)
									for key in handled_keys:
												SCREAMING_SNAKE_CASE__										=   value
						return new_cls





			@staticmethod
			def        SCREAMING_SNAKE_CASE					(       cls					:    Tuple							) ->     Dict:
						SCREAMING_SNAKE_CASE__										=   get_character()
						if char != KEYMAP["undefined"]:
									SCREAMING_SNAKE_CASE__										=   ord(__UpperCAmelCase							)
						SCREAMING_SNAKE_CASE__										=   cls.key_handler.get(__UpperCAmelCase							)
						if handler:
									SCREAMING_SNAKE_CASE__										=   char
									return handler(cls							)
						else:
									return None

def 						A							(  cls     ):




			'''simple docstring'''


			return KeyHandler(cls.__name__		,					cls.__bases__		,					cls.__dict__.copy()     )


