"""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					:					Union[str, Any]=			logging.get_logger(__name__)

@add_end_docstrings(lowercase						)
class 		UpperCamelCase      (   lowercase						):



				def __init__(self    :				List[Any]     ,  *_A    :				Any     ,  **_A    :				Union[str, Any])   ->     Any:
					super().__init__(*_A     ,  **_A)
					requires_backends(self     ,  'decord')
					self.check_model_type(_A)



				def  _lowercase					(self    :				Optional[Any]     ,  _A    :				Any=None     ,  _A    :				Any=None     ,  _A    :				Any=None)   ->     str:
					__snake_case      :							Dict						=			{}
					if frame_sampling_rate is not None:
						__snake_case      :							Tuple						=			frame_sampling_rate
					if num_frames is not None:
						__snake_case      :							str						=			num_frames

					__snake_case      :							Union[str, Any]						=			{}
					if top_k is not None:
						__snake_case      :							Optional[Any]						=			top_k
					return preprocess_params, {}, postprocess_params



				def __call__(self    :				Dict     ,  _A    :				Union[str, List[str]]     ,  **_A    :				List[str])   ->     List[Any]:
					return super().__call__(_A     ,  **_A)



				def  _lowercase					(self    :				Tuple     ,  _A    :				int     ,  _A    :				Tuple=None     ,  _A    :				Any=1)   ->     Any:
					if num_frames is None:
						__snake_case      :							Any						=			self.model.config.num_frames

					if video.startswith('http://') or video.startswith('https://'):
						__snake_case      :							int						=			BytesIO(requests.get(_A).content)

					__snake_case      :							Optional[Any]						=			VideoReader(_A)
					videoreader.seek(0)

					__snake_case      :							Union[str, Any]						=			0
					__snake_case      :							Tuple						=			num_frames * frame_sampling_rate - 1
					__snake_case      :							Any						=			np.linspace(_A     ,  _A     ,  num=_A     ,  dtype=np.intaa)

					__snake_case      :							List[Any]						=			videoreader.get_batch(_A).asnumpy()
					__snake_case      :							Tuple						=			list(_A)

					__snake_case      :							int						=			self.image_processor(_A     ,  return_tensors=self.framework)
					return model_inputs



				def  _lowercase					(self    :				int     ,  _A    :				Optional[Any])   ->     int:
					__snake_case      :							Optional[Any]						=			self.model(**_A)
					return model_outputs






				def  _lowercase					(self    :				Any     ,  _A    :				Optional[Any]     ,  _A    :				str=5)   ->     Dict:
					if top_k > self.model.config.num_labels:
						__snake_case      :							Optional[Any]						=			self.model.config.num_labels

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

					__snake_case      :							List[Any]						=			scores.tolist()
					__snake_case      :							Tuple						=			ids.tolist()
					return [{"score": score, "label": self.model.config.idalabel[_id]} for score, _id in zip(_A     ,  _A)]






"""simple docstring"""







import gc
import tempfile
import unittest

import numpy as np
import torch

from diffusers import VersatileDiffusionPipeline
from diffusers.utils.testing_utils import load_image, nightly, require_torch_gpu, torch_device


_a					:					int=			False

class 		UpperCamelCase      (   unittest.TestCase						):
				pass







@nightly
@require_torch_gpu
class 		UpperCamelCase      (   unittest.TestCase						):



				def  _lowercase					(self    :				Optional[Any])   ->     Tuple:
					# clean up the VRAM after each test
					super().tearDown()
					gc.collect()
					torch.cuda.empty_cache()



				def  _lowercase					(self    :				int)   ->     Tuple:
					__snake_case      :							str						=			VersatileDiffusionPipeline.from_pretrained('shi-labs/versatile-diffusion'     ,  torch_dtype=torch.floataa)
					pipe.to(_A)
					pipe.set_progress_bar_config(disable=_A)

					__snake_case      :							int						=			load_image(
					    'https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/versatile_diffusion/benz.jpg')

					__snake_case      :							List[Any]						=			torch.manual_seed(0)
					__snake_case      :							Optional[int]						=			pipe.dual_guided(
					    prompt='first prompt'     ,  image=_A     ,  text_to_image_strength=0.75     ,  generator=_A     ,  guidance_scale=7.5     ,  num_inference_steps=2     ,  output_type='numpy'     ,  ).images

					with tempfile.TemporaryDirectory() as tmpdirname:
						pipe.save_pretrained(_A)
						__snake_case      :							Optional[Any]						=			VersatileDiffusionPipeline.from_pretrained(_A     ,  torch_dtype=torch.floataa)
					pipe.to(_A)
					pipe.set_progress_bar_config(disable=_A)

					__snake_case      :							Tuple						=			generator.manual_seed(0)
					__snake_case      :							int						=			pipe.dual_guided(
					    prompt='first prompt'     ,  image=_A     ,  text_to_image_strength=0.75     ,  generator=_A     ,  guidance_scale=7.5     ,  num_inference_steps=2     ,  output_type='numpy'     ,  ).images

					assert np.abs(image - new_image).sum() < 1E-5, "Models don't have the same forward pass"






				def  _lowercase					(self    :				Optional[Any])   ->     Optional[int]:
					__snake_case      :							Optional[Any]						=			VersatileDiffusionPipeline.from_pretrained('shi-labs/versatile-diffusion'     ,  torch_dtype=torch.floataa)
					pipe.to(_A)
					pipe.set_progress_bar_config(disable=_A)

					__snake_case      :							Tuple						=			'cyberpunk 2077'
					__snake_case      :							int						=			load_image(
					    'https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/versatile_diffusion/benz.jpg')
					__snake_case      :							str						=			torch.manual_seed(0)
					__snake_case      :							Union[str, Any]						=			pipe.dual_guided(
					    prompt=_A     ,  image=_A     ,  text_to_image_strength=0.75     ,  generator=_A     ,  guidance_scale=7.5     ,  num_inference_steps=50     ,  output_type='numpy'     ,  ).images

					__snake_case      :							Optional[Any]						=			image[0, 2_53:2_56, 2_53:2_56, -1]

					assert image.shape == (1, 5_12, 5_12, 3)
					__snake_case      :							Tuple						=			np.array([0.1_448, 0.1_619, 0.1_741, 0.1_086, 0.1_147, 0.1_128, 0.1_199, 0.1_165, 0.1_001])

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

					__snake_case      :							List[str]						=			'A painting of a squirrel eating a burger '
					__snake_case      :							str						=			torch.manual_seed(0)
					__snake_case      :							str						=			pipe.text_to_image(
					    prompt=_A     ,  generator=_A     ,  guidance_scale=7.5     ,  num_inference_steps=50     ,  output_type='numpy').images

					__snake_case      :							str						=			image[0, 2_53:2_56, 2_53:2_56, -1]

					assert image.shape == (1, 5_12, 5_12, 3)
					__snake_case      :							Any						=			np.array([0.3_367, 0.3_169, 0.2_656, 0.3_870, 0.4_790, 0.3_796, 0.4_009, 0.4_878, 0.4_778])

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

					__snake_case      :							List[str]						=			pipe.image_variation(_A     ,  generator=_A     ,  output_type='numpy').images

					__snake_case      :							Dict						=			image[0, 2_53:2_56, 2_53:2_56, -1]

					assert image.shape == (1, 5_12, 5_12, 3)
					__snake_case      :							List[Any]						=			np.array([0.3_076, 0.3_123, 0.3_284, 0.3_782, 0.3_770, 0.3_894, 0.4_297, 0.4_331, 0.4_456])

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





