

'''simple docstring'''




import unittest

import numpy as np
import torch
from transformers import CLIPTextConfig, CLIPTextModel

from diffusers import DDIMScheduler, LDMPipeline, UNetaDModel, VQModel
from diffusers.utils.testing_utils import enable_full_determinism, require_torch, slow, torch_device


enable_full_determinism()



class 							lowercase_						(		unittest.TestCase			):

  '''simple docstring'''





  @property
  def     snake_case_      (  self  )				->					Tuple:


    """simple docstring"""
    torch.manual_seed(0  )
    UpperCAmelCase    				=						UNetaDModel(
        block_out_channels=(3_2, 6_4) ,			layers_per_block=2 ,			sample_size=3_2 ,			in_channels=3 ,			out_channels=3 ,			down_block_types=('DownBlock2D', 'AttnDownBlock2D') ,			up_block_types=('AttnUpBlock2D', 'UpBlock2D') ,			)
    return model
  @property
  def     snake_case_      (  self  )				->					Optional[int]:


    """simple docstring"""
    torch.manual_seed(0  )
    UpperCAmelCase    				=						VQModel(
        block_out_channels=[3_2, 6_4] ,			in_channels=3 ,			out_channels=3 ,			down_block_types=['DownEncoderBlock2D', 'DownEncoderBlock2D'] ,			up_block_types=['UpDecoderBlock2D', 'UpDecoderBlock2D'] ,			latent_channels=3 ,			)
    return model
  @property
  def     snake_case_      (  self  )				->					Optional[Any]:


    """simple docstring"""
    torch.manual_seed(0  )
    UpperCAmelCase    				=						CLIPTextConfig(
        bos_token_id=0 ,			eos_token_id=2 ,			hidden_size=3_2 ,			intermediate_size=3_7 ,			layer_norm_eps=1E-05 ,			num_attention_heads=4 ,			num_hidden_layers=5 ,			pad_token_id=1 ,			vocab_size=1_0_0_0 ,			)
    return CLIPTextModel(a_  )





  def     snake_case_      (  self  )				->					List[str]:


    """simple docstring"""
    UpperCAmelCase    				=						self.dummy_uncond_unet
    UpperCAmelCase    				=						DDIMScheduler()
    UpperCAmelCase    				=						self.dummy_vq_model

    UpperCAmelCase    				=						LDMPipeline(unet=a_ ,			vqvae=a_ ,			scheduler=a_  )
    ldm.to(a_  )
    ldm.set_progress_bar_config(disable=a_  )

    UpperCAmelCase    				=						torch.manual_seed(0  )
    UpperCAmelCase    				=						ldm(generator=a_ ,			num_inference_steps=2 ,			output_type='numpy'  ).images

    UpperCAmelCase    				=						torch.manual_seed(0  )
    UpperCAmelCase    				=						ldm(generator=a_ ,			num_inference_steps=2 ,			output_type='numpy' ,			return_dict=a_  )[0]

    UpperCAmelCase    				=						image[0, -3:, -3:, -1]
    UpperCAmelCase    				=						image_from_tuple[0, -3:, -3:, -1]

    assert image.shape == (1, 6_4, 6_4, 3)
    UpperCAmelCase    				=						np.array([0.8512, 0.818, 0.6411, 0.6808, 0.4465, 0.5618, 0.46, 0.6231, 0.5172]  )
    UpperCAmelCase    				=						1E-2 if torch_device != 'mps' else 3E-2

    assert np.abs(image_slice.flatten() - expected_slice  ).max() < tolerance
    assert np.abs(image_from_tuple_slice.flatten() - expected_slice  ).max() < tolerance




@slow
@require_torch
class 							lowercase_						(		unittest.TestCase			):

  '''simple docstring'''





  def     snake_case_      (  self  )				->					Tuple:


    """simple docstring"""
    UpperCAmelCase    				=						LDMPipeline.from_pretrained('CompVis/ldm-celebahq-256'  )
    ldm.to(a_  )
    ldm.set_progress_bar_config(disable=a_  )

    UpperCAmelCase    				=						torch.manual_seed(0  )
    UpperCAmelCase    				=						ldm(generator=a_ ,			num_inference_steps=5 ,			output_type='numpy'  ).images

    UpperCAmelCase    				=						image[0, -3:, -3:, -1]

    assert image.shape == (1, 2_5_6, 2_5_6, 3)
    UpperCAmelCase    				=						np.array([0.4399, 0.4_4975, 0.4_6825, 0.474, 0.4359, 0.4581, 0.4_5095, 0.4341, 0.4447]  )
    UpperCAmelCase    				=						1E-2 if torch_device != 'mps' else 3E-2

    assert np.abs(image_slice.flatten() - expected_slice  ).max() < tolerance





'''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_  )]


