





import gc
import random
import unittest

import numpy as np
import torch
from transformers import (
    CLIPImageProcessor,
    CLIPTextConfig,
    CLIPTextModelWithProjection,
    CLIPTokenizer,
    CLIPVisionConfig,
    CLIPVisionModelWithProjection,
)

from diffusers import (
    DiffusionPipeline,
    UnCLIPImageVariationPipeline,
    UnCLIPScheduler,
    UNetaDConditionModel,
    UNetaDModel,
)
from diffusers.pipelines.unclip.text_proj import UnCLIPTextProjModel
from diffusers.utils import floats_tensor, load_numpy, slow, torch_device
from diffusers.utils.testing_utils import enable_full_determinism, load_image, require_torch_gpu, skip_mps

from ..pipeline_params import IMAGE_VARIATION_BATCH_PARAMS, IMAGE_VARIATION_PARAMS
from ..test_pipelines_common import PipelineTesterMixin, assert_mean_pixel_difference


enable_full_determinism()



class 					A__   (       __a       ,   unittest.TestCase			):
      _UpperCAmelCase	:int						    =   UnCLIPImageVariationPipeline
      _UpperCAmelCase	:Any						    =   IMAGE_VARIATION_PARAMS - {"""height""", """width""", """guidance_scale"""}
      _UpperCAmelCase	:Any						    =   IMAGE_VARIATION_BATCH_PARAMS

      _UpperCAmelCase	:Dict						    =   [
          """generator""",
          """return_dict""",
          """decoder_num_inference_steps""",
          """super_res_num_inference_steps""",
      ]
      _UpperCAmelCase	:int						    =   False





      @property
      def        __UpperCamelCase( self   ):
       '''simple docstring'''




       return 32





      @property
      def        __UpperCamelCase( self   ):
       '''simple docstring'''




       return 32





      @property
      def        __UpperCamelCase( self   ):
       '''simple docstring'''




       return self.time_input_dim





      @property
      def        __UpperCamelCase( self   ):
       '''simple docstring'''




       return self.time_input_dim * 4





      @property
      def        __UpperCamelCase( self   ):
       '''simple docstring'''




       return 100





      @property
      def        __UpperCamelCase( self   ):
       '''simple docstring'''




       UpperCamelCase	:					str       	=   CLIPTokenizer.from_pretrained("hf-internal-testing/tiny-random-clip"   )
       return tokenizer





      @property
      def        __UpperCamelCase( self   ):
       '''simple docstring'''




       torch.manual_seed(0   )
       UpperCamelCase	:					str       	=   CLIPTextConfig(
           bos_token_id=0       ,							eos_token_id=2       ,							hidden_size=self.text_embedder_hidden_size       ,							projection_dim=self.text_embedder_hidden_size       ,							intermediate_size=37       ,							layer_norm_eps=1e-05       ,							num_attention_heads=4       ,							num_hidden_layers=5       ,							pad_token_id=1       ,							vocab_size=1000       ,							)
       return CLIPTextModelWithProjection(A_   )





      @property
      def        __UpperCamelCase( self   ):
       '''simple docstring'''




       torch.manual_seed(0   )
       UpperCamelCase	:					int       	=   CLIPVisionConfig(
           hidden_size=self.text_embedder_hidden_size       ,							projection_dim=self.text_embedder_hidden_size       ,							num_hidden_layers=5       ,							num_attention_heads=4       ,							image_size=32       ,							intermediate_size=37       ,							patch_size=1       ,							)
       return CLIPVisionModelWithProjection(A_   )





      @property
      def        __UpperCamelCase( self   ):
       '''simple docstring'''




       torch.manual_seed(0   )

       UpperCamelCase	:					int       	=   {
           "clip_embeddings_dim": self.text_embedder_hidden_size,
           "time_embed_dim": self.time_embed_dim,
           "cross_attention_dim": self.cross_attention_dim,
       }

       UpperCamelCase	:					Optional[Any]       	=   UnCLIPTextProjModel(**A_   )
       return model





      @property
      def        __UpperCamelCase( self   ):
       '''simple docstring'''




       torch.manual_seed(0   )

       UpperCamelCase	:					List[str]       	=   {
           "sample_size": 32,
           # RGB in channels
           "in_channels": 3,
           # Out channels is double in channels because predicts mean and variance
           "out_channels": 6,
           "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,
           "cross_attention_dim": self.cross_attention_dim,
           "attention_head_dim": 4,
           "resnet_time_scale_shift": "scale_shift",
           "class_embed_type": "identity",
       }

       UpperCamelCase	:					Optional[Any]       	=   UNetaDConditionModel(**A_   )
       return model





      @property
      def        __UpperCamelCase( self   ):
       '''simple docstring'''




       return {
           "sample_size": 64,
           "layers_per_block": 1,
           "down_block_types": ("ResnetDownsampleBlock2D", "ResnetDownsampleBlock2D"),
           "up_block_types": ("ResnetUpsampleBlock2D", "ResnetUpsampleBlock2D"),
           "block_out_channels": (self.block_out_channels_a, self.block_out_channels_a * 2),
           "in_channels": 6,
           "out_channels": 3,
       }





      @property
      def        __UpperCamelCase( self   ):
       '''simple docstring'''




       torch.manual_seed(0   )

       UpperCamelCase	:					Dict       	=   UNetaDModel(**self.dummy_super_res_kwargs   )
       return model





      @property
      def        __UpperCamelCase( self   ):
       '''simple docstring'''




       torch.manual_seed(1   )

       UpperCamelCase	:					Union[str, Any]       	=   UNetaDModel(**self.dummy_super_res_kwargs   )
       return model





      def        __UpperCamelCase( self   ):
       '''simple docstring'''




       UpperCamelCase	:					Union[str, Any]       	=   self.dummy_decoder
       UpperCamelCase	:					Tuple       	=   self.dummy_text_proj
       UpperCamelCase	:					Dict       	=   self.dummy_text_encoder
       UpperCamelCase	:					str       	=   self.dummy_tokenizer
       UpperCamelCase	:					int       	=   self.dummy_super_res_first
       UpperCamelCase	:					List[str]       	=   self.dummy_super_res_last

       UpperCamelCase	:					str       	=   UnCLIPScheduler(
           variance_type="learned_range"       ,							prediction_type="epsilon"       ,							num_train_timesteps=1000       ,							)

       UpperCamelCase	:					Optional[Any]       	=   UnCLIPScheduler(
           variance_type="fixed_small_log"       ,							prediction_type="epsilon"       ,							num_train_timesteps=1000       ,							)

       UpperCamelCase	:					Optional[Any]       	=   CLIPImageProcessor(crop_size=32       ,							size=32   )

       UpperCamelCase	:					Any       	=   self.dummy_image_encoder

       return {
           "decoder": decoder,
           "text_encoder": text_encoder,
           "tokenizer": tokenizer,
           "text_proj": text_proj,
           "feature_extractor": feature_extractor,
           "image_encoder": image_encoder,
           "super_res_first": super_res_first,
           "super_res_last": super_res_last,
           "decoder_scheduler": decoder_scheduler,
           "super_res_scheduler": super_res_scheduler,
       }





      def        __UpperCamelCase( self       ,							A_       ,							A_=0       ,							A_=True   ):
       '''simple docstring'''




       UpperCamelCase	:					Tuple       	=   floats_tensor((1, 3, 32, 32)       ,							rng=random.Random(A_   )   ).to(A_   )
       if str(A_   ).startswith("mps"   ):
        UpperCamelCase	:					int       	=   torch.manual_seed(A_   )
       else:
        UpperCamelCase	:					Optional[int]       	=   torch.Generator(device=A_   ).manual_seed(A_   )

       if pil_image:
        UpperCamelCase	:					Optional[Any]       	=   input_image * 0.5 + 0.5
        UpperCamelCase	:					Tuple       	=   input_image.clamp(0       ,							1   )
        UpperCamelCase	:					List[Any]       	=   input_image.cpu().permute(0       ,							2       ,							3       ,							1   ).float().numpy()
        UpperCamelCase	:					str       	=   DiffusionPipeline.numpy_to_pil(A_   )[0]

       return {
           "image": input_image,
           "generator": generator,
           "decoder_num_inference_steps": 2,
           "super_res_num_inference_steps": 2,
           "output_type": "np",
       }





      def        __UpperCamelCase( self   ):
       '''simple docstring'''




       UpperCamelCase	:					Dict       	=   "cpu"

       UpperCamelCase	:					List[Any]       	=   self.get_dummy_components()

       UpperCamelCase	:					int       	=   self.pipeline_class(**A_   )
       UpperCamelCase	:					Optional[int]       	=   pipe.to(A_   )

       pipe.set_progress_bar_config(disable=A_   )

       UpperCamelCase	:					List[str]       	=   self.get_dummy_inputs(A_       ,							pil_image=A_   )

       UpperCamelCase	:					Optional[Any]       	=   pipe(**A_   )
       UpperCamelCase	:					Dict       	=   output.images

       UpperCamelCase	:					str       	=   self.get_dummy_inputs(A_       ,							pil_image=A_   )

       UpperCamelCase	:					Dict       	=   pipe(
           **A_       ,							return_dict=A_       ,							)[0]

       UpperCamelCase	:					List[str]       	=   image[0, -3:, -3:, -1]
       UpperCamelCase	:					str       	=   image_from_tuple[0, -3:, -3:, -1]

       assert image.shape == (1, 64, 64, 3)

       UpperCamelCase	:					List[str]       	=   np.array(
           [
               0.99_97,
               0.00_02,
               0.99_97,
               0.99_97,
               0.99_69,
               0.00_23,
               0.99_97,
               0.99_69,
               0.99_70,
           ]   )

       assert np.abs(image_slice.flatten() - expected_slice   ).max() < 1e-2
       assert np.abs(image_from_tuple_slice.flatten() - expected_slice   ).max() < 1e-2





      def        __UpperCamelCase( self   ):
       '''simple docstring'''




       UpperCamelCase	:					Optional[int]       	=   "cpu"

       UpperCamelCase	:					Optional[int]       	=   self.get_dummy_components()

       UpperCamelCase	:					Optional[int]       	=   self.pipeline_class(**A_   )
       UpperCamelCase	:					str       	=   pipe.to(A_   )

       pipe.set_progress_bar_config(disable=A_   )

       UpperCamelCase	:					Optional[Any]       	=   self.get_dummy_inputs(A_       ,							pil_image=A_   )

       UpperCamelCase	:					str       	=   pipe(**A_   )
       UpperCamelCase	:					Tuple       	=   output.images

       UpperCamelCase	:					Optional[int]       	=   self.get_dummy_inputs(A_       ,							pil_image=A_   )

       UpperCamelCase	:					Tuple       	=   pipe(
           **A_       ,							return_dict=A_       ,							)[0]

       UpperCamelCase	:					Optional[int]       	=   image[0, -3:, -3:, -1]
       UpperCamelCase	:					Tuple       	=   image_from_tuple[0, -3:, -3:, -1]

       assert image.shape == (1, 64, 64, 3)

       UpperCamelCase	:					Dict       	=   np.array([0.99_97, 0.00_03, 0.99_97, 0.99_97, 0.99_70, 0.00_24, 0.99_97, 0.99_71, 0.99_71]   )

       assert np.abs(image_slice.flatten() - expected_slice   ).max() < 1e-2
       assert np.abs(image_from_tuple_slice.flatten() - expected_slice   ).max() < 1e-2





      def        __UpperCamelCase( self   ):
       '''simple docstring'''




       UpperCamelCase	:					str       	=   "cpu"

       UpperCamelCase	:					Optional[Any]       	=   self.get_dummy_components()

       UpperCamelCase	:					int       	=   self.pipeline_class(**A_   )
       UpperCamelCase	:					int       	=   pipe.to(A_   )

       pipe.set_progress_bar_config(disable=A_   )

       UpperCamelCase	:					int       	=   self.get_dummy_inputs(A_       ,							pil_image=A_   )
       UpperCamelCase	:					Optional[int]       	=   [
           pipeline_inputs["image"],
           pipeline_inputs["image"],
       ]

       UpperCamelCase	:					Optional[int]       	=   pipe(**A_   )
       UpperCamelCase	:					int       	=   output.images

       UpperCamelCase	:					Dict       	=   self.get_dummy_inputs(A_       ,							pil_image=A_   )
       UpperCamelCase	:					Union[str, Any]       	=   [
           tuple_pipeline_inputs["image"],
           tuple_pipeline_inputs["image"],
       ]

       UpperCamelCase	:					int       	=   pipe(
           **A_       ,							return_dict=A_       ,							)[0]

       UpperCamelCase	:					int       	=   image[0, -3:, -3:, -1]
       UpperCamelCase	:					Dict       	=   image_from_tuple[0, -3:, -3:, -1]

       assert image.shape == (2, 64, 64, 3)

       UpperCamelCase	:					Union[str, Any]       	=   np.array(
           [
               0.99_97,
               0.99_89,
               0.00_08,
               0.00_21,
               0.99_60,
               0.00_18,
               0.00_14,
               0.00_02,
               0.99_33,
           ]   )

       assert np.abs(image_slice.flatten() - expected_slice   ).max() < 1e-2
       assert np.abs(image_from_tuple_slice.flatten() - expected_slice   ).max() < 1e-2





      def        __UpperCamelCase( self   ):
       '''simple docstring'''




       UpperCamelCase	:					List[str]       	=   torch.device("cpu"   )



       class 					A__   :
             _UpperCAmelCase	:Optional[Any]						    =   1


       UpperCamelCase	:					List[Any]       	=   self.get_dummy_components()

       UpperCamelCase	:					Union[str, Any]       	=   self.pipeline_class(**A_   )
       UpperCamelCase	:					Any       	=   pipe.to(A_   )

       pipe.set_progress_bar_config(disable=A_   )

       UpperCamelCase	:					str       	=   torch.Generator(device=A_   ).manual_seed(0   )
       UpperCamelCase	:					int       	=   pipe.decoder.dtype
       UpperCamelCase	:					str       	=   1

       UpperCamelCase	:					Optional[int]       	=   (
           batch_size,
           pipe.decoder.config.in_channels,
           pipe.decoder.config.sample_size,
           pipe.decoder.config.sample_size,
       )
       UpperCamelCase	:					Tuple       	=   pipe.prepare_latents(
           A_       ,							dtype=A_       ,							device=A_       ,							generator=A_       ,							latents=A_       ,							scheduler=DummyScheduler()   )

       UpperCamelCase	:					Any       	=   (
           batch_size,
           pipe.super_res_first.config.in_channels // 2,
           pipe.super_res_first.config.sample_size,
           pipe.super_res_first.config.sample_size,
       )
       UpperCamelCase	:					Any       	=   pipe.prepare_latents(
           A_       ,							dtype=A_       ,							device=A_       ,							generator=A_       ,							latents=A_       ,							scheduler=DummyScheduler()   )

       UpperCamelCase	:					str       	=   self.get_dummy_inputs(A_       ,							pil_image=A_   )

       UpperCamelCase	:					Union[str, Any]       	=   pipe(
           **A_       ,							decoder_latents=A_       ,							super_res_latents=A_   ).images

       UpperCamelCase	:					Optional[Any]       	=   self.get_dummy_inputs(A_       ,							pil_image=A_   )
       # Don't pass image, instead pass embedding
       UpperCamelCase	:					Union[str, Any]       	=   pipeline_inputs.pop("image"   )
       UpperCamelCase	:					int       	=   pipe.image_encoder(A_   ).image_embeds

       UpperCamelCase	:					Optional[int]       	=   pipe(
           **A_       ,							decoder_latents=A_       ,							super_res_latents=A_       ,							image_embeddings=A_       ,							).images

       # make sure passing text embeddings manually is identical
       assert np.abs(img_out_a - img_out_a   ).max() < 1e-4





      @skip_mps
      def        __UpperCamelCase( self   ):
       '''simple docstring'''




       UpperCamelCase	:					int       	=   torch_device == "cpu"

       # Check is relaxed because there is not a torch 2.0 sliced attention added kv processor
       UpperCamelCase	:					Union[str, Any]       	=   1e-2

       self._test_attention_slicing_forward_pass(
           test_max_difference=A_       ,							expected_max_diff=A_   )





      @skip_mps
      def        __UpperCamelCase( self   ):
       '''simple docstring'''




       UpperCamelCase	:					int       	=   torch_device == "cpu"
       UpperCamelCase	:					List[str]       	=   True
       UpperCamelCase	:					List[Any]       	=   [
           "decoder_num_inference_steps",
           "super_res_num_inference_steps",
       ]

       self._test_inference_batch_single_identical(
           test_max_difference=A_       ,							relax_max_difference=A_       ,							additional_params_copy_to_batched_inputs=A_       ,							)





      def        __UpperCamelCase( self   ):
       '''simple docstring'''




       UpperCamelCase	:					Dict       	=   [
           "decoder_num_inference_steps",
           "super_res_num_inference_steps",
       ]

       if torch_device == "mps":
        # TODO: MPS errors with larger batch sizes
        UpperCamelCase	:					List[Any]       	=   [2, 3]
        self._test_inference_batch_consistent(
            batch_sizes=A_       ,							additional_params_copy_to_batched_inputs=A_       ,							)
       else:
        self._test_inference_batch_consistent(
            additional_params_copy_to_batched_inputs=A_   )





      @skip_mps
      def        __UpperCamelCase( self   ):
       '''simple docstring'''




       return super().test_dict_tuple_outputs_equivalent()





      @skip_mps
      def        __UpperCamelCase( self   ):
       '''simple docstring'''




       return super().test_save_load_local()





      @skip_mps
      def        __UpperCamelCase( self   ):
       '''simple docstring'''




       return super().test_save_load_optional_components()



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





      def        __UpperCamelCase( self   ):
       '''simple docstring'''




       super().tearDown()
       gc.collect()
       torch.cuda.empty_cache()





      def        __UpperCamelCase( self   ):
       '''simple docstring'''




       UpperCamelCase	:					Dict       	=   load_image(
           "https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/unclip/cat.png"   )
       UpperCamelCase	:					str       	=   load_numpy(
           "https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main"
           "/unclip/karlo_v1_alpha_cat_variation_fp16.npy"   )

       UpperCamelCase	:					Dict       	=   UnCLIPImageVariationPipeline.from_pretrained(
           "kakaobrain/karlo-v1-alpha-image-variations"       ,							torch_dtype=torch.floataa   )
       UpperCamelCase	:					List[Any]       	=   pipeline.to(A_   )
       pipeline.set_progress_bar_config(disable=A_   )

       UpperCamelCase	:					Dict       	=   torch.Generator(device="cpu"   ).manual_seed(0   )
       UpperCamelCase	:					Optional[int]       	=   pipeline(
           A_       ,							generator=A_       ,							output_type="np"       ,							)

       UpperCamelCase	:					int       	=   output.images[0]

       assert image.shape == (256, 256, 3)

       assert_mean_pixel_difference(A_       ,							A_       ,							15   )






'''simple docstring'''







from datetime import datetime

import requests
from bsa import BeautifulSoup

if __name__ == "__main__":
   A     =input('Enter image url: ').strip()
   print(f"""Downloading image from {url} ...""")
   A     =BeautifulSoup(requests.get(url).content, 'html.parser')
   # The image URL is in the content field of the first meta tag with property og:image
   A     =soup.find('meta', {'property': 'og:image'})['content']
   A     =requests.get(image_url).content
   A     =f"""{datetime.now():%Y-%m-%d_%H:%M:%S}.jpg"""
   with open(file_name, 'wb') as fp:
      fp.write(image_data)
   print(f"""Done. Image saved to disk as {file_name}.""")




