






import gc
import random
import unittest

import numpy as np
import torch
from PIL import Image
from transformers import CLIPImageProcessor, CLIPVisionConfig, CLIPVisionModelWithProjection

from diffusers import (
    AutoencoderKL,
    DPMSolverMultistepScheduler,
    PNDMScheduler,
    StableDiffusionImageVariationPipeline,
    UNetaDConditionModel,
)
from diffusers.utils import floats_tensor, load_image, load_numpy, nightly, slow, torch_device
from diffusers.utils.testing_utils import enable_full_determinism, require_torch_gpu

from ..pipeline_params import IMAGE_VARIATION_BATCH_PARAMS, IMAGE_VARIATION_PARAMS
from ..test_pipelines_common import PipelineKarrasSchedulerTesterMixin, PipelineLatentTesterMixin, PipelineTesterMixin


enable_full_determinism()
class 							__snake_case(       __UpperCamelCase  ,							__UpperCamelCase  ,							__UpperCamelCase  ,							unittest.TestCase   ):
 lowercase			:							List[Any]		   =		StableDiffusionImageVariationPipeline
 lowercase			:							Union[str, Any]		   =		IMAGE_VARIATION_PARAMS
 lowercase			:							Union[str, Any]		   =		IMAGE_VARIATION_BATCH_PARAMS
 lowercase			:							Optional[Any]		   =		frozenset([]   )
 # TO-DO: update image_params once pipeline is refactored with VaeImageProcessor.preprocess
 lowercase			:							Union[str, Any]		   =		frozenset([]   )



 def     UpperCAmelCase     (						self : List[str]						):

    """simple docstring"""






    torch.manual_seed(0						)
    lowercase__    :						Union[str, Any]       =			UNetaDConditionModel(
        block_out_channels=(32, 64)			,		layers_per_block=2			,		sample_size=32			,		in_channels=4			,		out_channels=4			,		down_block_types=("""DownBlock2D""", """CrossAttnDownBlock2D""")			,		up_block_types=("""CrossAttnUpBlock2D""", """UpBlock2D""")			,		cross_attention_dim=32			,		)
    lowercase__    :						Union[str, Any]       =			PNDMScheduler(skip_prk_steps=UpperCAmelCase__						)
    torch.manual_seed(0						)
    lowercase__    :						Any       =			AutoencoderKL(
        block_out_channels=[32, 64]			,		in_channels=3			,		out_channels=3			,		down_block_types=["""DownEncoderBlock2D""", """DownEncoderBlock2D"""]			,		up_block_types=["""UpDecoderBlock2D""", """UpDecoderBlock2D"""]			,		latent_channels=4			,		)
    torch.manual_seed(0						)
    lowercase__    :						Dict       =			CLIPVisionConfig(
        hidden_size=32			,		projection_dim=32			,		intermediate_size=37			,		layer_norm_eps=1e-05			,		num_attention_heads=4			,		num_hidden_layers=5			,		image_size=32			,		patch_size=4			,		)
    lowercase__    :						Optional[Any]       =			CLIPVisionModelWithProjection(UpperCAmelCase__						)
    lowercase__    :						Tuple       =			CLIPImageProcessor(crop_size=32			,		size=32						)

    lowercase__    :						List[str]       =			{
        """unet""": unet,
        """scheduler""": scheduler,
        """vae""": vae,
        """image_encoder""": image_encoder,
        """feature_extractor""": feature_extractor,
        """safety_checker""": None,
    }
    return components



 def     UpperCAmelCase     (						self : int			,		UpperCAmelCase__ : List[Any]			,		UpperCAmelCase__ : List[Any]=0						):

    """simple docstring"""






    lowercase__    :						str       =			floats_tensor((1, 3, 32, 32)			,		rng=random.Random(UpperCAmelCase__						)						)
    lowercase__    :						Optional[int]       =			image.cpu().permute(0			,		2			,		3			,		1						)[0]
    lowercase__    :						Tuple       =			Image.fromarray(np.uinta(UpperCAmelCase__						)						).convert("""RGB"""						).resize((32, 32)						)
    if str(UpperCAmelCase__						).startswith("""mps"""						):
       lowercase__    :						List[str]       =			torch.manual_seed(UpperCAmelCase__						)
    else:
       lowercase__    :						List[Any]       =			torch.Generator(device=UpperCAmelCase__						).manual_seed(UpperCAmelCase__						)
    lowercase__    :						Tuple       =			{
        """image""": image,
        """generator""": generator,
        """num_inference_steps""": 2,
        """guidance_scale""": 6.0,
        """output_type""": """numpy""",
    }
    return inputs



 def     UpperCAmelCase     (						self : Union[str, Any]						):

    """simple docstring"""






    lowercase__    :						int       =			"""cpu"""  # ensure determinism for the device-dependent torch.Generator
    lowercase__    :						Union[str, Any]       =			self.get_dummy_components()
    lowercase__    :						Optional[Any]       =			StableDiffusionImageVariationPipeline(**UpperCAmelCase__						)
    lowercase__    :						List[str]       =			sd_pipe.to(UpperCAmelCase__						)
    sd_pipe.set_progress_bar_config(disable=UpperCAmelCase__						)

    lowercase__    :						int       =			self.get_dummy_inputs(UpperCAmelCase__						)
    lowercase__    :						str       =			sd_pipe(**UpperCAmelCase__						).images
    lowercase__    :						List[Any]       =			image[0, -3:, -3:, -1]

    assert image.shape == (1, 64, 64, 3)
    lowercase__    :						str       =			np.array([0.5_239, 0.5_723, 0.4_796, 0.5_049, 0.5_550, 0.4_685, 0.5_329, 0.4_891, 0.4_921]						)

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



 def     UpperCAmelCase     (						self : List[str]						):

    """simple docstring"""






    lowercase__    :						Union[str, Any]       =			"""cpu"""  # ensure determinism for the device-dependent torch.Generator
    lowercase__    :						Optional[Any]       =			self.get_dummy_components()
    lowercase__    :						str       =			StableDiffusionImageVariationPipeline(**UpperCAmelCase__						)
    lowercase__    :						Any       =			sd_pipe.to(UpperCAmelCase__						)
    sd_pipe.set_progress_bar_config(disable=UpperCAmelCase__						)

    lowercase__    :						List[str]       =			self.get_dummy_inputs(UpperCAmelCase__						)
    lowercase__    :						Any       =			2 * [inputs["""image"""]]
    lowercase__    :						List[Any]       =			sd_pipe(**UpperCAmelCase__						)

    lowercase__    :						int       =			output.images

    lowercase__    :						Dict       =			image[-1, -3:, -3:, -1]

    assert image.shape == (2, 64, 64, 3)
    lowercase__    :						Optional[Any]       =			np.array([0.6_892, 0.5_637, 0.5_836, 0.5_771, 0.6_254, 0.6_409, 0.5_580, 0.5_569, 0.5_289]						)

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







 def     UpperCAmelCase     (						self : Dict						):

    """simple docstring"""






    super().test_inference_batch_single_identical(expected_max_diff=3e-3						)
@slow
@require_torch_gpu
class 							__snake_case(       unittest.TestCase   ):



 def     UpperCAmelCase     (						self : List[str]						):

    """simple docstring"""






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



 def     UpperCAmelCase     (						self : Any			,		UpperCAmelCase__ : Any			,		UpperCAmelCase__ : Union[str, Any]="cpu"			,		UpperCAmelCase__ : Optional[Any]=torch.floataa			,		UpperCAmelCase__ : Tuple=0						):

    """simple docstring"""






    lowercase__    :						Tuple       =			torch.Generator(device=UpperCAmelCase__						).manual_seed(UpperCAmelCase__						)
    lowercase__    :						Optional[int]       =			load_image(
        """https://huggingface.co/datasets/diffusers/test-arrays/resolve/main"""
        """/stable_diffusion_imgvar/input_image_vermeer.png"""						)
    lowercase__    :						Any       =			np.random.RandomState(UpperCAmelCase__						).standard_normal((1, 4, 64, 64)						)
    lowercase__    :						Any       =			torch.from_numpy(UpperCAmelCase__						).to(device=UpperCAmelCase__			,		dtype=UpperCAmelCase__						)
    lowercase__    :						str       =			{
        """image""": init_image,
        """latents""": latents,
        """generator""": generator,
        """num_inference_steps""": 3,
        """guidance_scale""": 7.5,
        """output_type""": """numpy""",
    }
    return inputs



 def     UpperCAmelCase     (						self : Any						):

    """simple docstring"""






    lowercase__    :						Dict       =			StableDiffusionImageVariationPipeline.from_pretrained(
        """lambdalabs/sd-image-variations-diffusers"""			,		safety_checker=UpperCAmelCase__						)
    lowercase__    :						List[Any]       =			sd_pipe.to(UpperCAmelCase__						)
    sd_pipe.set_progress_bar_config(disable=UpperCAmelCase__						)

    lowercase__    :						str       =			self.get_inputs(UpperCAmelCase__						)
    lowercase__    :						Optional[int]       =			sd_pipe(**UpperCAmelCase__						).images
    lowercase__    :						Dict       =			image[0, -3:, -3:, -1].flatten()

    assert image.shape == (1, 512, 512, 3)
    lowercase__    :						List[str]       =			np.array([0.84_491, 0.90_789, 0.75_708, 0.78_734, 0.83_485, 0.70_099, 0.66_938, 0.68_727, 0.61_379]						)
    assert np.abs(image_slice - expected_slice						).max() < 6e-3



 def     UpperCAmelCase     (						self : Any						):

    """simple docstring"""






    lowercase__    :						str       =			0

    def callback_fn(UpperCAmelCase__ : int			,		UpperCAmelCase__ : int			,		UpperCAmelCase__ : torch.FloatTensor						) -> None:
       lowercase__    :						Tuple       =			True
       nonlocal number_of_steps
       number_of_steps += 1
       if step == 1:
          lowercase__    :						Optional[Any]       =			latents.detach().cpu().numpy()
          assert latents.shape == (1, 4, 64, 64)
          lowercase__    :						List[Any]       =			latents[0, -3:, -3:, -1]
          lowercase__    :						int       =			np.array(
              [-0.1_621, 0.2_837, -0.7_979, -0.1_221, -1.3_057, 0.7_681, -2.1_191, 0.0_464, 1.6_309]						)

          assert np.abs(latents_slice.flatten() - expected_slice						).max() < 5e-2
       elif step == 2:
          lowercase__    :						Any       =			latents.detach().cpu().numpy()
          assert latents.shape == (1, 4, 64, 64)
          lowercase__    :						List[str]       =			latents[0, -3:, -3:, -1]
          lowercase__    :						Optional[int]       =			np.array([0.6_299, 1.7_500, 1.1_992, -2.1_582, -1.8_994, 0.7_334, -0.7_090, 1.0_137, 1.5_273]						)

          assert np.abs(latents_slice.flatten() - expected_slice						).max() < 5e-2

    lowercase__    :						List[str]       =			False

    lowercase__    :						int       =			StableDiffusionImageVariationPipeline.from_pretrained(
        """fusing/sd-image-variations-diffusers"""			,		safety_checker=UpperCAmelCase__			,		torch_dtype=torch.floataa			,		)
    pipe.to(UpperCAmelCase__						)
    pipe.set_progress_bar_config(disable=UpperCAmelCase__						)
    pipe.enable_attention_slicing()

    lowercase__    :						str       =			self.get_inputs(UpperCAmelCase__			,		dtype=torch.floataa						)
    pipe(**UpperCAmelCase__			,		callback=UpperCAmelCase__			,		callback_steps=1						)
    assert callback_fn.has_been_called
    assert number_of_steps == inputs["num_inference_steps"]







 def     UpperCAmelCase     (						self : Optional[int]						):

    """simple docstring"""






    torch.cuda.empty_cache()
    torch.cuda.reset_max_memory_allocated()
    torch.cuda.reset_peak_memory_stats()

    lowercase__    :						int       =			"""fusing/sd-image-variations-diffusers"""
    lowercase__    :						Union[str, Any]       =			StableDiffusionImageVariationPipeline.from_pretrained(
        UpperCAmelCase__			,		safety_checker=UpperCAmelCase__			,		torch_dtype=torch.floataa						)
    lowercase__    :						Optional[Any]       =			pipe.to(UpperCAmelCase__						)
    pipe.set_progress_bar_config(disable=UpperCAmelCase__						)
    pipe.enable_attention_slicing(1						)
    pipe.enable_sequential_cpu_offload()

    lowercase__    :						Any       =			self.get_inputs(UpperCAmelCase__			,		dtype=torch.floataa						)
    lowercase__    :						Union[str, Any]       =			pipe(**UpperCAmelCase__						)

    lowercase__    :						Dict       =			torch.cuda.max_memory_allocated()
    # make sure that less than 2.6 GB is allocated
    assert mem_bytes < 2.6 * 10**9
@nightly
@require_torch_gpu
class 							__snake_case(       unittest.TestCase   ):



 def     UpperCAmelCase     (						self : str						):

    """simple docstring"""






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



 def     UpperCAmelCase     (						self : int			,		UpperCAmelCase__ : List[Any]			,		UpperCAmelCase__ : Optional[int]="cpu"			,		UpperCAmelCase__ : Tuple=torch.floataa			,		UpperCAmelCase__ : Any=0						):

    """simple docstring"""






    lowercase__    :						Optional[Any]       =			torch.Generator(device=UpperCAmelCase__						).manual_seed(UpperCAmelCase__						)
    lowercase__    :						str       =			load_image(
        """https://huggingface.co/datasets/diffusers/test-arrays/resolve/main"""
        """/stable_diffusion_imgvar/input_image_vermeer.png"""						)
    lowercase__    :						List[Any]       =			np.random.RandomState(UpperCAmelCase__						).standard_normal((1, 4, 64, 64)						)
    lowercase__    :						Any       =			torch.from_numpy(UpperCAmelCase__						).to(device=UpperCAmelCase__			,		dtype=UpperCAmelCase__						)
    lowercase__    :						Optional[Any]       =			{
        """image""": init_image,
        """latents""": latents,
        """generator""": generator,
        """num_inference_steps""": 50,
        """guidance_scale""": 7.5,
        """output_type""": """numpy""",
    }
    return inputs



 def     UpperCAmelCase     (						self : List[str]						):

    """simple docstring"""






    lowercase__    :						str       =			StableDiffusionImageVariationPipeline.from_pretrained("""fusing/sd-image-variations-diffusers"""						)
    sd_pipe.to(UpperCAmelCase__						)
    sd_pipe.set_progress_bar_config(disable=UpperCAmelCase__						)

    lowercase__    :						Union[str, Any]       =			self.get_inputs(UpperCAmelCase__						)
    lowercase__    :						Optional[int]       =			sd_pipe(**UpperCAmelCase__						).images[0]

    lowercase__    :						Optional[Any]       =			load_numpy(
        """https://huggingface.co/datasets/diffusers/test-arrays/resolve/main"""
        """/stable_diffusion_imgvar/lambdalabs_variations_pndm.npy"""						)
    lowercase__    :						Any       =			np.abs(expected_image - image						).max()
    assert max_diff < 1e-3







 def     UpperCAmelCase     (						self : Any						):

    """simple docstring"""






    lowercase__    :						List[str]       =			StableDiffusionImageVariationPipeline.from_pretrained("""fusing/sd-image-variations-diffusers"""						)
    lowercase__    :						List[str]       =			DPMSolverMultistepScheduler.from_config(sd_pipe.scheduler.config						)
    sd_pipe.to(UpperCAmelCase__						)
    sd_pipe.set_progress_bar_config(disable=UpperCAmelCase__						)

    lowercase__    :						Union[str, Any]       =			self.get_inputs(UpperCAmelCase__						)
    lowercase__    :						Optional[int]       =			25
    lowercase__    :						Optional[int]       =			sd_pipe(**UpperCAmelCase__						).images[0]

    lowercase__    :						Any       =			load_numpy(
        """https://huggingface.co/datasets/diffusers/test-arrays/resolve/main"""
        """/stable_diffusion_imgvar/lambdalabs_variations_dpm_multi.npy"""						)
    lowercase__    :						Any       =			np.abs(expected_image - image						).max()
    assert max_diff < 1e-3










from datetime import date

import requests

a 			=							'https://api.carbonintensity.org.uk/intensity'
def    a__    (	):
   lowercase__    :						Union[str, Any]       =			requests.get(A       ).json()["""data"""][0]
   return last_half_hour["intensity"]["actual"]
def    a__    (	A		,					A       ):
   return requests.get(F'''{BASE_URL}/{start}/{end}'''       ).json()["data"]


if __name__ == "__main__":
 for entry in fetch_from_to(start=date(2020, 10, 1), end=date(2020, 10, 3)):
  print('from {from} to {to}: {intensity[actual]}'.format(**entry))
 print(F'{fetch_last_half_hour() = }')


