





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


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


_UpperCamelCase    =	'''
    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 			_lowerCamelCase	(      a      ):



      """simple docstring"""






      def __init__(	self			,  UpperCAmelCase			,  UpperCAmelCase				)    ->				None:






        '''simple docstring'''






        super().__init__()

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

        __snake_case				:			Optional[int]            =    None







      def   UpperCAmelCase						(	self			,  UpperCAmelCase=0				)    ->				Tuple:






        '''simple docstring'''






        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"				)

        __snake_case				:			List[str]            =    torch.device(F"""cuda:{gpu_id}"""				)

        if self.device.type != "cpu":
          self.to("cpu"			,  silence_dtype_warnings=UpperCAmelCase				)
          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(UpperCAmelCase			,  UpperCAmelCase				)

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







      def   UpperCAmelCase						(	self			,  UpperCAmelCase=0				)    ->				Any:






        '''simple docstring'''






        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."				)

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

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

        __snake_case				:			Dict            =    None
        for cpu_offloaded_model in [self.unet]:
          __snake_case      ,				__snake_case				:			Any            =    cpu_offload_with_hook(UpperCAmelCase			,  UpperCAmelCase			,  prev_module_hook=UpperCAmelCase				)

        if self.safety_checker is not None:
          __snake_case      ,				__snake_case				:			Union[str, Any]            =    cpu_offload_with_hook(self.safety_checker			,  UpperCAmelCase			,  prev_module_hook=UpperCAmelCase				)

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







      @property
      # Copied from diffusers.pipelines.stable_diffusion.pipeline_stable_diffusion.StableDiffusionPipeline._execution_device
      def   UpperCAmelCase						(	self				)    ->				Dict:






        '''simple docstring'''






        if not hasattr(self.unet			,  "_hf_hook"				):
          return self.device
        for module in self.unet.modules():
          if (
              hasattr(UpperCAmelCase			,  "_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			,  UpperCAmelCase			,  UpperCAmelCase			,  UpperCAmelCase			,  UpperCAmelCase			,  UpperCAmelCase			,  UpperCAmelCase			,  UpperCAmelCase			,  UpperCAmelCase=None				)    ->				Tuple:






        '''simple docstring'''






        __snake_case				:			Dict            =    (batch_size, num_channels, height, width)
        if isinstance(UpperCAmelCase			,  UpperCAmelCase				) and len(UpperCAmelCase				) != batch_size:
          raise ValueError(
              F"""You have passed a list of generators of length {len(UpperCAmelCase				)}, 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:
          __snake_case				:			Tuple            =    randn_tensor(UpperCAmelCase			,  generator=UpperCAmelCase			,  device=UpperCAmelCase			,  dtype=UpperCAmelCase				)
        else:
          __snake_case				:			Optional[Any]            =    latents.to(device=UpperCAmelCase			,  dtype=UpperCAmelCase				)

        # scale the initial noise by the standard deviation required by the scheduler
        __snake_case				:			str            =    latents * self.scheduler.init_noise_sigma
        return latents







      def   UpperCAmelCase						(	self			,  UpperCAmelCase			,  UpperCAmelCase = "pil"				)    ->				List[str]:






        '''simple docstring'''






        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
        __snake_case				:			Optional[Any]            =    (sample / 2 + 0.5).clamp(0			,  1				)
        if output_type == "pt":
          return sample

        # Equivalent to diffusers.VaeImageProcessor.pt_to_numpy
        __snake_case				:			List[Any]            =    sample.cpu().permute(0			,  2			,  3			,  1				).numpy()
        if output_type == "np":
          return sample

        # Output_type must be 'pil'
        __snake_case				:			Optional[int]            =    self.numpy_to_pil(UpperCAmelCase				)
        return sample







      def   UpperCAmelCase						(	self			,  UpperCAmelCase			,  UpperCAmelCase			,  UpperCAmelCase=None				)    ->				Any:






        '''simple docstring'''






        if self.unet.config.num_class_embeds is not None:
          if isinstance(UpperCAmelCase			,  UpperCAmelCase				):
            __snake_case				:			List[Any]            =    torch.tensor(UpperCAmelCase			,  dtype=torch.int				)
          elif isinstance(UpperCAmelCase			,  UpperCAmelCase				):
            assert batch_size == 1, "Batch size must be 1 if classes is an int"
            __snake_case				:			List[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
            __snake_case				:			str            =    torch.randint(0			,  self.unet.config.num_class_embeds			,  size=(batch_size,)				)
          __snake_case				:			Optional[Any]            =    class_labels.to(UpperCAmelCase				)
        else:
          __snake_case				:			List[Any]            =    None
        return class_labels







      def   UpperCAmelCase						(	self			,  UpperCAmelCase			,  UpperCAmelCase			,  UpperCAmelCase			,  UpperCAmelCase			,  UpperCAmelCase			,  UpperCAmelCase				)    ->				Any:






        '''simple docstring'''






        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:
          __snake_case				:			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(UpperCAmelCase			,  UpperCAmelCase				) or callback_steps <= 0)
        ):
          raise ValueError(
              F"""`callback_steps` has to be a positive integer but is {callback_steps} of type"""
              F""" {type(UpperCAmelCase				)}."""				)







      @torch.no_grad()
      @replace_example_docstring(UpperCAmelCase				)
      def __call__(	self			,  UpperCAmelCase = 1			,  UpperCAmelCase = None			,  UpperCAmelCase = 1			,  UpperCAmelCase = None			,  UpperCAmelCase = None			,  UpperCAmelCase = None			,  UpperCAmelCase = "pil"			,  UpperCAmelCase = True			,  UpperCAmelCase = None			,  UpperCAmelCase = 1			,  )    ->				Optional[int]:






        '''simple docstring'''






        __snake_case				:			Union[str, Any]            =    self.unet.config.sample_size
        __snake_case				:			str            =    self._execution_device

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

        # 2. Prepare image latents
        # Sample image latents x_0 ~ N(0, sigma_0^2 * I)
        __snake_case				:			str            =    self.prepare_latents(
            batch_size=UpperCAmelCase			,  num_channels=self.unet.config.in_channels			,  height=UpperCAmelCase			,  width=UpperCAmelCase			,  dtype=self.unet.dtype			,  device=UpperCAmelCase			,  generator=UpperCAmelCase			,  latents=UpperCAmelCase			,  )

        # 3. Handle class_labels for class-conditional models
        __snake_case				:			str            =    self.prepare_class_labels(UpperCAmelCase			,  UpperCAmelCase			,  class_labels=UpperCAmelCase				)

        # 4. Prepare timesteps
        if timesteps is not None:
          self.scheduler.set_timesteps(timesteps=UpperCAmelCase			,  device=UpperCAmelCase				)
          __snake_case				:			int            =    self.scheduler.timesteps
          __snake_case				:			Union[str, Any]            =    len(UpperCAmelCase				)
        else:
          self.scheduler.set_timesteps(UpperCAmelCase				)
          __snake_case				:			str            =    self.scheduler.timesteps

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

            __snake_case				:			Dict            =    self.scheduler.step(UpperCAmelCase			,  UpperCAmelCase			,  UpperCAmelCase			,  generator=UpperCAmelCase				)[0]

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

        # 6. Post-process image sample
        __snake_case				:			str            =    self.postprocess_image(UpperCAmelCase			,  output_type=UpperCAmelCase				)

        # 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=UpperCAmelCase				)








import numpy as np
from cva import destroyAllWindows, imread, imshow, waitKey




class 			_lowerCamelCase	:



      """simple docstring"""






      def __init__(	self			,  UpperCAmelCase			,  UpperCAmelCase				)    ->				str:






        '''simple docstring'''






        __snake_case				:			int            =    0
        # max greyscale value for #FFFFFF
        __snake_case				:			int            =    int(self.get_greyscale(255			,  255			,  255				)				)

        if not self.min_threshold < threshold < self.max_threshold:
          __snake_case				:			Tuple            =    F"""Factor value should be from 0 to {self.max_threshold}"""
          raise ValueError(UpperCAmelCase				)

        __snake_case				:			List[Any]            =    input_img
        __snake_case				:			Any            =    threshold
        __snake_case      ,				__snake_case				:			str            =    self.input_img.shape[1], self.input_img.shape[0]

        # error table size (+4 columns and +1 row) greater than input image because of
        # lack of if statements
        __snake_case				:			Dict            =    [
            [0 for _ in range(self.height + 4				)] for __ in range(self.width + 1				)
        ]
        __snake_case				:			str            =    np.ones((self.width, self.height, 3)			,  np.uinta				) * 255







      @classmethod
      def   UpperCAmelCase						(	cls			,  UpperCAmelCase			,  UpperCAmelCase			,  UpperCAmelCase				)    ->				float:






        '''simple docstring'''






        return 0.114 * blue + 0.587 * green + 0.2_126 * red







      def   UpperCAmelCase						(	self				)    ->				None:






        '''simple docstring'''






        for y in range(self.height				):
          for x in range(self.width				):
            __snake_case				:			Optional[int]            =    int(self.get_greyscale(*self.input_img[y][x]				)				)
            if self.threshold > greyscale + self.error_table[y][x]:
              __snake_case				:			Any            =    (0, 0, 0)
              __snake_case				:			Optional[int]            =    greyscale + self.error_table[x][y]
            else:
              __snake_case				:			str            =    (255, 255, 255)
              __snake_case				:			Dict            =    greyscale + self.error_table[x][y] - 255
            self.error_table[y][x + 1] += int(8 / 32 * current_error				)
            self.error_table[y][x + 2] += int(4 / 32 * current_error				)
            self.error_table[y + 1][x] += int(8 / 32 * current_error				)
            self.error_table[y + 1][x + 1] += int(4 / 32 * current_error				)
            self.error_table[y + 1][x + 2] += int(2 / 32 * current_error				)
            self.error_table[y + 1][x - 1] += int(4 / 32 * current_error				)
            self.error_table[y + 1][x - 2] += int(2 / 32 * current_error				)


if __name__ == "__main__":
     # create Burke's instances with original images in greyscale
     _UpperCamelCase    =	[
         Burkes(imread('''image_data/lena.jpg''', 1), threshold)
         for threshold in (1, 126, 130, 140)
     ]

     for burkes in burkes_instances:
          burkes.process()

     for burkes in burkes_instances:
          imshow(
              F'''Original image with dithering threshold: {burkes.threshold}''',
              burkes.output_img,
          )

     waitKey(0)
     destroyAllWindows()

