





"""simple docstring"""



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 	__snake_case      (	SCREAMING_SNAKE_CASE__  ):





 """simple docstring"""


 def __init__(					self      ,		__lowerCamelCase      ,		__lowerCamelCase						):



     '''simple docstring'''




     super().__init__()

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

     __A					:		Union[str, Any]						=  None

 def       UpperCamelCase__(					self      ,		__lowerCamelCase=0						):



     '''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					:		Optional[int]						=  torch.device(F"""cuda:{gpu_id}"""						)

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

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

 def       UpperCamelCase__(					self      ,		__lowerCamelCase=0						):



     '''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					:		Optional[int]						=  torch.device(F"""cuda:{gpu_id}"""						)

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

     __A					:		Optional[Any]						=  None
     for cpu_offloaded_model in [self.unet]:
         __A					:		Dict						=  cpu_offload_with_hook(__lowerCamelCase      ,		__lowerCamelCase      ,		prev_module_hook=__lowerCamelCase						)

     if self.safety_checker is not None:
         __A					:		Optional[int]						=  cpu_offload_with_hook(self.safety_checker      ,		__lowerCamelCase      ,		prev_module_hook=__lowerCamelCase						)

     # We'll offload the last model manually.
     __A					:		int						=  hook

 @property
 # Copied from diffusers.pipelines.stable_diffusion.pipeline_stable_diffusion.StableDiffusionPipeline._execution_device
 def       UpperCamelCase__(					self						):



     '''simple docstring'''




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



     '''simple docstring'''




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

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

 def       UpperCamelCase__(					self      ,		__lowerCamelCase      ,		__lowerCamelCase = "pil"						):



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

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

     # Output_type must be 'pil'
     __A					:		Optional[Any]						=  self.numpy_to_pil(__lowerCamelCase						)
     return sample

 def       UpperCamelCase__(					self      ,		__lowerCamelCase      ,		__lowerCamelCase      ,		__lowerCamelCase=None						):



     '''simple docstring'''




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

 def       UpperCamelCase__(					self      ,		__lowerCamelCase      ,		__lowerCamelCase      ,		__lowerCamelCase      ,		__lowerCamelCase      ,		__lowerCamelCase      ,		__lowerCamelCase						):



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







 @torch.no_grad()
 @replace_example_docstring(__lowerCamelCase						)
 def __call__(					self      ,		__lowerCamelCase = 1      ,		__lowerCamelCase = None      ,		__lowerCamelCase = 1      ,		__lowerCamelCase = None      ,		__lowerCamelCase = None      ,		__lowerCamelCase = None      ,		__lowerCamelCase = "pil"      ,		__lowerCamelCase = True      ,		__lowerCamelCase = None      ,		__lowerCamelCase = 1      ,		):



     '''simple docstring'''




     __A					:		Optional[Any]						=  self.unet.config.sample_size
     __A					:		str						=  self._execution_device

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

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

     # 3. Handle class_labels for class-conditional models
     __A					:		List[Any]						=  self.prepare_class_labels(__lowerCamelCase      ,		__lowerCamelCase      ,		class_labels=__lowerCamelCase						)

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

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

             __A					:		Union[str, Any]						=  self.scheduler.step(__lowerCamelCase      ,		__lowerCamelCase      ,		__lowerCamelCase      ,		generator=__lowerCamelCase						)[0]

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

        # 6. Post-process image sample
     __A					:		Optional[Any]						=  self.postprocess_image(__lowerCamelCase      ,		output_type=__lowerCamelCase						)

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











"""simple docstring"""



import hashlib
import importlib.util
import json
import os
import pathlib
from types import ModuleType

import pytest
import requests

a_				=    pathlib.Path.cwd().joinpath("""project_euler""")
a_				=    pathlib.Path.cwd().joinpath(
    """scripts""", """project_euler_answers.json"""
)

with open(PROJECT_EULER_ANSWERS_PATH) as file_handle:
       a_				=    json.load(file_handle)







def 	__lowercase  (     snake_case_						:						pathlib.Path		)     ->ModuleType:



    '''simple docstring'''







    __A					:		str						=  importlib.util.spec_from_file_location(file_path.name       ,str(snake_case_		)		)
    __A					:		Optional[int]						=  importlib.util.module_from_spec(snake_case_		)  # type: ignore
    spec.loader.exec_module(snake_case_		)  # type: ignore
    return module







def 	__lowercase  (     )     ->list[pathlib.Path]:



    '''simple docstring'''







    __A					:		Optional[int]						=  []
    for problem_dir_path in PROJECT_EULER_DIR_PATH.iterdir():
        if problem_dir_path.is_file() or problem_dir_path.name.startswith('''_'''		):
            continue
        for file_path in problem_dir_path.iterdir():
            if file_path.suffix != ".py" or file_path.name.startswith(('''_''', '''test''')		):
                continue
            solution_file_paths.append(snake_case_		)
    return solution_file_paths







def 	__lowercase  (     )     ->str:



    '''simple docstring'''







    with open(os.environ['''GITHUB_EVENT_PATH''']		) as file:
        __A					:		Any						=  json.load(snake_case_		)
    return event["pull_request"]["url"] + "/files"







def 	__lowercase  (     )     ->list[pathlib.Path]:



    '''simple docstring'''







    __A					:		int						=  []
    __A					:		Optional[int]						=  {
        '''Accept''': '''application/vnd.github.v3+json''',
        '''Authorization''': '''token ''' + os.environ['''GITHUB_TOKEN'''],
    }
    __A					:		Optional[Any]						=  requests.get(get_files_url()       ,headers=snake_case_		).json()
    for file in files:
        __A					:		Union[str, Any]						=  pathlib.Path.cwd().joinpath(file['''filename''']		)
        if (
            filepath.suffix != ".py"
            or filepath.name.startswith(('''_''', '''test''')		)
            or not filepath.name.startswith('''sol'''		)
        ):
            continue
        solution_file_paths.append(snake_case_		)
    return solution_file_paths







def 	__lowercase  (     )     ->list[pathlib.Path]:



    '''simple docstring'''







    if os.environ.get('''CI'''		) and os.environ.get('''GITHUB_EVENT_NAME'''		) == "pull_request":
        # Return only if there are any, otherwise default to all solutions
        if filepaths := added_solution_file_path():
            return filepaths
    return all_solution_file_paths()







@pytest.mark.parametrize(
    '''solution_path'''       ,collect_solution_file_paths()       ,ids=lambda snake_case_		: F"""{path.parent.name}/{path.name}"""       ,)
def 	__lowercase  (     snake_case_						:						pathlib.Path		)     ->None:



    '''simple docstring'''







    __A					:		str						=  solution_path.parent.name[8:].zfill(3		)
    __A					:		str						=  PROBLEM_ANSWERS[problem_number]
    __A					:		Union[str, Any]						=  convert_path_to_module(snake_case_		)
    __A					:		List[str]						=  str(solution_module.solution()		)  # type: ignore
    __A					:		List[str]						=  hashlib.shaaaa(answer.encode()		).hexdigest()
    assert (
        answer == expected
    ), F"""Expected solution to {problem_number} to have hash {expected}, got {answer}"""




