






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__            =	"\n    Examples:\n        ```py\n        >>> import torch\n\n        >>> from diffusers import ConsistencyModelPipeline\n\n        >>> device = \"cuda\"\n        >>> # Load the cd_imagenet64_l2 checkpoint.\n        >>> model_id_or_path = \"openai/diffusers-cd_imagenet64_l2\"\n        >>> pipe = ConsistencyModelPipeline.from_pretrained(model_id_or_path, torch_dtype=torch.float16)\n        >>> pipe.to(device)\n\n        >>> # Onestep Sampling\n        >>> image = pipe(num_inference_steps=1).images[0]\n        >>> image.save(\"cd_imagenet64_l2_onestep_sample.png\")\n\n        >>> # Onestep sampling, class-conditional image generation\n        >>> # ImageNet-64 class label 145 corresponds to king penguins\n        >>> image = pipe(num_inference_steps=1, class_labels=145).images[0]\n        >>> image.save(\"cd_imagenet64_l2_onestep_sample_penguin.png\")\n\n        >>> # Multistep sampling, class-conditional image generation\n        >>> # Timesteps can be explicitly specified; the particular timesteps below are from the original Github repo:\n        >>> # https://github.com/openai/consistency_models/blob/main/scripts/launch.sh#L77\n        >>> image = pipe(num_inference_steps=None, timesteps=[22, 0], class_labels=145).images[0]\n        >>> image.save(\"cd_imagenet64_l2_multistep_sample_penguin.png\")\n        ```\n"






class 	lowercase_      (			lowercase     ):






      '''simple docstring'''




      def __init__(   self   :		Tuple				,      __UpperCAmelCase   :		UNetaDModel				,      __UpperCAmelCase   :		CMStochasticIterativeScheduler						)					->None:




       """simple docstring"""


       super().__init__()

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

       a							=   None







      def    __lowerCAmelCase      (   self   :		Optional[Any]				,      __UpperCAmelCase   :		int=0						)					->Optional[Any]:




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

       a							=   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    __lowerCAmelCase      (   self   :		Any				,      __UpperCAmelCase   :		Union[str, Any]=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.'''						)

       a							=   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)

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

       if self.safety_checker is not None:
        a						,  a							=   cpu_offload_with_hook(self.safety_checker				,      __UpperCAmelCase				,      prev_module_hook=__UpperCAmelCase						)

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







      @property
      # Copied from diffusers.pipelines.stable_diffusion.pipeline_stable_diffusion.StableDiffusionPipeline._execution_device
      def    __lowerCAmelCase      (   self   :		List[Any]						)					->Union[str, Any]:




       """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    __lowerCAmelCase      (   self   :		Optional[Any]				,      __UpperCAmelCase   :		Optional[int]				,      __UpperCAmelCase   :		str				,      __UpperCAmelCase   :		List[Any]				,      __UpperCAmelCase   :		Optional[Any]				,      __UpperCAmelCase   :		Union[str, Any]				,      __UpperCAmelCase   :		Tuple				,      __UpperCAmelCase   :		List[Any]				,      __UpperCAmelCase   :		str=None						)					->List[str]:




       """simple docstring"""


       a							=   (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:
        a							=   randn_tensor(__UpperCAmelCase				,      generator=__UpperCAmelCase				,      device=__UpperCAmelCase				,      dtype=__UpperCAmelCase						)
       else:
        a							=   latents.to(device=__UpperCAmelCase				,      dtype=__UpperCAmelCase						)

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







      def    __lowerCAmelCase      (   self   :		str				,      __UpperCAmelCase   :		torch.FloatTensor				,      __UpperCAmelCase   :		str = "pil"						)					->Optional[int]:




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

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

       # Output_type must be 'pil'
       a							=   self.numpy_to_pil(__UpperCAmelCase						)
       return sample







      def    __lowerCAmelCase      (   self   :		int				,      __UpperCAmelCase   :		Union[str, Any]				,      __UpperCAmelCase   :		Dict				,      __UpperCAmelCase   :		Dict=None						)					->List[Any]:




       """simple docstring"""


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







      def    __lowerCAmelCase      (   self   :		Tuple				,      __UpperCAmelCase   :		int				,      __UpperCAmelCase   :		Optional[int]				,      __UpperCAmelCase   :		List[Any]				,      __UpperCAmelCase   :		Union[str, Any]				,      __UpperCAmelCase   :		Tuple				,      __UpperCAmelCase   :		int						)					->Union[str, 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:
        a							=   (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   :		List[str]				,      __UpperCAmelCase   :		int = 1				,      __UpperCAmelCase   :		Optional[Union[torch.Tensor, List[int], int]] = None				,      __UpperCAmelCase   :		int = 1				,      __UpperCAmelCase   :		List[int] = None				,      __UpperCAmelCase   :		Optional[Union[torch.Generator, List[torch.Generator]]] = None				,      __UpperCAmelCase   :		Optional[torch.FloatTensor] = None				,      __UpperCAmelCase   :		Optional[str] = "pil"				,      __UpperCAmelCase   :		bool = True				,      __UpperCAmelCase   :		Optional[Callable[[int, int, torch.FloatTensor], None]] = None				,      __UpperCAmelCase   :		int = 1				,      )					->str:




       """simple docstring"""


       a							=   self.unet.config.sample_size
       a							=   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)
       a							=   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
       a							=   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						)
        a							=   self.scheduler.timesteps
        a							=   len(__UpperCAmelCase						)
       else:
        self.scheduler.set_timesteps(__UpperCAmelCase						)
        a							=   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						):
         a							=   self.scheduler.scale_model_input(__UpperCAmelCase				,      __UpperCAmelCase						)
         a							=   self.unet(__UpperCAmelCase				,      __UpperCAmelCase				,      class_labels=__UpperCAmelCase				,      return_dict=__UpperCAmelCase						)[0]

         a							=   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
       a							=   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						)










def 		_a					(   a				:int				)							->    int:
 if not isinstance(a			,							a				):
  raise TypeError('''recursive_lucas_number accepts only integer arguments.'''				)
 if n_th_number == 0:
  return 2
 if n_th_number == 1:
  return 1

 return recursive_lucas_number(n_th_number - 1				) + recursive_lucas_number(
     n_th_number - 2				)
def 		_a					(   a				:int				)							->    int:
 if not isinstance(a			,							a				):
  raise TypeError('''dynamic_lucas_number accepts only integer arguments.'''				)
 a						,  a							=   2, 1
 for _ in range(a				):
  a						,  a							=   b, a + b
 return a


if __name__ == "__main__":
     from doctest import testmod

     testmod()
     UpperCAmelCase__            =	int(input("Enter the number of terms in lucas series:\n").strip())
     print("Using recursive function to calculate lucas series:")
     print(" ".join(str(recursive_lucas_number(i)) for i in range(n)))
     print("\nUsing dynamic function to calculate lucas series:")
     print(" ".join(str(dynamic_lucas_number(i)) for i in range(n)))


