




"""simple docstring"""





from pathlib import PurePosixPath
from typing import Optional

import fsspec
from fsspec import AbstractFileSystem
from huggingface_hub.hf_api import DatasetInfo

from ..utils.file_utils import get_authentication_headers_for_url
from ..utils.hub import hf_hub_url



class      a__				(	a__   ):
			lowercase_    =	""""""
			lowercase_    =	"""hf-legacy"""  # "hf://"" is reserved for hffs


			def __init__(      self      :    str      ,  UpperCamelCase_      :    Dict = None      ,  UpperCamelCase_      :    Dict = None      ,  **UpperCamelCase_      :    Tuple      ,  ):


						"""simple docstring"""




						super().__init__(self      ,  **lowerCAmelCase__)
						__UpperCAmelCase	: int		=					repo_info
						__UpperCAmelCase	: Dict		=					token
						__UpperCAmelCase	: Union[str, Any]		=					None


			def       a_     (      self      :    Any):


						"""simple docstring"""




						if self.dir_cache is None:
									__UpperCAmelCase	: Optional[Any]		=					{}
									for hf_file in self.repo_info.siblings:
												# TODO(QL): add sizes
												__UpperCAmelCase	: List[str]		=					{
												    "name": hf_file.rfilename,
												    "size": None,
												    "type": "file",
												}
												self.dir_cache.update(
												    {
												        str(lowerCAmelCase__): {"name": str(lowerCAmelCase__), "size": None, "type": "directory"}
												        for d in list(PurePosixPath(hf_file.rfilename).parents)[:-1]
												    })


			def       a_     (      self      :    str      ,  UpperCamelCase_      :    List[str]      ,  UpperCamelCase_      :    Optional[Any] = "rb"      ,  **UpperCamelCase_      :    Dict      ,  ):


						"""simple docstring"""




						if not isinstance(self.repo_info      ,  lowerCAmelCase__):
									raise NotImplementedError(F"Open is only implemented for dataset repositories, but got {self.repo_info}")
						__UpperCAmelCase	: Union[str, Any]		=					hf_hub_url(self.repo_info.id      ,  lowerCAmelCase__      ,  revision=self.repo_info.sha)
						return fsspec.open(
						    lowerCAmelCase__      ,  mode=lowerCAmelCase__      ,  headers=get_authentication_headers_for_url(lowerCAmelCase__      ,  use_auth_token=self.token)      ,  client_kwargs={"trust_env": True}      ,  ).open()


			def       a_     (      self      :    Tuple      ,  UpperCamelCase_      :    Union[str, Any]      ,  **UpperCamelCase_      :    str):


						"""simple docstring"""




						self._get_dirs()
						__UpperCAmelCase	: Any		=					self._strip_protocol(lowerCAmelCase__)
						if path in self.dir_cache:
									return self.dir_cache[path]
						else:
									raise FileNotFoundError(lowerCAmelCase__)


			def       a_     (      self      :    List[str]      ,  UpperCamelCase_      :    Any      ,  UpperCamelCase_      :    Tuple=False      ,  **UpperCamelCase_      :    int):


						"""simple docstring"""




						self._get_dirs()
						__UpperCAmelCase	: Optional[int]		=					PurePosixPath(path.strip("/"))
						__UpperCAmelCase	: Dict		=					{}
						for p, f in self.dir_cache.items():
									__UpperCAmelCase	: str		=					PurePosixPath(p.strip("/"))
									__UpperCAmelCase	: Tuple		=					p.parent
									if root == path:
												__UpperCAmelCase	: Union[str, Any]		=					f
						__UpperCAmelCase	: List[Any]		=					list(paths.values())
						if detail:
									return out
						else:
									return sorted(f["name"] for f in out)








"""simple docstring"""





import gc
import unittest

import numpy as np
import torch
from torch.backends.cuda import sdp_kernel

from diffusers import (
    CMStochasticIterativeScheduler,
    ConsistencyModelPipeline,
    UNetaDModel,
)
from diffusers.utils import randn_tensor, slow, torch_device
from diffusers.utils.testing_utils import enable_full_determinism, require_torch_a, require_torch_gpu

from ..pipeline_params import UNCONDITIONAL_IMAGE_GENERATION_BATCH_PARAMS, UNCONDITIONAL_IMAGE_GENERATION_PARAMS
from ..test_pipelines_common import PipelineTesterMixin


enable_full_determinism()



class      a__				(	__magic_name__							,    unittest.TestCase   ):
			lowercase_    =	ConsistencyModelPipeline
			lowercase_    =	UNCONDITIONAL_IMAGE_GENERATION_PARAMS
			lowercase_    =	UNCONDITIONAL_IMAGE_GENERATION_BATCH_PARAMS

			# Override required_optional_params to remove num_images_per_prompt
			lowercase_    =	frozenset(
			    [
			        "num_inference_steps",
			        "generator",
			        "latents",
			        "output_type",
			        "return_dict",
			        "callback",
			        "callback_steps",
			    ]   )


			@property
			def       a_     (      self      :    Optional[Any]):


						"""simple docstring"""




						__UpperCAmelCase	: Union[str, Any]		=					UNetaDModel.from_pretrained(
						    "diffusers/consistency-models-test"      ,  subfolder="test_unet"      ,  )
						return unet


			@property
			def       a_     (      self      :    List[Any]):


						"""simple docstring"""




						__UpperCAmelCase	: Dict		=					UNetaDModel.from_pretrained(
						    "diffusers/consistency-models-test"      ,  subfolder="test_unet_class_cond"      ,  )
						return unet


			def       a_     (      self      :    Any      ,  UpperCamelCase_      :    int=False):


						"""simple docstring"""




						if class_cond:
									__UpperCAmelCase	: List[Any]		=					self.dummy_cond_unet
						else:
									__UpperCAmelCase	: Optional[int]		=					self.dummy_uncond_unet

						# Default to CM multistep sampler
						__UpperCAmelCase	: List[str]		=					CMStochasticIterativeScheduler(
						    num_train_timesteps=40      ,  sigma_min=0.002      ,  sigma_max=80.0      ,  )

						__UpperCAmelCase	: Optional[int]		=					{
						    "unet": unet,
						    "scheduler": scheduler,
						}

						return components


			def       a_     (      self      :    List[str]      ,  UpperCamelCase_      :    Tuple      ,  UpperCamelCase_      :    List[Any]=0):


						"""simple docstring"""




						if str(UpperCamelCase_).startswith("mps"):
									__UpperCAmelCase	: str		=					torch.manual_seed(UpperCamelCase_)
						else:
									__UpperCAmelCase	: Optional[Any]		=					torch.Generator(device=UpperCamelCase_).manual_seed(UpperCamelCase_)

						__UpperCAmelCase	: List[Any]		=					{
						    "batch_size": 1,
						    "num_inference_steps": None,
						    "timesteps": [22, 0],
						    "generator": generator,
						    "output_type": "np",
						}

						return inputs


			def       a_     (      self      :    int):


						"""simple docstring"""




						__UpperCAmelCase	: Union[str, Any]		=					"cpu"  # ensure determinism for the device-dependent torch.Generator
						__UpperCAmelCase	: Union[str, Any]		=					self.get_dummy_components()
						__UpperCAmelCase	: str		=					ConsistencyModelPipeline(**UpperCamelCase_)
						__UpperCAmelCase	: Any		=					pipe.to(UpperCamelCase_)
						pipe.set_progress_bar_config(disable=UpperCamelCase_)

						__UpperCAmelCase	: Any		=					self.get_dummy_inputs(UpperCamelCase_)
						__UpperCAmelCase	: str		=					pipe(**UpperCamelCase_).images
						assert image.shape == (1, 32, 32, 3)

						__UpperCAmelCase	: Dict		=					image[0, -3:, -3:, -1]
						__UpperCAmelCase	: List[Any]		=					np.array([0.3572, 0.6273, 0.4031, 0.3961, 0.4321, 0.5730, 0.5266, 0.4780, 0.5004])

						assert np.abs(image_slice.flatten() - expected_slice).max() < 1e-3


			def       a_     (      self      :    Optional[int]):


						"""simple docstring"""




						__UpperCAmelCase	: List[Any]		=					"cpu"  # ensure determinism for the device-dependent torch.Generator
						__UpperCAmelCase	: Optional[int]		=					self.get_dummy_components(class_cond=UpperCamelCase_)
						__UpperCAmelCase	: Optional[Any]		=					ConsistencyModelPipeline(**UpperCamelCase_)
						__UpperCAmelCase	: str		=					pipe.to(UpperCamelCase_)
						pipe.set_progress_bar_config(disable=UpperCamelCase_)

						__UpperCAmelCase	: Tuple		=					self.get_dummy_inputs(UpperCamelCase_)
						__UpperCAmelCase	: Dict		=					0
						__UpperCAmelCase	: int		=					pipe(**UpperCamelCase_).images
						assert image.shape == (1, 32, 32, 3)

						__UpperCAmelCase	: Any		=					image[0, -3:, -3:, -1]
						__UpperCAmelCase	: List[Any]		=					np.array([0.3572, 0.6273, 0.4031, 0.3961, 0.4321, 0.5730, 0.5266, 0.4780, 0.5004])

						assert np.abs(image_slice.flatten() - expected_slice).max() < 1e-3


			def       a_     (      self      :    Optional[Any]):


						"""simple docstring"""




						__UpperCAmelCase	: Optional[int]		=					"cpu"  # ensure determinism for the device-dependent torch.Generator
						__UpperCAmelCase	: Optional[int]		=					self.get_dummy_components()
						__UpperCAmelCase	: Optional[Any]		=					ConsistencyModelPipeline(**UpperCamelCase_)
						__UpperCAmelCase	: Optional[Any]		=					pipe.to(UpperCamelCase_)
						pipe.set_progress_bar_config(disable=UpperCamelCase_)

						__UpperCAmelCase	: List[str]		=					self.get_dummy_inputs(UpperCamelCase_)
						__UpperCAmelCase	: Dict		=					1
						__UpperCAmelCase	: int		=					None
						__UpperCAmelCase	: List[Any]		=					pipe(**UpperCamelCase_).images
						assert image.shape == (1, 32, 32, 3)

						__UpperCAmelCase	: Tuple		=					image[0, -3:, -3:, -1]
						__UpperCAmelCase	: Union[str, Any]		=					np.array([0.5004, 0.5004, 0.4994, 0.5008, 0.4976, 0.5018, 0.4990, 0.4982, 0.4987])

						assert np.abs(image_slice.flatten() - expected_slice).max() < 1e-3


			def       a_     (      self      :    List[str]):


						"""simple docstring"""




						__UpperCAmelCase	: Dict		=					"cpu"  # ensure determinism for the device-dependent torch.Generator
						__UpperCAmelCase	: List[str]		=					self.get_dummy_components(class_cond=UpperCamelCase_)
						__UpperCAmelCase	: Tuple		=					ConsistencyModelPipeline(**UpperCamelCase_)
						__UpperCAmelCase	: int		=					pipe.to(UpperCamelCase_)
						pipe.set_progress_bar_config(disable=UpperCamelCase_)

						__UpperCAmelCase	: Tuple		=					self.get_dummy_inputs(UpperCamelCase_)
						__UpperCAmelCase	: Optional[int]		=					1
						__UpperCAmelCase	: Optional[int]		=					None
						__UpperCAmelCase	: Tuple		=					0
						__UpperCAmelCase	: Tuple		=					pipe(**UpperCamelCase_).images
						assert image.shape == (1, 32, 32, 3)

						__UpperCAmelCase	: List[str]		=					image[0, -3:, -3:, -1]
						__UpperCAmelCase	: Dict		=					np.array([0.5004, 0.5004, 0.4994, 0.5008, 0.4976, 0.5018, 0.4990, 0.4982, 0.4987])

						assert np.abs(image_slice.flatten() - expected_slice).max() < 1e-3






@slow
@require_torch_gpu
class      a__				(	unittest.TestCase   ):


			def       a_     (      self      :    Optional[int]):


						"""simple docstring"""




						super().tearDown()
						gc.collect()
						torch.cuda.empty_cache()


			def       a_     (      self      :    Union[str, Any]      ,  UpperCamelCase_      :    Optional[int]=0      ,  UpperCamelCase_      :    Tuple=False      ,  UpperCamelCase_      :    int="cpu"      ,  UpperCamelCase_      :    Any=torch.floataa      ,  UpperCamelCase_      :    List[str]=(1, 3, 64, 64)):


						"""simple docstring"""




						__UpperCAmelCase	: Optional[int]		=					torch.manual_seed(UpperCamelCase_)

						__UpperCAmelCase	: int		=					{
						    "num_inference_steps": None,
						    "timesteps": [22, 0],
						    "class_labels": 0,
						    "generator": generator,
						    "output_type": "np",
						}

						if get_fixed_latents:
									__UpperCAmelCase	: int		=					self.get_fixed_latents(seed=UpperCamelCase_      ,  device=UpperCamelCase_      ,  dtype=UpperCamelCase_      ,  shape=UpperCamelCase_)
									__UpperCAmelCase	: Optional[int]		=					latents

						return inputs


			def       a_     (      self      :    Union[str, Any]      ,  UpperCamelCase_      :    int=0      ,  UpperCamelCase_      :    Tuple="cpu"      ,  UpperCamelCase_      :    Tuple=torch.floataa      ,  UpperCamelCase_      :    Optional[Any]=(1, 3, 64, 64)):


						"""simple docstring"""




						if type(UpperCamelCase_) == str:
									__UpperCAmelCase	: Union[str, Any]		=					torch.device(UpperCamelCase_)
						__UpperCAmelCase	: Union[str, Any]		=					torch.Generator(device=UpperCamelCase_).manual_seed(UpperCamelCase_)
						__UpperCAmelCase	: Optional[int]		=					randn_tensor(UpperCamelCase_      ,  generator=UpperCamelCase_      ,  device=UpperCamelCase_      ,  dtype=UpperCamelCase_)
						return latents


			def       a_     (      self      :    List[Any]):


						"""simple docstring"""




						__UpperCAmelCase	: Optional[int]		=					UNetaDModel.from_pretrained("diffusers/consistency_models"      ,  subfolder="diffusers_cd_imagenet64_l2")
						__UpperCAmelCase	: Dict		=					CMStochasticIterativeScheduler(
						    num_train_timesteps=40      ,  sigma_min=0.002      ,  sigma_max=80.0      ,  )
						__UpperCAmelCase	: Dict		=					ConsistencyModelPipeline(unet=UpperCamelCase_      ,  scheduler=UpperCamelCase_)
						pipe.to(torch_device=UpperCamelCase_)
						pipe.set_progress_bar_config(disable=UpperCamelCase_)

						__UpperCAmelCase	: Dict		=					self.get_inputs()
						__UpperCAmelCase	: List[str]		=					pipe(**UpperCamelCase_).images
						assert image.shape == (1, 64, 64, 3)

						__UpperCAmelCase	: List[Any]		=					image[0, -3:, -3:, -1]

						__UpperCAmelCase	: Union[str, Any]		=					np.array([0.0888, 0.0881, 0.0666, 0.0479, 0.0292, 0.0195, 0.0201, 0.0163, 0.0254])

						assert np.abs(image_slice.flatten() - expected_slice).max() < 2e-2


			def       a_     (      self      :    Tuple):


						"""simple docstring"""




						__UpperCAmelCase	: List[str]		=					UNetaDModel.from_pretrained("diffusers/consistency_models"      ,  subfolder="diffusers_cd_imagenet64_l2")
						__UpperCAmelCase	: Optional[Any]		=					CMStochasticIterativeScheduler(
						    num_train_timesteps=40      ,  sigma_min=0.002      ,  sigma_max=80.0      ,  )
						__UpperCAmelCase	: Any		=					ConsistencyModelPipeline(unet=UpperCamelCase_      ,  scheduler=UpperCamelCase_)
						pipe.to(torch_device=UpperCamelCase_)
						pipe.set_progress_bar_config(disable=UpperCamelCase_)

						__UpperCAmelCase	: int		=					self.get_inputs()
						__UpperCAmelCase	: str		=					1
						__UpperCAmelCase	: Union[str, Any]		=					None
						__UpperCAmelCase	: Tuple		=					pipe(**UpperCamelCase_).images
						assert image.shape == (1, 64, 64, 3)

						__UpperCAmelCase	: str		=					image[0, -3:, -3:, -1]

						__UpperCAmelCase	: Union[str, Any]		=					np.array([0.0340, 0.0152, 0.0063, 0.0267, 0.0221, 0.0107, 0.0416, 0.0186, 0.0217])

						assert np.abs(image_slice.flatten() - expected_slice).max() < 2e-2


			@require_torch_a
			def       a_     (      self      :    Tuple):


						"""simple docstring"""




						__UpperCAmelCase	: int		=					UNetaDModel.from_pretrained("diffusers/consistency_models"      ,  subfolder="diffusers_cd_imagenet64_l2")
						__UpperCAmelCase	: Optional[int]		=					CMStochasticIterativeScheduler(
						    num_train_timesteps=40      ,  sigma_min=0.002      ,  sigma_max=80.0      ,  )
						__UpperCAmelCase	: int		=					ConsistencyModelPipeline(unet=UpperCamelCase_      ,  scheduler=UpperCamelCase_)
						pipe.to(torch_device=UpperCamelCase_      ,  torch_dtype=torch.floataa)
						pipe.set_progress_bar_config(disable=UpperCamelCase_)

						__UpperCAmelCase	: Tuple		=					self.get_inputs(get_fixed_latents=UpperCamelCase_      ,  device=UpperCamelCase_)
						# Ensure usage of flash attention in torch 2.0
						with sdp_kernel(enable_flash=UpperCamelCase_      ,  enable_math=UpperCamelCase_      ,  enable_mem_efficient=UpperCamelCase_):
									__UpperCAmelCase	: List[str]		=					pipe(**UpperCamelCase_).images
						assert image.shape == (1, 64, 64, 3)

						__UpperCAmelCase	: List[str]		=					image[0, -3:, -3:, -1]

						__UpperCAmelCase	: Optional[int]		=					np.array([0.1875, 0.1428, 0.1289, 0.2151, 0.2092, 0.1477, 0.1877, 0.1641, 0.1353])

						assert np.abs(image_slice.flatten() - expected_slice).max() < 1e-3


			@require_torch_a
			def       a_     (      self      :    str):


						"""simple docstring"""




						__UpperCAmelCase	: Tuple		=					UNetaDModel.from_pretrained("diffusers/consistency_models"      ,  subfolder="diffusers_cd_imagenet64_l2")
						__UpperCAmelCase	: Union[str, Any]		=					CMStochasticIterativeScheduler(
						    num_train_timesteps=40      ,  sigma_min=0.002      ,  sigma_max=80.0      ,  )
						__UpperCAmelCase	: List[Any]		=					ConsistencyModelPipeline(unet=UpperCamelCase_      ,  scheduler=UpperCamelCase_)
						pipe.to(torch_device=UpperCamelCase_      ,  torch_dtype=torch.floataa)
						pipe.set_progress_bar_config(disable=UpperCamelCase_)

						__UpperCAmelCase	: Optional[int]		=					self.get_inputs(get_fixed_latents=UpperCamelCase_      ,  device=UpperCamelCase_)
						__UpperCAmelCase	: List[str]		=					1
						__UpperCAmelCase	: Any		=					None
						# Ensure usage of flash attention in torch 2.0
						with sdp_kernel(enable_flash=UpperCamelCase_      ,  enable_math=UpperCamelCase_      ,  enable_mem_efficient=UpperCamelCase_):
									__UpperCAmelCase	: List[str]		=					pipe(**UpperCamelCase_).images
						assert image.shape == (1, 64, 64, 3)

						__UpperCAmelCase	: int		=					image[0, -3:, -3:, -1]

						__UpperCAmelCase	: List[str]		=					np.array([0.1663, 0.1948, 0.2275, 0.1680, 0.1204, 0.1245, 0.1858, 0.1338, 0.2095])

						assert np.abs(image_slice.flatten() - expected_slice).max() < 1e-3


