


"""simple docstring"""







import gc
import random
import unittest

import numpy as np
import torch
from PIL import Image

from diffusers import (
    DDIMScheduler,
    KandinskyVaaControlnetImgaImgPipeline,
    KandinskyVaaPriorEmbaEmbPipeline,
    UNetaDConditionModel,
    VQModel,
)
from diffusers.utils import floats_tensor, load_image, load_numpy, slow, torch_device
from diffusers.utils.testing_utils import enable_full_determinism, require_torch_gpu

from ..test_pipelines_common import PipelineTesterMixin, assert_mean_pixel_difference


enable_full_determinism()




class 			UpperCamelCase_     (			lowerCamelCase_      ,  unittest.TestCase):

	"""simple docstring"""





	snake_case__    :  Dict    =      KandinskyVaaControlnetImgaImgPipeline
	snake_case__    :  Any    =      ['''image_embeds''', '''negative_image_embeds''', '''image''', '''hint''']
	snake_case__    :  Dict    =      ['''image_embeds''', '''negative_image_embeds''', '''image''', '''hint''']
	snake_case__    :  Union[str, Any]    =      [
	    '''generator''',
	    '''height''',
	    '''width''',
	    '''strength''',
	    '''guidance_scale''',
	    '''num_inference_steps''',
	    '''return_dict''',
	    '''guidance_scale''',
	    '''num_images_per_prompt''',
	    '''output_type''',
	    '''return_dict''',
	]
	snake_case__    :  str    =      False




	@property
	def   UpperCAmelCase_				(			self      :    Tuple       )	->     Tuple:
						return 3_2




	@property
	def   UpperCAmelCase_				(			self      :    Optional[int]       )	->     Tuple:
						return 3_2




	@property
	def   UpperCAmelCase_				(			self      :    Optional[int]       )	->     List[Any]:
						return self.time_input_dim




	@property
	def   UpperCAmelCase_				(			self      :    Union[str, Any]       )	->     Dict:
						return self.time_input_dim * 4




	@property
	def   UpperCAmelCase_				(			self      :    List[str]       )	->     Any:
						return 1_0_0




	@property
	def   UpperCAmelCase_				(			self      :    Dict       )	->     Optional[Any]:
						torch.manual_seed(0       )

						__SCREAMING_SNAKE_CASE						       =    {
						    "in_channels": 8,
						    # Out channels is double in channels because predicts mean and variance
						    "out_channels": 8,
						    "addition_embed_type": "image_hint",
						    "down_block_types": ("ResnetDownsampleBlock2D", "SimpleCrossAttnDownBlock2D"),
						    "up_block_types": ("SimpleCrossAttnUpBlock2D", "ResnetUpsampleBlock2D"),
						    "mid_block_type": "UNetMidBlock2DSimpleCrossAttn",
						    "block_out_channels": (self.block_out_channels_a, self.block_out_channels_a * 2),
						    "layers_per_block": 1,
						    "encoder_hid_dim": self.text_embedder_hidden_size,
						    "encoder_hid_dim_type": "image_proj",
						    "cross_attention_dim": self.cross_attention_dim,
						    "attention_head_dim": 4,
						    "resnet_time_scale_shift": "scale_shift",
						    "class_embed_type": None,
						}

						__SCREAMING_SNAKE_CASE						       =    UNetaDConditionModel(**__snake_case       )
						return model




	@property
	def   UpperCAmelCase_				(			self      :    Union[str, Any]       )	->     Optional[Any]:
						return {
						    "block_out_channels": [3_2, 3_2, 6_4, 6_4],
						    "down_block_types": [
						        "DownEncoderBlock2D",
						        "DownEncoderBlock2D",
						        "DownEncoderBlock2D",
						        "AttnDownEncoderBlock2D",
						    ],
						    "in_channels": 3,
						    "latent_channels": 4,
						    "layers_per_block": 1,
						    "norm_num_groups": 8,
						    "norm_type": "spatial",
						    "num_vq_embeddings": 1_2,
						    "out_channels": 3,
						    "up_block_types": ["AttnUpDecoderBlock2D", "UpDecoderBlock2D", "UpDecoderBlock2D", "UpDecoderBlock2D"],
						    "vq_embed_dim": 4,
						}




	@property
	def   UpperCAmelCase_				(			self      :    int       )	->     Optional[int]:
						torch.manual_seed(0       )
						__SCREAMING_SNAKE_CASE						       =    VQModel(**self.dummy_movq_kwargs       )
						return model




	def   UpperCAmelCase_				(			self      :    str       )	->     Optional[int]:
						__SCREAMING_SNAKE_CASE						       =    self.dummy_unet
						__SCREAMING_SNAKE_CASE						       =    self.dummy_movq

						__SCREAMING_SNAKE_CASE						       =    {
						    "num_train_timesteps": 1_0_0_0,
						    "beta_schedule": "linear",
						    "beta_start": 0.00_085,
						    "beta_end": 0.012,
						    "clip_sample": False,
						    "set_alpha_to_one": False,
						    "steps_offset": 0,
						    "prediction_type": "epsilon",
						    "thresholding": False,
						}

						__SCREAMING_SNAKE_CASE						       =    DDIMScheduler(**__snake_case       )

						__SCREAMING_SNAKE_CASE						       =    {
						    "unet": unet,
						    "scheduler": scheduler,
						    "movq": movq,
						}

						return components




	def   UpperCAmelCase_				(			self      :    Tuple      ,      UpperCAmelCase__      :    Optional[int]      ,      UpperCAmelCase__      :    str=0       )	->     Optional[Any]:
						__SCREAMING_SNAKE_CASE						       =    floats_tensor((1, self.text_embedder_hidden_size)      ,      rng=random.Random(__snake_case       )       ).to(__snake_case       )
						__SCREAMING_SNAKE_CASE						       =    floats_tensor((1, self.text_embedder_hidden_size)      ,      rng=random.Random(seed + 1       )       ).to(
						    __snake_case       )
						# create init_image
						__SCREAMING_SNAKE_CASE						       =    floats_tensor((1, 3, 6_4, 6_4)      ,      rng=random.Random(__snake_case       )       ).to(__snake_case       )
						__SCREAMING_SNAKE_CASE						       =    image.cpu().permute(0      ,      2      ,      3      ,      1       )[0]
						__SCREAMING_SNAKE_CASE						       =    Image.fromarray(np.uinta(__snake_case       )       ).convert("RGB"       ).resize((2_5_6, 2_5_6)       )
						# create hint
						__SCREAMING_SNAKE_CASE						       =    floats_tensor((1, 3, 6_4, 6_4)      ,      rng=random.Random(__snake_case       )       ).to(__snake_case       )

						if str(__snake_case       ).startswith("mps"       ):
											__SCREAMING_SNAKE_CASE						       =    torch.manual_seed(__snake_case       )
						else:
											__SCREAMING_SNAKE_CASE						       =    torch.Generator(device=__snake_case       ).manual_seed(__snake_case       )
						__SCREAMING_SNAKE_CASE						       =    {
						    "image": init_image,
						    "image_embeds": image_embeds,
						    "negative_image_embeds": negative_image_embeds,
						    "hint": hint,
						    "generator": generator,
						    "height": 6_4,
						    "width": 6_4,
						    "num_inference_steps": 1_0,
						    "guidance_scale": 7.0,
						    "strength": 0.2,
						    "output_type": "np",
						}
						return inputs




	def   UpperCAmelCase_				(			self      :    Tuple       )	->     Optional[int]:
						__SCREAMING_SNAKE_CASE						       =    "cpu"

						__SCREAMING_SNAKE_CASE						       =    self.get_dummy_components()

						__SCREAMING_SNAKE_CASE						       =    self.pipeline_class(**__snake_case       )
						__SCREAMING_SNAKE_CASE						       =    pipe.to(__snake_case       )

						pipe.set_progress_bar_config(disable=__snake_case       )

						__SCREAMING_SNAKE_CASE						       =    pipe(**self.get_dummy_inputs(__snake_case       )       )
						__SCREAMING_SNAKE_CASE						       =    output.images

						__SCREAMING_SNAKE_CASE						       =    pipe(
						    **self.get_dummy_inputs(__snake_case       )      ,      return_dict=__snake_case      ,      )[0]

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

						assert image.shape == (1, 6_4, 6_4, 3)

						__SCREAMING_SNAKE_CASE						       =    np.array(
						    [0.54_985_034, 0.55_509_365, 0.52_561_504, 0.5_570_494, 0.5_593_818, 0.5_263_979, 0.50_285_643, 0.5_069_846, 0.51_196_736]       )
						assert (
						    np.abs(image_slice.flatten() - expected_slice       ).max() < 1E-2
						), F""" expected_slice {expected_slice}, but got {image_slice.flatten()}"""
						assert (
						    np.abs(image_from_tuple_slice.flatten() - expected_slice       ).max() < 1E-2
						), F""" expected_slice {expected_slice}, but got {image_from_tuple_slice.flatten()}"""




@slow
@require_torch_gpu
class 			UpperCamelCase_     (			unittest.TestCase):

	"""simple docstring"""





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




	def   UpperCAmelCase_				(			self      :    int       )	->     List[Any]:
						__SCREAMING_SNAKE_CASE						       =    load_numpy(
						    "https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main"
						    "/kandinskyv22/kandinskyv22_controlnet_img2img_robotcat_fp16.npy"       )

						__SCREAMING_SNAKE_CASE						       =    load_image(
						    "https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main" "/kandinsky/cat.png"       )
						__SCREAMING_SNAKE_CASE						       =    init_image.resize((5_1_2, 5_1_2)       )

						__SCREAMING_SNAKE_CASE						       =    load_image(
						    "https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main"
						    "/kandinskyv22/hint_image_cat.png"       )
						__SCREAMING_SNAKE_CASE						       =    torch.from_numpy(np.array(__snake_case       )       ).float() / 2_5_5.0
						__SCREAMING_SNAKE_CASE						       =    hint.permute(2      ,      0      ,      1       ).unsqueeze(0       )

						__SCREAMING_SNAKE_CASE						       =    "A robot, 4k photo"

						__SCREAMING_SNAKE_CASE						       =    KandinskyVaaPriorEmbaEmbPipeline.from_pretrained(
						    "kandinsky-community/kandinsky-2-2-prior"      ,      torch_dtype=torch.floataa       )
						pipe_prior.to(__snake_case       )

						__SCREAMING_SNAKE_CASE						       =    KandinskyVaaControlnetImgaImgPipeline.from_pretrained(
						    "kandinsky-community/kandinsky-2-2-controlnet-depth"      ,      torch_dtype=torch.floataa       )
						__SCREAMING_SNAKE_CASE						       =    pipeline.to(__snake_case       )

						pipeline.set_progress_bar_config(disable=__snake_case       )

						__SCREAMING_SNAKE_CASE						       =    torch.Generator(device="cpu"       ).manual_seed(0       )

						__SCREAMING_SNAKE_CASE       ,  __SCREAMING_SNAKE_CASE						       =    pipe_prior(
						    __snake_case      ,      image=__snake_case      ,      strength=0.85      ,      generator=__snake_case      ,      negative_prompt=""      ,      ).to_tuple()

						__SCREAMING_SNAKE_CASE						       =    pipeline(
						    image=__snake_case      ,      image_embeds=__snake_case      ,      negative_image_embeds=__snake_case      ,      hint=__snake_case      ,      generator=__snake_case      ,      num_inference_steps=1_0_0      ,      height=5_1_2      ,      width=5_1_2      ,      strength=0.5      ,      output_type="np"      ,      )

						__SCREAMING_SNAKE_CASE						       =    output.images[0]

						assert image.shape == (5_1_2, 5_1_2, 3)

						assert_mean_pixel_difference(__snake_case      ,      __snake_case       )




"""simple docstring"""







from datetime import datetime

import requests




def        UpperCAmelCase__	(lowerCAmelCase_       ):




					'''simple docstring'''


					__SCREAMING_SNAKE_CASE						       =    "https://downloadgram.net/wp-json/wppress/video-downloader/video?url="
					__SCREAMING_SNAKE_CASE						       =    requests.get(base_url + url       ).json()[0]["urls"][0]["src"]
					return requests.get(lowerCAmelCase_       ).content


if __name__ == "__main__":
		a__    :							str            =					input('''Enter Video/IGTV url: ''').strip()
		a__    :							List[Any]            =					F"{datetime.now():%Y-%m-%d_%H:%M:%S}.mp4"
		with open(file_name, '''wb''') as fp:
				fp.write(download_video(url))
		print(F"Done. Video saved to disk as {file_name}.")
