




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      :						Tuple							=  logging.get_logger(__name__)


@add_end_docstrings(UpperCAmelCase__	)
class 							UpperCamelCase_       (					UpperCAmelCase__	):

    '''simple docstring'''






    def __init__(     self :      str   ,					*UpperCAmelCase__ :      Tuple   ,					**UpperCAmelCase__ :      int)		->Optional[int]:





           '''simple docstring'''

           super().__init__(*UpperCAmelCase__   ,					**UpperCAmelCase__)
           requires_backends(self   ,					'''decord''')
           self.check_model_type(UpperCAmelCase__)


    def 			SCREAMING_SNAKE_CASE				(     self :      Tuple   ,					UpperCAmelCase__ :      List[Any]=None   ,					UpperCAmelCase__ :      Optional[Any]=None   ,					UpperCAmelCase__ :      List[Any]=None)		->Dict:





           '''simple docstring'''

           A__ 					=       {}
           if frame_sampling_rate is not None:
                  A__ 					=       frame_sampling_rate
           if num_frames is not None:
                  A__ 					=       num_frames

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


    def __call__(     self :      str   ,					UpperCAmelCase__ :      Union[str, List[str]]   ,					**UpperCAmelCase__ :      Optional[int])		->Dict:





           '''simple docstring'''

           return super().__call__(UpperCAmelCase__   ,					**UpperCAmelCase__)


    def 			SCREAMING_SNAKE_CASE				(     self :      Tuple   ,					UpperCAmelCase__ :      Optional[int]   ,					UpperCAmelCase__ :      Tuple=None   ,					UpperCAmelCase__ :      List[Any]=1)		->Any:





           '''simple docstring'''

           if num_frames is None:
                  A__ 					=       self.model.config.num_frames

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

           A__ 					=       VideoReader(UpperCAmelCase__)
           videoreader.seek(0)

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

           A__ 					=       videoreader.get_batch(UpperCAmelCase__).asnumpy()
           A__ 					=       list(UpperCAmelCase__)

           A__ 					=       self.image_processor(UpperCAmelCase__   ,					return_tensors=self.framework)
           return model_inputs


    def 			SCREAMING_SNAKE_CASE				(     self :      Any   ,					UpperCAmelCase__ :      Dict)		->int:





           '''simple docstring'''

           A__ 					=       self.model(**UpperCAmelCase__)
           return model_outputs





    def 			SCREAMING_SNAKE_CASE				(     self :      int   ,					UpperCAmelCase__ :      Optional[int]   ,					UpperCAmelCase__ :      str=5)		->Optional[int]:





           '''simple docstring'''

           if top_k > self.model.config.num_labels:
                  A__ 					=       self.model.config.num_labels

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

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



'''simple docstring'''






import unittest

from diffusers.pipelines.pipeline_utils import is_safetensors_compatible







class lowercase_       (		unittest.TestCase			):







   """simple docstring"""

   def 	__UpperCAmelCase   (      self      :			Optional[Any]     )  ->			Dict:
        _A    		=  [
            'safety_checker/pytorch_model.bin',
            'safety_checker/model.safetensors',
            'vae/diffusion_pytorch_model.bin',
            'vae/diffusion_pytorch_model.safetensors',
            'text_encoder/pytorch_model.bin',
            'text_encoder/model.safetensors',
            'unet/diffusion_pytorch_model.bin',
            'unet/diffusion_pytorch_model.safetensors',
        ]
        self.assertTrue(is_safetensors_compatible(UpperCamelCase__     )     )
   def 	__UpperCAmelCase   (      self      :			Union[str, Any]     )  ->			Optional[Any]:
        _A    		=  [
            'unet/diffusion_pytorch_model.bin',
            'unet/diffusion_pytorch_model.safetensors',
        ]
        self.assertTrue(is_safetensors_compatible(UpperCamelCase__     )     )
   def 	__UpperCAmelCase   (      self      :			str     )  ->			Union[str, Any]:
        _A    		=  [
            'safety_checker/pytorch_model.bin',
            'safety_checker/model.safetensors',
            'vae/diffusion_pytorch_model.bin',
            'vae/diffusion_pytorch_model.safetensors',
            'text_encoder/pytorch_model.bin',
            'text_encoder/model.safetensors',
            'unet/diffusion_pytorch_model.bin',
            # Removed: 'unet/diffusion_pytorch_model.safetensors',
        ]
        self.assertFalse(is_safetensors_compatible(UpperCamelCase__     )     )
   def 	__UpperCAmelCase   (      self      :			int     )  ->			Dict:
        _A    		=  [
            'text_encoder/pytorch_model.bin',
            'text_encoder/model.safetensors',
        ]
        self.assertTrue(is_safetensors_compatible(UpperCamelCase__     )     )
   def 	__UpperCAmelCase   (      self      :			Union[str, Any]     )  ->			str:
        _A    		=  [
            'safety_checker/pytorch_model.bin',
            'safety_checker/model.safetensors',
            'vae/diffusion_pytorch_model.bin',
            'vae/diffusion_pytorch_model.safetensors',
            'text_encoder/pytorch_model.bin',
            # Removed: 'text_encoder/model.safetensors',
            'unet/diffusion_pytorch_model.bin',
            'unet/diffusion_pytorch_model.safetensors',
        ]
        self.assertFalse(is_safetensors_compatible(UpperCamelCase__     )     )
   def 	__UpperCAmelCase   (      self      :			str     )  ->			Dict:
        _A    		=  [
            'safety_checker/pytorch_model.fp16.bin',
            'safety_checker/model.fp16.safetensors',
            'vae/diffusion_pytorch_model.fp16.bin',
            'vae/diffusion_pytorch_model.fp16.safetensors',
            'text_encoder/pytorch_model.fp16.bin',
            'text_encoder/model.fp16.safetensors',
            'unet/diffusion_pytorch_model.fp16.bin',
            'unet/diffusion_pytorch_model.fp16.safetensors',
        ]
        _A    		=  'fp16'
        self.assertTrue(is_safetensors_compatible(UpperCamelCase__, variant=UpperCamelCase__     )     )
   def 	__UpperCAmelCase   (      self      :			Optional[int]     )  ->			List[str]:
        _A    		=  [
            'unet/diffusion_pytorch_model.fp16.bin',
            'unet/diffusion_pytorch_model.fp16.safetensors',
        ]
        _A    		=  'fp16'
        self.assertTrue(is_safetensors_compatible(UpperCamelCase__, variant=UpperCamelCase__     )     )
   def 	__UpperCAmelCase   (      self      :			List[str]     )  ->			List[str]:
        # pass variant but use the non-variant filenames
        _A    		=  [
            'unet/diffusion_pytorch_model.bin',
            'unet/diffusion_pytorch_model.safetensors',
        ]
        _A    		=  'fp16'
        self.assertTrue(is_safetensors_compatible(UpperCamelCase__, variant=UpperCamelCase__     )     )
   def 	__UpperCAmelCase   (      self      :			List[Any]     )  ->			Optional[Any]:
        _A    		=  [
            'safety_checker/pytorch_model.fp16.bin',
            'safety_checker/model.fp16.safetensors',
            'vae/diffusion_pytorch_model.fp16.bin',
            'vae/diffusion_pytorch_model.fp16.safetensors',
            'text_encoder/pytorch_model.fp16.bin',
            'text_encoder/model.fp16.safetensors',
            'unet/diffusion_pytorch_model.fp16.bin',
            # Removed: 'unet/diffusion_pytorch_model.fp16.safetensors',
        ]
        _A    		=  'fp16'
        self.assertFalse(is_safetensors_compatible(UpperCamelCase__, variant=UpperCamelCase__     )     )
   def 	__UpperCAmelCase   (      self      :			Dict     )  ->			Union[str, Any]:
        _A    		=  [
            'text_encoder/pytorch_model.fp16.bin',
            'text_encoder/model.fp16.safetensors',
        ]
        _A    		=  'fp16'
        self.assertTrue(is_safetensors_compatible(UpperCamelCase__, variant=UpperCamelCase__     )     )
   def 	__UpperCAmelCase   (      self      :			Tuple     )  ->			str:
        # pass variant but use the non-variant filenames
        _A    		=  [
            'text_encoder/pytorch_model.bin',
            'text_encoder/model.safetensors',
        ]
        _A    		=  'fp16'
        self.assertTrue(is_safetensors_compatible(UpperCamelCase__, variant=UpperCamelCase__     )     )

   def 	__UpperCAmelCase   (      self      :			List[Any]     )  ->			int:
        _A    		=  [
            'safety_checker/pytorch_model.fp16.bin',
            'safety_checker/model.fp16.safetensors',
            'vae/diffusion_pytorch_model.fp16.bin',
            'vae/diffusion_pytorch_model.fp16.safetensors',
            'text_encoder/pytorch_model.fp16.bin',
            # 'text_encoder/model.fp16.safetensors',
            'unet/diffusion_pytorch_model.fp16.bin',
            'unet/diffusion_pytorch_model.fp16.safetensors',
        ]
        _A    		=  'fp16'
        self.assertFalse(is_safetensors_compatible(UpperCamelCase__, variant=UpperCamelCase__     )     )



