

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     						=      logging.get_logger(__name__)  # pylint: disable=invalid-name


__A     						=      '''
    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     lowercase_       (    __lowercase	):



      def __init__(     self :							Dict     ,   A__ :							UNetaDModel     ,   A__ :							CMStochasticIterativeScheduler						)       ->	None:
        super().__init__()

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

        _snake_case		=				None



      def 		UpperCamelCase_						(     self :							str     ,   A__ :							List[Any]=0						)       ->	Union[str, Any]:
        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		=				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__ :							List[str]=0						)       ->	Dict:
        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		=				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)

        _snake_case		=				None
        for cpu_offloaded_model in [self.unet]:
          _snake_case,	_snake_case		=				cpu_offload_with_hook(A__     ,   A__     ,   prev_module_hook=A__						)

        if self.safety_checker is not None:
          _snake_case,	_snake_case		=				cpu_offload_with_hook(self.safety_checker     ,   A__     ,   prev_module_hook=A__						)

        # We'll offload the last model manually.
        _snake_case		=				hook



      @property
      # Copied from diffusers.pipelines.stable_diffusion.pipeline_stable_diffusion.StableDiffusionPipeline._execution_device
      def 		UpperCamelCase_						(     self :							int						)       ->	str:
        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 :							List[Any]     ,   A__ :							Any     ,   A__ :							int     ,   A__ :							Optional[Any]     ,   A__ :							str     ,   A__ :							Any     ,   A__ :							Optional[Any]     ,   A__ :							int     ,   A__ :							Optional[Any]=None						)       ->	Optional[int]:
        _snake_case		=				(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:
          _snake_case		=				randn_tensor(A__     ,   generator=A__     ,   device=A__     ,   dtype=A__						)
        else:
          _snake_case		=				latents.to(device=A__     ,   dtype=A__						)

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



      def 		UpperCamelCase_						(     self :							Dict     ,   A__ :							torch.FloatTensor     ,   A__ :							str = "pil"						)       ->	Tuple:
        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		=				(sample / 2 + 0.5).clamp(0     ,   1						)
        if output_type == "pt":
          return sample

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

        # Output_type must be 'pil'
        _snake_case		=				self.numpy_to_pil(A__						)
        return sample



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



      def 		UpperCamelCase_						(     self :							List[Any]     ,   A__ :							Dict     ,   A__ :							str     ,   A__ :							str     ,   A__ :							Optional[int]     ,   A__ :							Any     ,   A__ :							List[Any]						)       ->	Union[str, Any]:
        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		=				(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 :							Tuple     ,   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     ,   )       ->	Optional[Any]:
        _snake_case		=				self.unet.config.sample_size
        _snake_case		=				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)
        _snake_case		=				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
        _snake_case		=				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__						)
          _snake_case		=				self.scheduler.timesteps
          _snake_case		=				len(A__						)
        else:
          self.scheduler.set_timesteps(A__						)
          _snake_case		=				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__						):
            _snake_case		=				self.scheduler.scale_model_input(A__     ,   A__						)
            _snake_case		=				self.unet(A__     ,   A__     ,   class_labels=A__     ,   return_dict=A__						)[0]

            _snake_case		=				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
        _snake_case		=				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 functools
from math import sqrt
from time import time



def  snake_case_(_UpperCamelCase					,			*_UpperCamelCase					,			**_UpperCamelCase	)				->      str:



  """simple docstring"""






  _snake_case		=				time()
  _snake_case		=				func(*_UpperCamelCase					,			**_UpperCamelCase	)
  _snake_case		=				time()
  if int(end - start	) > 0:
    print(F"""{func.__name__} runtime: {(end - start):0.4f} s"""	)
  else:
    print(F"""{func.__name__} runtime: {(end - start) * 1_000:0.4f} ms"""	)
  return output



def  snake_case_(_UpperCamelCase	)				->      list[int]:



  """simple docstring"""






  if n < 0:
    raise Exception('''n is negative'''	)
  if n == 0:
    return [0]
  _snake_case		=				[0, 1]
  for _ in range(n - 1	):
    fib.append(fib[-1] + fib[-2]	)
  return fib



def  snake_case_(_UpperCamelCase	)				->      list[int]:



  """simple docstring"""






  def fib_recursive_term(_UpperCamelCase	) -> int:
    if i < 0:
      raise Exception('''n is negative'''	)
    if i < 2:
      return i
    return fib_recursive_term(i - 1	) + fib_recursive_term(i - 2	)

  if n < 0:
    raise Exception('''n is negative'''	)
  return [fib_recursive_term(_UpperCamelCase	) for i in range(n + 1	)]



def  snake_case_(_UpperCamelCase	)				->      list[int]:



  """simple docstring"""






  @functools.cache
  def fib_recursive_term(_UpperCamelCase	) -> int:
    if i < 0:
      raise Exception('''n is negative'''	)
    if i < 2:
      return i
    return fib_recursive_term(i - 1	) + fib_recursive_term(i - 2	)

  if n < 0:
    raise Exception('''n is negative'''	)
  return [fib_recursive_term(_UpperCamelCase	) for i in range(n + 1	)]



def  snake_case_(_UpperCamelCase	)				->      list[int]:



  """simple docstring"""






  if n < 0:
    raise Exception('''n is negative'''	)
  # Cache must be outside recursuive function
  # other it will reset every time it calls itself.
  _snake_case		=				{0: 0, 1: 1, 2: 1}  # Prefilled cache

  def rec_fn_memoized(_UpperCamelCase	) -> int:
    if num in cache:
      return cache[num]

    _snake_case		=				rec_fn_memoized(num - 1	) + rec_fn_memoized(num - 2	)
    _snake_case		=				value
    return value

  return [rec_fn_memoized(_UpperCamelCase	) for i in range(n + 1	)]



def  snake_case_(_UpperCamelCase	)				->      list[int]:



  """simple docstring"""






  if n < 0:
    raise Exception('''n is negative'''	)
  if n >= 1_475:
    raise Exception('''n is too large'''	)
  _snake_case		=				sqrt(5	)
  _snake_case		=				(1 + sqrt_a) / 2
  return [round(phi**i / sqrt_a	) for i in range(n + 1	)]


if __name__ == "__main__":
 __A     						=      30
 time_func(fib_iterative, num)
 time_func(fib_recursive, num)  # Around 3s runtime
 time_func(fib_recursive_cached, num)  # Around 0ms runtime
 time_func(fib_memoization, num)
 time_func(fib_binet, num)

