



from typing import Callable, List, Optional, Union

import torch

from ...models import UNetaDModel
from ...schedulers import CMStochasticIterativeScheduler
from ...utils import (
    is_accelerate_available,
    is_accelerate_version,
    logging,
    randn_tensor,
    replace_example_docstring,
)
from ..pipeline_utils import DiffusionPipeline, ImagePipelineOutput


A__						:						str          =      logging.get_logger(__name__)  # pylint: disable=invalid-name


A__						:						List[str]          =      '''
    Examples:
        ```py
        >>> import torch

        >>> from diffusers import ConsistencyModelPipeline

        >>> device = "cuda"
        >>> # Load the cd_imagenet64_l2 checkpoint.
        >>> model_id_or_path = "openai/diffusers-cd_imagenet64_l2"
        >>> pipe = ConsistencyModelPipeline.from_pretrained(model_id_or_path, torch_dtype=torch.float16)
        >>> pipe.to(device)

        >>> # Onestep Sampling
        >>> image = pipe(num_inference_steps=1).images[0]
        >>> image.save("cd_imagenet64_l2_onestep_sample.png")

        >>> # Onestep sampling, class-conditional image generation
        >>> # ImageNet-64 class label 145 corresponds to king penguins
        >>> image = pipe(num_inference_steps=1, class_labels=145).images[0]
        >>> image.save("cd_imagenet64_l2_onestep_sample_penguin.png")

        >>> # Multistep sampling, class-conditional image generation
        >>> # Timesteps can be explicitly specified; the particular timesteps below are from the original Github repo:
        >>> # https://github.com/openai/consistency_models/blob/main/scripts/launch.sh#L77
        >>> image = pipe(num_inference_steps=None, timesteps=[22, 0], class_labels=145).images[0]
        >>> image.save("cd_imagenet64_l2_multistep_sample_penguin.png")
        ```
'''



class        __snake_case		(  UpperCamelCase_		):




							def __init__( self  : str       ,       A_  : UNetaDModel       ,       A_  : CMStochasticIterativeScheduler):
									super().__init__()

									self.register_modules(
									    unet=A_       ,       scheduler=A_       ,       )

									lowerCAmelCase_    :  int             =				None




							def      UpperCAmelCase__    ( self  : Optional[int]       ,       A_  : Optional[Any]=0):
									if is_accelerate_available() and is_accelerate_version('''>='''       ,       '''0.14.0'''):
											from accelerate import cpu_offload
									else:
											raise ImportError('''`enable_sequential_cpu_offload` requires `accelerate v0.14.0` or higher''')

									lowerCAmelCase_    :  Dict             =				torch.device(F"""cuda:{gpu_id}""")

									if self.device.type != "cpu":
											self.to('''cpu'''       ,       silence_dtype_warnings=A_)
											torch.cuda.empty_cache()  # otherwise we don't see the memory savings (but they probably exist)

									for cpu_offloaded_model in [self.unet]:
											cpu_offload(A_       ,       A_)

									if self.safety_checker is not None:
											cpu_offload(self.safety_checker       ,       execution_device=A_       ,       offload_buffers=A_)




							def      UpperCAmelCase__    ( self  : List[str]       ,       A_  : int=0):
									if is_accelerate_available() and is_accelerate_version('''>='''       ,       '''0.17.0.dev0'''):
											from accelerate import cpu_offload_with_hook
									else:
											raise ImportError('''`enable_model_cpu_offload` requires `accelerate v0.17.0` or higher.''')

									lowerCAmelCase_    :  Optional[int]             =				torch.device(F"""cuda:{gpu_id}""")

									if self.device.type != "cpu":
											self.to('''cpu'''       ,       silence_dtype_warnings=A_)
											torch.cuda.empty_cache()  # otherwise we don't see the memory savings (but they probably exist)

									lowerCAmelCase_    :  Optional[Any]             =				None
									for cpu_offloaded_model in [self.unet]:
											lowerCAmelCase_      ,						lowerCAmelCase_    :  Union[str, Any]             =				cpu_offload_with_hook(A_       ,       A_       ,       prev_module_hook=A_)

									if self.safety_checker is not None:
											lowerCAmelCase_      ,						lowerCAmelCase_    :  List[Any]             =				cpu_offload_with_hook(self.safety_checker       ,       A_       ,       prev_module_hook=A_)

									# We'll offload the last model manually.
									lowerCAmelCase_    :  Union[str, Any]             =				hook




							@property
							# Copied from diffusers.pipelines.stable_diffusion.pipeline_stable_diffusion.StableDiffusionPipeline._execution_device
							def      UpperCAmelCase__    ( self  : Any):
									if not hasattr(self.unet       ,       '''_hf_hook'''):
											return self.device
									for module in self.unet.modules():
											if (
											    hasattr(A_       ,       '''_hf_hook''')
											    and hasattr(module._hf_hook       ,       '''execution_device''')
											    and module._hf_hook.execution_device is not None
											):
													return torch.device(module._hf_hook.execution_device)
									return self.device




							def      UpperCAmelCase__    ( self  : Union[str, Any]       ,       A_  : int       ,       A_  : int       ,       A_  : List[str]       ,       A_  : int       ,       A_  : List[str]       ,       A_  : Dict       ,       A_  : Union[str, Any]       ,       A_  : Dict=None):
									lowerCAmelCase_    :  str             =				(batch_size, num_channels, height, width)
									if isinstance(A_       ,       A_) and len(A_) != batch_size:
											raise ValueError(
											    F"""You have passed a list of generators of length {len(A_)}, but requested an effective batch"""
											    F""" size of {batch_size}. Make sure the batch size matches the length of the generators.""")

									if latents is None:
											lowerCAmelCase_    :  Optional[int]             =				randn_tensor(A_       ,       generator=A_       ,       device=A_       ,       dtype=A_)
									else:
											lowerCAmelCase_    :  Tuple             =				latents.to(device=A_       ,       dtype=A_)

									# scale the initial noise by the standard deviation required by the scheduler
									lowerCAmelCase_    :  List[Any]             =				latents * self.scheduler.init_noise_sigma
									return latents




							def      UpperCAmelCase__    ( self  : List[Any]       ,       A_  : torch.FloatTensor       ,       A_  : str = "pil"):
									if output_type not in ["pt", "np", "pil"]:
											raise ValueError(
											    F"""output_type={output_type} is not supported. Make sure to choose one of ['pt', 'np', or 'pil']""")

									# Equivalent to diffusers.VaeImageProcessor.denormalize
									lowerCAmelCase_    :  List[Any]             =				(sample / 2 + 0.5).clamp(0       ,       1)
									if output_type == "pt":
											return sample

									# Equivalent to diffusers.VaeImageProcessor.pt_to_numpy
									lowerCAmelCase_    :  str             =				sample.cpu().permute(0       ,       2       ,       3       ,       1).numpy()
									if output_type == "np":
											return sample

									# Output_type must be 'pil'
									lowerCAmelCase_    :  int             =				self.numpy_to_pil(A_)
									return sample




							def      UpperCAmelCase__    ( self  : str       ,       A_  : Optional[int]       ,       A_  : Union[str, Any]       ,       A_  : Optional[int]=None):
									if self.unet.config.num_class_embeds is not None:
											if isinstance(A_       ,       A_):
													lowerCAmelCase_    :  str             =				torch.tensor(A_       ,       dtype=torch.int)
											elif isinstance(A_       ,       A_):
													assert batch_size == 1, "Batch size must be 1 if classes is an int"
													lowerCAmelCase_    :  str             =				torch.tensor([class_labels]       ,       dtype=torch.int)
											elif class_labels is None:
													# Randomly generate batch_size class labels
													# TODO: should use generator here? int analogue of randn_tensor is not exposed in ...utils
													lowerCAmelCase_    :  Tuple             =				torch.randint(0       ,       self.unet.config.num_class_embeds       ,       size=(batch_size,))
											lowerCAmelCase_    :  Union[str, Any]             =				class_labels.to(A_)
									else:
											lowerCAmelCase_    :  Dict             =				None
									return class_labels




							def      UpperCAmelCase__    ( self  : Any       ,       A_  : List[Any]       ,       A_  : Optional[Any]       ,       A_  : Optional[Any]       ,       A_  : Optional[int]       ,       A_  : str       ,       A_  : Dict):
									if num_inference_steps is None and timesteps is None:
											raise ValueError('''Exactly one of `num_inference_steps` or `timesteps` must be supplied.''')

									if num_inference_steps is not None and timesteps is not None:
											logger.warning(
											    F"""Both `num_inference_steps`: {num_inference_steps} and `timesteps`: {timesteps} are supplied;"""
											    ''' `timesteps` will be used over `num_inference_steps`.''')

									if latents is not None:
											lowerCAmelCase_    :  Optional[int]             =				(batch_size, 3, img_size, img_size)
											if latents.shape != expected_shape:
													raise ValueError(F"""The shape of latents is {latents.shape} but is expected to be {expected_shape}.""")

									if (callback_steps is None) or (
									    callback_steps is not None and (not isinstance(A_       ,       A_) or callback_steps <= 0)
									):
											raise ValueError(
											    F"""`callback_steps` has to be a positive integer but is {callback_steps} of type"""
											    F""" {type(A_)}.""")







							@torch.no_grad()
							@replace_example_docstring(A_)
							def __call__( self  : Union[str, Any]       ,       A_  : int = 1       ,       A_  : Optional[Union[torch.Tensor, List[int], int]] = None       ,       A_  : int = 1       ,       A_  : List[int] = None       ,       A_  : Optional[Union[torch.Generator, List[torch.Generator]]] = None       ,       A_  : Optional[torch.FloatTensor] = None       ,       A_  : Optional[str] = "pil"       ,       A_  : bool = True       ,       A_  : Optional[Callable[[int, int, torch.FloatTensor], None]] = None       ,       A_  : int = 1       ,       ):
									lowerCAmelCase_    :  Any             =				self.unet.config.sample_size
									lowerCAmelCase_    :  List[str]             =				self._execution_device

									# 1. Check inputs
									self.check_inputs(A_       ,       A_       ,       A_       ,       A_       ,       A_       ,       A_)

									# 2. Prepare image latents
									# Sample image latents x_0 ~ N(0, sigma_0^2 * I)
									lowerCAmelCase_    :  List[Any]             =				self.prepare_latents(
									    batch_size=A_       ,       num_channels=self.unet.config.in_channels       ,       height=A_       ,       width=A_       ,       dtype=self.unet.dtype       ,       device=A_       ,       generator=A_       ,       latents=A_       ,       )

									# 3. Handle class_labels for class-conditional models
									lowerCAmelCase_    :  Optional[Any]             =				self.prepare_class_labels(A_       ,       A_       ,       class_labels=A_)

									# 4. Prepare timesteps
									if timesteps is not None:
											self.scheduler.set_timesteps(timesteps=A_       ,       device=A_)
											lowerCAmelCase_    :  int             =				self.scheduler.timesteps
											lowerCAmelCase_    :  List[Any]             =				len(A_)
									else:
											self.scheduler.set_timesteps(A_)
											lowerCAmelCase_    :  int             =				self.scheduler.timesteps

									# 5. Denoising loop
									# Multistep sampling: implements Algorithm 1 in the paper
									with self.progress_bar(total=A_) as progress_bar:
											for i, t in enumerate(A_):
													lowerCAmelCase_    :  int             =				self.scheduler.scale_model_input(A_       ,       A_)
													lowerCAmelCase_    :  Any             =				self.unet(A_       ,       A_       ,       class_labels=A_       ,       return_dict=A_)[0]

													lowerCAmelCase_    :  int             =				self.scheduler.step(A_       ,       A_       ,       A_       ,       generator=A_)[0]

													# call the callback, if provided
													progress_bar.update()
													if callback is not None and i % callback_steps == 0:
															callback(A_       ,       A_       ,       A_)

        # 6. Post-process image sample
									lowerCAmelCase_    :  List[Any]             =				self.postprocess_image(A_       ,       output_type=A_)

									# Offload last model to CPU
									if hasattr(self       ,       '''final_offload_hook''') and self.final_offload_hook is not None:
											self.final_offload_hook.offload()

									if not return_dict:
											return (image,)

									return ImagePipelineOutput(images=A_)










import unittest

import numpy as np

from transformers.testing_utils import require_torch, require_vision
from transformers.utils import is_torch_available, is_vision_available

from ...test_image_processing_common import ImageProcessingSavingTestMixin, prepare_image_inputs


if is_torch_available():
						import torch

if is_vision_available():
						from PIL import Image

						from transformers import DeiTImageProcessor



class        __snake_case		(  unittest.TestCase		):




							def __init__( self  : Union[str, Any]       ,       A_  : Optional[Any]       ,       A_  : Optional[Any]=7       ,       A_  : str=3       ,       A_  : Optional[Any]=1_8       ,       A_  : Union[str, Any]=3_0       ,       A_  : str=4_0_0       ,       A_  : List[str]=True       ,       A_  : str=None       ,       A_  : str=True       ,       A_  : List[Any]=None       ,       A_  : List[Any]=True       ,       A_  : Optional[Any]=[0.5, 0.5, 0.5]       ,       A_  : Optional[int]=[0.5, 0.5, 0.5]       ,       ):
									lowerCAmelCase_    :  List[str]             =				size if size is not None else {'''height''': 2_0, '''width''': 2_0}
									lowerCAmelCase_    :  List[Any]             =				crop_size if crop_size is not None else {'''height''': 1_8, '''width''': 1_8}

									lowerCAmelCase_    :  str             =				parent
									lowerCAmelCase_    :  Union[str, Any]             =				batch_size
									lowerCAmelCase_    :  Optional[Any]             =				num_channels
									lowerCAmelCase_    :  List[str]             =				image_size
									lowerCAmelCase_    :  List[str]             =				min_resolution
									lowerCAmelCase_    :  Union[str, Any]             =				max_resolution
									lowerCAmelCase_    :  Optional[int]             =				do_resize
									lowerCAmelCase_    :  Dict             =				size
									lowerCAmelCase_    :  Dict             =				do_center_crop
									lowerCAmelCase_    :  str             =				crop_size
									lowerCAmelCase_    :  List[Any]             =				do_normalize
									lowerCAmelCase_    :  str             =				image_mean
									lowerCAmelCase_    :  Dict             =				image_std







							def      UpperCAmelCase__    ( self  : int):
									return {
									    "do_resize": self.do_resize,
									    "size": self.size,
									    "do_center_crop": self.do_center_crop,
									    "crop_size": self.crop_size,
									    "do_normalize": self.do_normalize,
									    "image_mean": self.image_mean,
									    "image_std": self.image_std,
									}





@require_torch
@require_vision
class        __snake_case		(  UpperCamelCase_		,unittest.TestCase		):
							_a       =				DeiTImageProcessor if is_vision_available() else None
							_a       =				True




							def      UpperCAmelCase__    ( self  : Tuple):
									lowerCAmelCase_    :  int             =				DeiTImageProcessingTester(self)




							@property
							def      UpperCAmelCase__    ( self  : Optional[Any]):
									return self.image_processor_tester.prepare_image_processor_dict()




							def      UpperCAmelCase__    ( self  : str):
									lowerCAmelCase_    :  List[Any]             =				self.image_processing_class(**self.image_processor_dict)
									self.assertTrue(hasattr(A_       ,       '''do_resize'''))
									self.assertTrue(hasattr(A_       ,       '''size'''))
									self.assertTrue(hasattr(A_       ,       '''do_center_crop'''))
									self.assertTrue(hasattr(A_       ,       '''center_crop'''))
									self.assertTrue(hasattr(A_       ,       '''do_normalize'''))
									self.assertTrue(hasattr(A_       ,       '''image_mean'''))
									self.assertTrue(hasattr(A_       ,       '''image_std'''))




							def      UpperCAmelCase__    ( self  : Union[str, Any]):
									lowerCAmelCase_    :  Tuple             =				self.image_processing_class.from_dict(self.image_processor_dict)
									self.assertEqual(image_processor.size       ,       {'''height''': 2_0, '''width''': 2_0})
									self.assertEqual(image_processor.crop_size       ,       {'''height''': 1_8, '''width''': 1_8})

									lowerCAmelCase_    :  Any             =				self.image_processing_class.from_dict(self.image_processor_dict       ,       size=4_2       ,       crop_size=8_4)
									self.assertEqual(image_processor.size       ,       {'''height''': 4_2, '''width''': 4_2})
									self.assertEqual(image_processor.crop_size       ,       {'''height''': 8_4, '''width''': 8_4})




							def      UpperCAmelCase__    ( self  : Union[str, Any]):
									pass




							def      UpperCAmelCase__    ( self  : Any):
									# Initialize image_processing
									lowerCAmelCase_    :  List[str]             =				self.image_processing_class(**self.image_processor_dict)
									# create random PIL images
									lowerCAmelCase_    :  Any             =				prepare_image_inputs(self.image_processor_tester       ,       equal_resolution=A_)
									for image in image_inputs:
											self.assertIsInstance(A_       ,       Image.Image)

									# Test not batched input
									lowerCAmelCase_    :  Union[str, Any]             =				image_processing(image_inputs[0]       ,       return_tensors='''pt''').pixel_values
									self.assertEqual(
									    encoded_images.shape       ,       (
									        1,
									        self.image_processor_tester.num_channels,
									        self.image_processor_tester.crop_size['''height'''],
									        self.image_processor_tester.crop_size['''width'''],
									    )       ,       )

									# Test batched
									lowerCAmelCase_    :  Dict             =				image_processing(A_       ,       return_tensors='''pt''').pixel_values
									self.assertEqual(
									    encoded_images.shape       ,       (
									        self.image_processor_tester.batch_size,
									        self.image_processor_tester.num_channels,
									        self.image_processor_tester.crop_size['''height'''],
									        self.image_processor_tester.crop_size['''width'''],
									    )       ,       )




							def      UpperCAmelCase__    ( self  : List[str]):
									# Initialize image_processing
									lowerCAmelCase_    :  Union[str, Any]             =				self.image_processing_class(**self.image_processor_dict)
									# create random numpy tensors
									lowerCAmelCase_    :  Union[str, Any]             =				prepare_image_inputs(self.image_processor_tester       ,       equal_resolution=A_       ,       numpify=A_)
									for image in image_inputs:
											self.assertIsInstance(A_       ,       np.ndarray)

									# Test not batched input
									lowerCAmelCase_    :  Optional[Any]             =				image_processing(image_inputs[0]       ,       return_tensors='''pt''').pixel_values
									self.assertEqual(
									    encoded_images.shape       ,       (
									        1,
									        self.image_processor_tester.num_channels,
									        self.image_processor_tester.crop_size['''height'''],
									        self.image_processor_tester.crop_size['''width'''],
									    )       ,       )

									# Test batched
									lowerCAmelCase_    :  str             =				image_processing(A_       ,       return_tensors='''pt''').pixel_values
									self.assertEqual(
									    encoded_images.shape       ,       (
									        self.image_processor_tester.batch_size,
									        self.image_processor_tester.num_channels,
									        self.image_processor_tester.crop_size['''height'''],
									        self.image_processor_tester.crop_size['''width'''],
									    )       ,       )







							def      UpperCAmelCase__    ( self  : Any):
									# Initialize image_processing
									lowerCAmelCase_    :  str             =				self.image_processing_class(**self.image_processor_dict)
									# create random PyTorch tensors
									lowerCAmelCase_    :  Union[str, Any]             =				prepare_image_inputs(self.image_processor_tester       ,       equal_resolution=A_       ,       torchify=A_)
									for image in image_inputs:
											self.assertIsInstance(A_       ,       torch.Tensor)

									# Test not batched input
									lowerCAmelCase_    :  Dict             =				image_processing(image_inputs[0]       ,       return_tensors='''pt''').pixel_values
									self.assertEqual(
									    encoded_images.shape       ,       (
									        1,
									        self.image_processor_tester.num_channels,
									        self.image_processor_tester.crop_size['''height'''],
									        self.image_processor_tester.crop_size['''width'''],
									    )       ,       )

									# Test batched
									lowerCAmelCase_    :  Tuple             =				image_processing(A_       ,       return_tensors='''pt''').pixel_values
									self.assertEqual(
									    encoded_images.shape       ,       (
									        self.image_processor_tester.batch_size,
									        self.image_processor_tester.num_channels,
									        self.image_processor_tester.crop_size['''height'''],
									        self.image_processor_tester.crop_size['''width'''],
									    )       ,       )





