






import gc
import unittest

import numpy as np
import torch
from torch.backends.cuda import sdp_kernel

from diffusers import (
    CMStochasticIterativeScheduler,
    ConsistencyModelPipeline,
    UNetaDModel,
)
from diffusers.utils import randn_tensor, slow, torch_device
from diffusers.utils.testing_utils import enable_full_determinism, require_torch_a, require_torch_gpu

from ..pipeline_params import UNCONDITIONAL_IMAGE_GENERATION_BATCH_PARAMS, UNCONDITIONAL_IMAGE_GENERATION_PARAMS
from ..test_pipelines_common import PipelineTesterMixin


enable_full_determinism()


class    __lowercase   (						lowercase_      ,   unittest.TestCase	):







     '''simple docstring'''
     SCREAMING_SNAKE_CASE     =					ConsistencyModelPipeline
     SCREAMING_SNAKE_CASE     =					UNCONDITIONAL_IMAGE_GENERATION_PARAMS
     SCREAMING_SNAKE_CASE     =					UNCONDITIONAL_IMAGE_GENERATION_BATCH_PARAMS

     # Override required_optional_params to remove num_images_per_prompt
     SCREAMING_SNAKE_CASE     =					frozenset(
         [
             "num_inference_steps",
             "generator",
             "latents",
             "output_type",
             "return_dict",
             "callback",
             "callback_steps",
         ]	)




     @property
     def 		lowerCAmelCase_  (					self							:      List[str]					):


         """simple docstring"""



         __A       =						UNetaDModel.from_pretrained(
             """diffusers/consistency-models-test"""			,						subfolder="""test_unet"""			,						)
         return unet




     @property
     def 		lowerCAmelCase_  (					self							:      Tuple					):


         """simple docstring"""



         __A       =						UNetaDModel.from_pretrained(
             """diffusers/consistency-models-test"""			,						subfolder="""test_unet_class_cond"""			,						)
         return unet




     def 		lowerCAmelCase_  (					self							:      Optional[Any]			,						UpperCamelCase_							:      Optional[int]=False					):


         """simple docstring"""



         if class_cond:
             __A       =						self.dummy_cond_unet
         else:
             __A       =						self.dummy_uncond_unet

         # Default to CM multistep sampler
         __A       =						CMStochasticIterativeScheduler(
             num_train_timesteps=40			,						sigma_min=0.002			,						sigma_max=80.0			,						)

         __A       =						{
             """unet""": unet,
             """scheduler""": scheduler,
         }

         return components




     def 		lowerCAmelCase_  (					self							:      Any			,						UpperCamelCase_							:      Union[str, Any]			,						UpperCamelCase_							:      Dict=0					):


         """simple docstring"""



         if str(UpperCamelCase_					).startswith("""mps"""					):
             __A       =						torch.manual_seed(UpperCamelCase_					)
         else:
             __A       =						torch.Generator(device=UpperCamelCase_					).manual_seed(UpperCamelCase_					)

         __A       =						{
             """batch_size""": 1,
             """num_inference_steps""": None,
             """timesteps""": [22, 0],
             """generator""": generator,
             """output_type""": """np""",
         }

         return inputs




     def 		lowerCAmelCase_  (					self							:      Tuple					):


         """simple docstring"""



         __A       =						"""cpu"""  # ensure determinism for the device-dependent torch.Generator
         __A       =						self.get_dummy_components()
         __A       =						ConsistencyModelPipeline(**UpperCamelCase_					)
         __A       =						pipe.to(UpperCamelCase_					)
         pipe.set_progress_bar_config(disable=UpperCamelCase_					)

         __A       =						self.get_dummy_inputs(UpperCamelCase_					)
         __A       =						pipe(**UpperCamelCase_					).images
         assert image.shape == (1, 32, 32, 3)

         __A       =						image[0, -3:, -3:, -1]
         __A       =						np.array([0.3572, 0.6273, 0.4031, 0.3961, 0.4321, 0.5730, 0.5266, 0.4780, 0.5004]					)

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




     def 		lowerCAmelCase_  (					self							:      List[str]					):


         """simple docstring"""



         __A       =						"""cpu"""  # ensure determinism for the device-dependent torch.Generator
         __A       =						self.get_dummy_components(class_cond=UpperCamelCase_					)
         __A       =						ConsistencyModelPipeline(**UpperCamelCase_					)
         __A       =						pipe.to(UpperCamelCase_					)
         pipe.set_progress_bar_config(disable=UpperCamelCase_					)

         __A       =						self.get_dummy_inputs(UpperCamelCase_					)
         __A       =						0
         __A       =						pipe(**UpperCamelCase_					).images
         assert image.shape == (1, 32, 32, 3)

         __A       =						image[0, -3:, -3:, -1]
         __A       =						np.array([0.3572, 0.6273, 0.4031, 0.3961, 0.4321, 0.5730, 0.5266, 0.4780, 0.5004]					)

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




     def 		lowerCAmelCase_  (					self							:      int					):


         """simple docstring"""



         __A       =						"""cpu"""  # ensure determinism for the device-dependent torch.Generator
         __A       =						self.get_dummy_components()
         __A       =						ConsistencyModelPipeline(**UpperCamelCase_					)
         __A       =						pipe.to(UpperCamelCase_					)
         pipe.set_progress_bar_config(disable=UpperCamelCase_					)

         __A       =						self.get_dummy_inputs(UpperCamelCase_					)
         __A       =						1
         __A       =						None
         __A       =						pipe(**UpperCamelCase_					).images
         assert image.shape == (1, 32, 32, 3)

         __A       =						image[0, -3:, -3:, -1]
         __A       =						np.array([0.5004, 0.5004, 0.4994, 0.5008, 0.4976, 0.5018, 0.4990, 0.4982, 0.4987]					)

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





     def 		lowerCAmelCase_  (					self							:      str					):


         """simple docstring"""



         __A       =						"""cpu"""  # ensure determinism for the device-dependent torch.Generator
         __A       =						self.get_dummy_components(class_cond=UpperCamelCase_					)
         __A       =						ConsistencyModelPipeline(**UpperCamelCase_					)
         __A       =						pipe.to(UpperCamelCase_					)
         pipe.set_progress_bar_config(disable=UpperCamelCase_					)

         __A       =						self.get_dummy_inputs(UpperCamelCase_					)
         __A       =						1
         __A       =						None
         __A       =						0
         __A       =						pipe(**UpperCamelCase_					).images
         assert image.shape == (1, 32, 32, 3)

         __A       =						image[0, -3:, -3:, -1]
         __A       =						np.array([0.5004, 0.5004, 0.4994, 0.5008, 0.4976, 0.5018, 0.4990, 0.4982, 0.4987]					)

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


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







     '''simple docstring'''




     def 		lowerCAmelCase_  (					self							:      int					):


         """simple docstring"""



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




     def 		lowerCAmelCase_  (					self							:      Dict			,						UpperCamelCase_							:      List[Any]=0			,						UpperCamelCase_							:      Any=False			,						UpperCamelCase_							:      Tuple="cpu"			,						UpperCamelCase_							:      Dict=torch.floataa			,						UpperCamelCase_							:      Dict=(1, 3, 64, 64)					):


         """simple docstring"""



         __A       =						torch.manual_seed(UpperCamelCase_					)

         __A       =						{
             """num_inference_steps""": None,
             """timesteps""": [22, 0],
             """class_labels""": 0,
             """generator""": generator,
             """output_type""": """np""",
         }

         if get_fixed_latents:
             __A       =						self.get_fixed_latents(seed=UpperCamelCase_			,						device=UpperCamelCase_			,						dtype=UpperCamelCase_			,						shape=UpperCamelCase_					)
             __A       =						latents

         return inputs




     def 		lowerCAmelCase_  (					self							:      Tuple			,						UpperCamelCase_							:      List[str]=0			,						UpperCamelCase_							:      Any="cpu"			,						UpperCamelCase_							:      Tuple=torch.floataa			,						UpperCamelCase_							:      str=(1, 3, 64, 64)					):


         """simple docstring"""



         if type(UpperCamelCase_					) == str:
             __A       =						torch.device(UpperCamelCase_					)
         __A       =						torch.Generator(device=UpperCamelCase_					).manual_seed(UpperCamelCase_					)
         __A       =						randn_tensor(UpperCamelCase_			,						generator=UpperCamelCase_			,						device=UpperCamelCase_			,						dtype=UpperCamelCase_					)
         return latents




     def 		lowerCAmelCase_  (					self							:      List[Any]					):


         """simple docstring"""



         __A       =						UNetaDModel.from_pretrained("""diffusers/consistency_models"""			,						subfolder="""diffusers_cd_imagenet64_l2"""					)
         __A       =						CMStochasticIterativeScheduler(
             num_train_timesteps=40			,						sigma_min=0.002			,						sigma_max=80.0			,						)
         __A       =						ConsistencyModelPipeline(unet=UpperCamelCase_			,						scheduler=UpperCamelCase_					)
         pipe.to(torch_device=UpperCamelCase_					)
         pipe.set_progress_bar_config(disable=UpperCamelCase_					)

         __A       =						self.get_inputs()
         __A       =						pipe(**UpperCamelCase_					).images
         assert image.shape == (1, 64, 64, 3)

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

         __A       =						np.array([0.0888, 0.0881, 0.0666, 0.0479, 0.0292, 0.0195, 0.0201, 0.0163, 0.0254]					)

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




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


         """simple docstring"""



         __A       =						UNetaDModel.from_pretrained("""diffusers/consistency_models"""			,						subfolder="""diffusers_cd_imagenet64_l2"""					)
         __A       =						CMStochasticIterativeScheduler(
             num_train_timesteps=40			,						sigma_min=0.002			,						sigma_max=80.0			,						)
         __A       =						ConsistencyModelPipeline(unet=UpperCamelCase_			,						scheduler=UpperCamelCase_					)
         pipe.to(torch_device=UpperCamelCase_					)
         pipe.set_progress_bar_config(disable=UpperCamelCase_					)

         __A       =						self.get_inputs()
         __A       =						1
         __A       =						None
         __A       =						pipe(**UpperCamelCase_					).images
         assert image.shape == (1, 64, 64, 3)

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

         __A       =						np.array([0.0340, 0.0152, 0.0063, 0.0267, 0.0221, 0.0107, 0.0416, 0.0186, 0.0217]					)

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




     @require_torch_a
     def 		lowerCAmelCase_  (					self							:      List[str]					):


         """simple docstring"""



         __A       =						UNetaDModel.from_pretrained("""diffusers/consistency_models"""			,						subfolder="""diffusers_cd_imagenet64_l2"""					)
         __A       =						CMStochasticIterativeScheduler(
             num_train_timesteps=40			,						sigma_min=0.002			,						sigma_max=80.0			,						)
         __A       =						ConsistencyModelPipeline(unet=UpperCamelCase_			,						scheduler=UpperCamelCase_					)
         pipe.to(torch_device=UpperCamelCase_			,						torch_dtype=torch.floataa					)
         pipe.set_progress_bar_config(disable=UpperCamelCase_					)

         __A       =						self.get_inputs(get_fixed_latents=UpperCamelCase_			,						device=UpperCamelCase_					)
         # Ensure usage of flash attention in torch 2.0
         with sdp_kernel(enable_flash=UpperCamelCase_			,						enable_math=UpperCamelCase_			,						enable_mem_efficient=UpperCamelCase_					):
             __A       =						pipe(**UpperCamelCase_					).images
         assert image.shape == (1, 64, 64, 3)

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

         __A       =						np.array([0.1875, 0.1428, 0.1289, 0.2151, 0.2092, 0.1477, 0.1877, 0.1641, 0.1353]					)

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





     @require_torch_a
     def 		lowerCAmelCase_  (					self							:      Any					):


         """simple docstring"""



         __A       =						UNetaDModel.from_pretrained("""diffusers/consistency_models"""			,						subfolder="""diffusers_cd_imagenet64_l2"""					)
         __A       =						CMStochasticIterativeScheduler(
             num_train_timesteps=40			,						sigma_min=0.002			,						sigma_max=80.0			,						)
         __A       =						ConsistencyModelPipeline(unet=UpperCamelCase_			,						scheduler=UpperCamelCase_					)
         pipe.to(torch_device=UpperCamelCase_			,						torch_dtype=torch.floataa					)
         pipe.set_progress_bar_config(disable=UpperCamelCase_					)

         __A       =						self.get_inputs(get_fixed_latents=UpperCamelCase_			,						device=UpperCamelCase_					)
         __A       =						1
         __A       =						None
         # Ensure usage of flash attention in torch 2.0
         with sdp_kernel(enable_flash=UpperCamelCase_			,						enable_math=UpperCamelCase_			,						enable_mem_efficient=UpperCamelCase_					):
             __A       =						pipe(**UpperCamelCase_					).images
         assert image.shape == (1, 64, 64, 3)

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

         __A       =						np.array([0.1663, 0.1948, 0.2275, 0.1680, 0.1204, 0.1245, 0.1858, 0.1338, 0.2095]					)

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












from __future__ import annotations

from PIL import Image

# Define glider example
__a   :							Tuple						= [
    [0, 1, 0, 0, 0, 0, 0, 0],
    [0, 0, 1, 0, 0, 0, 0, 0],
    [1, 1, 1, 0, 0, 0, 0, 0],
    [0, 0, 0, 0, 0, 0, 0, 0],
    [0, 0, 0, 0, 0, 0, 0, 0],
    [0, 0, 0, 0, 0, 0, 0, 0],
    [0, 0, 0, 0, 0, 0, 0, 0],
    [0, 0, 0, 0, 0, 0, 0, 0],
]

# Define blinker example
__a   :							Optional[Any]						= [[0, 1, 0], [0, 1, 0], [0, 1, 0]]



def       _SCREAMING_SNAKE_CASE    (			__lowercase :      list[list[int]]     )    ->   list[list[int]]:





    """simple docstring"""

    __A       =						[]
    for i in range(len(__lowercase     )     ):
        __A       =						[]
        for j in range(len(cells[i]     )     ):
            # Get the number of live neighbours
            __A       =						0
            if i > 0 and j > 0:
                neighbour_count += cells[i - 1][j - 1]
            if i > 0:
                neighbour_count += cells[i - 1][j]
            if i > 0 and j < len(cells[i]     ) - 1:
                neighbour_count += cells[i - 1][j + 1]
            if j > 0:
                neighbour_count += cells[i][j - 1]
            if j < len(cells[i]     ) - 1:
                neighbour_count += cells[i][j + 1]
            if i < len(__lowercase     ) - 1 and j > 0:
                neighbour_count += cells[i + 1][j - 1]
            if i < len(__lowercase     ) - 1:
                neighbour_count += cells[i + 1][j]
            if i < len(__lowercase     ) - 1 and j < len(cells[i]     ) - 1:
                neighbour_count += cells[i + 1][j + 1]

            # Rules of the game of life (excerpt from Wikipedia):
            # 1. Any live cell with two or three live neighbours survives.
            # 2. Any dead cell with three live neighbours becomes a live cell.
            # 3. All other live cells die in the next generation.
            #    Similarly, all other dead cells stay dead.
            __A       =						cells[i][j] == 1
            if (
                (alive and 2 <= neighbour_count <= 3)
                or not alive
                and neighbour_count == 3
            ):
                next_generation_row.append(1     )
            else:
                next_generation_row.append(0     )

        next_generation.append(__lowercase     )
    return next_generation



def       _SCREAMING_SNAKE_CASE    (			__lowercase :      list[list[int]]      ,					__lowercase :      int     )    ->   list[Image.Image]:





    """simple docstring"""

    __A       =						[]
    for _ in range(__lowercase     ):
        # Create output image
        __A       =						Image.new("""RGB"""      ,					(len(cells[0]     ), len(__lowercase     ))     )
        __A       =						img.load()

        # Save cells to image
        for x in range(len(__lowercase     )     ):
            for y in range(len(cells[0]     )     ):
                __A       =						2_5_5 - cells[y][x] * 2_5_5
                __A       =						(colour, colour, colour)

        # Save image
        images.append(__lowercase     )
        __A       =						new_generation(__lowercase     )
    return images


if __name__ == "__main__":
  __a   :							Optional[int]						= generate_images(GLIDER, 16)
  images[0].save("out.gif", save_all=True, append_images=images[1:])




