





import importlib
import inspect
import json
import os
import re
import shutil
import sys
from pathlib import Path
from typing import Dict, Optional, Union
from urllib import request

from huggingface_hub import HfFolder, cached_download, hf_hub_download, model_info
from packaging import version

from .. import __version__
from . import DIFFUSERS_DYNAMIC_MODULE_NAME, HF_MODULES_CACHE, logging


_a 						=      (
    "https://raw.githubusercontent.com/huggingface/diffusers/{revision}/examples/community/{pipeline}.py"
)


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





def lowerCAmelCase__()				->      Tuple:







	'''simple docstring'''







	lowerCamelCase__								=		'''https://pypi.org/pypi/diffusers/json'''
	lowerCamelCase__								=		json.loads(request.urlopen(__a							).read()							)['''releases'''].keys()
	return sorted(__a       ,key=lambda __snake_case							: version.Version(__a							)							)





def lowerCAmelCase__()				->      Tuple:







	'''simple docstring'''







	if HF_MODULES_CACHE in sys.path:
		return

	sys.path.append(__a							)
	os.makedirs(__a       ,exist_ok=__a							)
	lowerCamelCase__								=		Path(__a							) / '''__init__.py'''
	if not init_path.exists():
		init_path.touch()





def lowerCAmelCase__(__snake_case							)				->      List[Any]:







	'''simple docstring'''







	init_hf_modules()
	lowerCamelCase__								=		Path(__a							) / name
	# If the parent module does not exist yet, recursively create it.
	if not dynamic_module_path.parent.exists():
		create_dynamic_module(dynamic_module_path.parent							)
	os.makedirs(__a       ,exist_ok=__a							)
	lowerCamelCase__								=		dynamic_module_path / '''__init__.py'''
	if not init_path.exists():
		init_path.touch()





def lowerCAmelCase__(__snake_case							)				->      List[Any]:







	'''simple docstring'''







	with open(__a       ,'''r'''       ,encoding='''utf-8'''							) as f:
		lowerCamelCase__								=		f.read()

	# Imports of the form `import .xxx`
	lowerCamelCase__								=		re.findall('''^\s*import\s+\.(\S+)\s*$'''       ,__a       ,flags=re.MULTILINE							)
	# Imports of the form `from .xxx import yyy`
	relative_imports += re.findall('''^\s*from\s+\.(\S+)\s+import'''       ,__a       ,flags=re.MULTILINE							)
	# Unique-ify
	return list(set(__a							)							)





def lowerCAmelCase__(__snake_case							)				->      str:







	'''simple docstring'''







	lowerCamelCase__								=		False
	lowerCamelCase__								=		[module_file]
	lowerCamelCase__								=		[]

	# Let's recurse through all relative imports
	while not no_change:
		lowerCamelCase__								=		[]
		for f in files_to_check:
			new_imports.extend(get_relative_imports(__a							)							)

		lowerCamelCase__								=		Path(__a							).parent
		lowerCamelCase__								=		[str(module_path / m							) for m in new_imports]
		lowerCamelCase__								=		[f for f in new_import_files if f not in all_relative_imports]
		lowerCamelCase__								=		[F'{f}.py' for f in new_import_files]

		lowerCamelCase__								=		len(__a							) == 0
		all_relative_imports.extend(__a							)

	return all_relative_imports





def lowerCAmelCase__(__snake_case							)				->      Tuple:







	'''simple docstring'''







	with open(__a       ,'''r'''       ,encoding='''utf-8'''							) as f:
		lowerCamelCase__								=		f.read()

	# Imports of the form `import xxx`
	lowerCamelCase__								=		re.findall('''^\s*import\s+(\S+)\s*$'''       ,__a       ,flags=re.MULTILINE							)
	# Imports of the form `from xxx import yyy`
	imports += re.findall('''^\s*from\s+(\S+)\s+import'''       ,__a       ,flags=re.MULTILINE							)
	# Only keep the top-level module
	lowerCamelCase__								=		[imp.split('''.'''							)[0] for imp in imports if not imp.startswith('''.'''							)]

	# Unique-ify and test we got them all
	lowerCamelCase__								=		list(set(__a							)							)
	lowerCamelCase__								=		[]
	for imp in imports:
		try:
			importlib.import_module(__a							)
		except ImportError:
			missing_packages.append(__a							)

	if len(__a							) > 0:
		raise ImportError(
		    '''This modeling file requires the following packages that were not found in your environment: '''
		    F'{", ".join(__a							)}. Run `pip install {" ".join(__a							)}`'							)

	return get_relative_imports(__a							)





def lowerCAmelCase__(__snake_case       ,__snake_case							)				->      int:







	'''simple docstring'''







	lowerCamelCase__								=		module_path.replace(os.path.sep       ,'''.'''							)
	lowerCamelCase__								=		importlib.import_module(__a							)

	if class_name is None:
		return find_pipeline_class(__a							)
	return getattr(__a       ,__a							)





def lowerCAmelCase__(__snake_case							)				->      Optional[int]:







	'''simple docstring'''







	from ..pipelines import DiffusionPipeline

	lowerCamelCase__								=		dict(inspect.getmembers(__a       ,inspect.isclass							)							)

	lowerCamelCase__								=		None
	for cls_name, cls in cls_members.items():
		if (
		    cls_name != DiffusionPipeline.__name__
		    and issubclass(cls       ,__a							)
		    and cls.__module__.split('''.'''							)[0] != "diffusers"
		):
			if pipeline_class is not None:
				raise ValueError(
				    F'Multiple classes that inherit from {DiffusionPipeline.__name__} have been found:'
				    F' {pipeline_class.__name__}, and {cls_name}. Please make sure to define only one in'
				    F' {loaded_module}.'							)
			lowerCamelCase__								=		cls

	return pipeline_class





def lowerCAmelCase__(__snake_case       ,__snake_case       ,__snake_case = None       ,__snake_case = False       ,__snake_case = False       ,__snake_case = None       ,__snake_case = None       ,__snake_case = None       ,__snake_case = False       ,)				->      List[Any]:







	'''simple docstring'''







	lowerCamelCase__								=		str(__a							)

	lowerCamelCase__								=		os.path.join(__a       ,__a							)

	if os.path.isfile(__a							):
		lowerCamelCase__								=		module_file_or_url
		lowerCamelCase__								=		'''local'''
	elif pretrained_model_name_or_path.count('''/'''							) == 0:
		lowerCamelCase__								=		get_diffusers_versions()
		# cut ".dev0"
		lowerCamelCase__								=		'''v''' + '''.'''.join(__version__.split('''.'''							)[:3]							)

		# retrieve github version that matches
		if revision is None:
			lowerCamelCase__								=		latest_version if latest_version[1:] in available_versions else '''main'''
			logger.info(F'Defaulting to latest_version: {revision}.'							)
		elif revision in available_versions:
			lowerCamelCase__								=		F'v{revision}'
		elif revision == "main":
			lowerCamelCase__								=		revision
		else:
			raise ValueError(
			    F'`custom_revision`: {revision} does not exist. Please make sure to choose one of'
			    F' {", ".join(available_versions + ["main"]							)}.'							)

		# community pipeline on GitHub
		lowerCamelCase__								=		COMMUNITY_PIPELINES_URL.format(revision=__a       ,pipeline=__a							)
		try:
			lowerCamelCase__								=		cached_download(
			    __a       ,cache_dir=__a       ,force_download=__a       ,proxies=__a       ,resume_download=__a       ,local_files_only=__a       ,use_auth_token=__a       ,)
			lowerCamelCase__								=		'''git'''
			lowerCamelCase__								=		pretrained_model_name_or_path + '''.py'''
		except EnvironmentError:
			logger.error(F'Could not locate the {module_file} inside {pretrained_model_name_or_path}.'							)
			raise
	else:
		try:
			# Load from URL or cache if already cached
			lowerCamelCase__								=		hf_hub_download(
			    __a       ,__a       ,cache_dir=__a       ,force_download=__a       ,proxies=__a       ,resume_download=__a       ,local_files_only=__a       ,use_auth_token=__a       ,)
			lowerCamelCase__								=		os.path.join('''local'''       ,'''--'''.join(pretrained_model_name_or_path.split('''/'''							)							)							)
		except EnvironmentError:
			logger.error(F'Could not locate the {module_file} inside {pretrained_model_name_or_path}.'							)
			raise

    # Check we have all the requirements in our environment
	lowerCamelCase__								=		check_imports(__a							)

	# Now we move the module inside our cached dynamic modules.
	lowerCamelCase__								=		DIFFUSERS_DYNAMIC_MODULE_NAME + os.path.sep + submodule
	create_dynamic_module(__a							)
	lowerCamelCase__								=		Path(__a							) / full_submodule
	if submodule == "local" or submodule == "git":
		# We always copy local files (we could hash the file to see if there was a change, and give them the name of
		# that hash, to only copy when there is a modification but it seems overkill for now).
		# The only reason we do the copy is to avoid putting too many folders in sys.path.
		shutil.copy(__a       ,submodule_path / module_file							)
		for module_needed in modules_needed:
			lowerCamelCase__								=		F'{module_needed}.py'
			shutil.copy(os.path.join(__a       ,__a							)       ,submodule_path / module_needed							)
	else:
		# Get the commit hash
		# TODO: we will get this info in the etag soon, so retrieve it from there and not here.
		if isinstance(__a       ,__a							):
			lowerCamelCase__								=		use_auth_token
		elif use_auth_token is True:
			lowerCamelCase__								=		HfFolder.get_token()
		else:
			lowerCamelCase__								=		None

		lowerCamelCase__								=		model_info(__a       ,revision=__a       ,token=__a							).sha

		# The module file will end up being placed in a subfolder with the git hash of the repo. This way we get the
		# benefit of versioning.
		lowerCamelCase__								=		submodule_path / commit_hash
		lowerCamelCase__								=		full_submodule + os.path.sep + commit_hash
		create_dynamic_module(__a							)

		if not (submodule_path / module_file).exists():
			shutil.copy(__a       ,submodule_path / module_file							)
		# Make sure we also have every file with relative
		for module_needed in modules_needed:
			if not (submodule_path / module_needed).exists():
				get_cached_module_file(
				    __a       ,F'{module_needed}.py'       ,cache_dir=__a       ,force_download=__a       ,resume_download=__a       ,proxies=__a       ,use_auth_token=__a       ,revision=__a       ,local_files_only=__a       ,)
	return os.path.join(__a       ,__a							)





def lowerCAmelCase__(__snake_case       ,__snake_case       ,__snake_case = None       ,__snake_case = None       ,__snake_case = False       ,__snake_case = False       ,__snake_case = None       ,__snake_case = None       ,__snake_case = None       ,__snake_case = False       ,**__snake_case       ,)				->      str:







	'''simple docstring'''







	lowerCamelCase__								=		get_cached_module_file(
	    __a       ,__a       ,cache_dir=__a       ,force_download=__a       ,resume_download=__a       ,proxies=__a       ,use_auth_token=__a       ,revision=__a       ,local_files_only=__a       ,)
	return get_class_in_module(__a       ,final_module.replace('''.py'''       ,''''''							)							)






import torch

from diffusers import DDPMScheduler

from .test_schedulers import SchedulerCommonTest

class 				UpperCAmelCase_		(			__lowercase   ):
							"""simple docstring"""







							UpperCAmelCase__       :						List[Any]	    =    (DDPMScheduler,)


							def 				__lowercase				(			self       ,							**_a       )    ->    Any:
												_a	:    List[Any]         =		{
												    '''num_train_timesteps''': 1_0_0_0,
												    '''beta_start''': 0.0001,
												    '''beta_end''': 0.02,
												    '''beta_schedule''': '''linear''',
												    '''variance_type''': '''fixed_small''',
												    '''clip_sample''': True,
												}

												config.update(**_a       )
												return config


							def 				__lowercase				(			self       )    ->    Any:
												for timesteps in [1, 5, 1_0_0, 1_0_0_0]:
																	self.check_over_configs(num_train_timesteps=_a       )


							def 				__lowercase				(			self       )    ->    List[Any]:
												for beta_start, beta_end in zip([0.0001, 0.001, 0.01, 0.1]       ,							[0.002, 0.02, 0.2, 2]       ):
																	self.check_over_configs(beta_start=_a       ,							beta_end=_a       )


							def 				__lowercase				(			self       )    ->    List[str]:
												for schedule in ["linear", "squaredcos_cap_v2"]:
																	self.check_over_configs(beta_schedule=_a       )


							def 				__lowercase				(			self       )    ->    Optional[Any]:
												for variance in ["fixed_small", "fixed_large", "other"]:
																	self.check_over_configs(variance_type=_a       )


							def 				__lowercase				(			self       )    ->    str:
												for clip_sample in [True, False]:
																	self.check_over_configs(clip_sample=_a       )


							def 				__lowercase				(			self       )    ->    Dict:
												self.check_over_configs(thresholding=_a       )
												for threshold in [0.5, 1.0, 2.0]:
																	for prediction_type in ["epsilon", "sample", "v_prediction"]:
																						self.check_over_configs(
																						    thresholding=_a       ,							prediction_type=_a       ,							sample_max_value=_a       ,							)


							def 				__lowercase				(			self       )    ->    Optional[Any]:
												for prediction_type in ["epsilon", "sample", "v_prediction"]:
																	self.check_over_configs(prediction_type=_a       )


							def 				__lowercase				(			self       )    ->    int:
												for t in [0, 5_0_0, 9_9_9]:
																	self.check_over_forward(time_step=_a       )


							def 				__lowercase				(			self       )    ->    int:
												_a	:    int         =		self.scheduler_classes[0]
												_a	:    List[Any]         =		self.get_scheduler_config()
												_a	:    Dict         =		scheduler_class(**_a       )

												assert torch.sum(torch.abs(scheduler._get_variance(0       ) - 0.0       )       ) < 1e-5
												assert torch.sum(torch.abs(scheduler._get_variance(4_8_7       ) - 0.0_0979       )       ) < 1e-5
												assert torch.sum(torch.abs(scheduler._get_variance(9_9_9       ) - 0.02       )       ) < 1e-5


							def 				__lowercase				(			self       )    ->    Tuple:
												_a	:    int         =		self.scheduler_classes[0]
												_a	:    int         =		self.get_scheduler_config()
												_a	:    int         =		scheduler_class(**_a       )

												_a	:    Optional[int]         =		len(_a       )

												_a	:    Optional[Any]         =		self.dummy_model()
												_a	:    str         =		self.dummy_sample_deter
												_a	:    Union[str, Any]         =		torch.manual_seed(0       )

												for t in reversed(range(_a       )       ):
																	# 1. predict noise residual
																	_a	:    str         =		model(_a       ,							_a       )

																	# 2. predict previous mean of sample x_t-1
																	_a	:    Optional[int]         =		scheduler.step(_a       ,							_a       ,							_a       ,							generator=_a       ).prev_sample

																	# if t > 0:
																	#     noise = self.dummy_sample_deter
																	#     variance = scheduler.get_variance(t) ** (0.5) * noise
																	#
																	# sample = pred_prev_sample + variance
																	_a	:    List[Any]         =		pred_prev_sample

												_a	:    str         =		torch.sum(torch.abs(_a       )       )
												_a	:    Optional[Any]         =		torch.mean(torch.abs(_a       )       )

												assert abs(result_sum.item() - 258.9606       ) < 1e-2
												assert abs(result_mean.item() - 0.3372       ) < 1e-3


							def 				__lowercase				(			self       )    ->    Optional[Any]:
												_a	:    Optional[int]         =		self.scheduler_classes[0]
												_a	:    Optional[Any]         =		self.get_scheduler_config(prediction_type='''v_prediction'''       )
												_a	:    Union[str, Any]         =		scheduler_class(**_a       )

												_a	:    Dict         =		len(_a       )

												_a	:    int         =		self.dummy_model()
												_a	:    Tuple         =		self.dummy_sample_deter
												_a	:    List[Any]         =		torch.manual_seed(0       )

												for t in reversed(range(_a       )       ):
																	# 1. predict noise residual
																	_a	:    Dict         =		model(_a       ,							_a       )

																	# 2. predict previous mean of sample x_t-1
																	_a	:    int         =		scheduler.step(_a       ,							_a       ,							_a       ,							generator=_a       ).prev_sample

																	# if t > 0:
																	#     noise = self.dummy_sample_deter
																	#     variance = scheduler.get_variance(t) ** (0.5) * noise
																	#
																	# sample = pred_prev_sample + variance
																	_a	:    str         =		pred_prev_sample

												_a	:    str         =		torch.sum(torch.abs(_a       )       )
												_a	:    Tuple         =		torch.mean(torch.abs(_a       )       )

												assert abs(result_sum.item() - 202.0296       ) < 1e-2
												assert abs(result_mean.item() - 0.2631       ) < 1e-3


							def 				__lowercase				(			self       )    ->    Dict:
												_a	:    Union[str, Any]         =		self.scheduler_classes[0]
												_a	:    Tuple         =		self.get_scheduler_config()
												_a	:    Any         =		scheduler_class(**_a       )

												_a	:    Optional[Any]         =		[1_0_0, 8_7, 5_0, 1, 0]

												scheduler.set_timesteps(timesteps=_a       )

												_a	:    Optional[int]         =		scheduler.timesteps

												for i, timestep in enumerate(_a       ):
																	if i == len(_a       ) - 1:
																						_a	:    Dict         =		-1
																	else:
																						_a	:    Tuple         =		timesteps[i + 1]

																	_a	:    Optional[Any]         =		scheduler.previous_timestep(_a       )
																	_a	:    Optional[Any]         =		prev_t.item()

																	self.assertEqual(_a       ,							_a       )


							def 				__lowercase				(			self       )    ->    Optional[Any]:
												_a	:    Dict         =		self.scheduler_classes[0]
												_a	:    List[str]         =		self.get_scheduler_config()
												_a	:    Tuple         =		scheduler_class(**_a       )

												_a	:    str         =		[1_0_0, 8_7, 5_0, 5_1, 0]

												with self.assertRaises(_a       ,							msg='''`custom_timesteps` must be in descending order.'''       ):
																	scheduler.set_timesteps(timesteps=_a       )


							def 				__lowercase				(			self       )    ->    str:
												_a	:    List[str]         =		self.scheduler_classes[0]
												_a	:    List[str]         =		self.get_scheduler_config()
												_a	:    Dict         =		scheduler_class(**_a       )

												_a	:    Union[str, Any]         =		[1_0_0, 8_7, 5_0, 1, 0]
												_a	:    Optional[Any]         =		len(_a       )

												with self.assertRaises(_a       ,							msg='''Can only pass one of `num_inference_steps` or `custom_timesteps`.'''       ):
																	scheduler.set_timesteps(num_inference_steps=_a       ,							timesteps=_a       )




							def 				__lowercase				(			self       )    ->    Optional[int]:
												_a	:    Dict         =		self.scheduler_classes[0]
												_a	:    Union[str, Any]         =		self.get_scheduler_config()
												_a	:    int         =		scheduler_class(**_a       )

												_a	:    str         =		[scheduler.config.num_train_timesteps]

												with self.assertRaises(
												    _a       ,							msg='''`timesteps` must start before `self.config.train_timesteps`: {scheduler.config.num_train_timesteps}}'''       ,							):
																	scheduler.set_timesteps(timesteps=_a       )



