





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


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


lowerCAmelCase							:       Union[str, Any]					 =	logging.get_logger(__name__)  # pylint: disable=invalid-name






def 		lowerCAmelCase  (	)      ->    str:

						"""simple docstring"""





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






def 		lowerCAmelCase  (	)      ->    Union[str, Any]:

						"""simple docstring"""





						if HF_MODULES_CACHE in sys.path:
												return

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






def 		lowerCAmelCase  (	UpperCamelCase__   :   Union[str, os.PathLike]					)      ->    List[str]:

						"""simple docstring"""





						init_hf_modules()
						__SCREAMING_SNAKE_CASE:				Dict           =   Path(lowerCamelCase__					) / 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(lowerCamelCase__       ,			exist_ok=lowerCamelCase__					)
						__SCREAMING_SNAKE_CASE:				List[str]           =   dynamic_module_path / '''__init__.py'''
						if not init_path.exists():
												init_path.touch()






def 		lowerCAmelCase  (	UpperCamelCase__   :   Any					)      ->    Union[str, Any]:

						"""simple docstring"""





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

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






def 		lowerCAmelCase  (	UpperCamelCase__   :   Optional[Any]					)      ->    Union[str, Any]:

						"""simple docstring"""





						__SCREAMING_SNAKE_CASE:				Tuple           =   False
						__SCREAMING_SNAKE_CASE:				Dict           =   [module_file]
						__SCREAMING_SNAKE_CASE:				Union[str, Any]           =   []

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

												__SCREAMING_SNAKE_CASE:				Optional[Any]           =   Path(lowerCamelCase__					).parent
												__SCREAMING_SNAKE_CASE:				Optional[int]           =   [str(module_path / m					) for m in new_imports]
												__SCREAMING_SNAKE_CASE:				Optional[int]           =   [f for f in new_import_files if f not in all_relative_imports]
												__SCREAMING_SNAKE_CASE:				List[str]           =   [F"""{f}.py""" for f in new_import_files]

												__SCREAMING_SNAKE_CASE:				str           =   len(lowerCamelCase__					) == 0
												all_relative_imports.extend(lowerCamelCase__					)

						return all_relative_imports






def 		lowerCAmelCase  (	UpperCamelCase__   :   int					)      ->    str:

						"""simple docstring"""





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

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

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

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

						return get_relative_imports(lowerCamelCase__					)






def 		lowerCAmelCase  (	UpperCamelCase__   :   List[str]       ,			UpperCamelCase__   :   Optional[Any]					)      ->    Optional[Any]:

						"""simple docstring"""





						__SCREAMING_SNAKE_CASE:				Union[str, Any]           =   module_path.replace(os.path.sep       ,			'''.'''					)
						__SCREAMING_SNAKE_CASE:				List[str]           =   importlib.import_module(lowerCamelCase__					)

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






def 		lowerCAmelCase  (	UpperCamelCase__   :   List[str]					)      ->    Optional[Any]:

						"""simple docstring"""





						from ..pipelines import DiffusionPipeline

						__SCREAMING_SNAKE_CASE:				List[Any]           =   dict(inspect.getmembers(lowerCamelCase__       ,			inspect.isclass					)					)

						__SCREAMING_SNAKE_CASE:				str           =   None
						for cls_name, cls in cls_members.items():
												if (
												    cls_name != DiffusionPipeline.__name__
												    and issubclass(cls       ,			lowerCamelCase__					)
												    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}."""					)
																		__SCREAMING_SNAKE_CASE:				Optional[int]           =   cls

						return pipeline_class






def 		lowerCAmelCase  (	UpperCamelCase__   :   Union[str, os.PathLike]       ,			UpperCamelCase__   :   str       ,			UpperCamelCase__   :   Optional[Union[str, os.PathLike]] = None       ,			UpperCamelCase__   :   bool = False       ,			UpperCamelCase__   :   bool = False       ,			UpperCamelCase__   :   Optional[Dict[str, str]] = None       ,			UpperCamelCase__   :   Optional[Union[bool, str]] = None       ,			UpperCamelCase__   :   Optional[str] = None       ,			UpperCamelCase__   :   bool = False       ,			)      ->    str:

						"""simple docstring"""





						__SCREAMING_SNAKE_CASE:				int           =   str(lowerCamelCase__					)

						__SCREAMING_SNAKE_CASE:				str           =   os.path.join(lowerCamelCase__       ,			lowerCamelCase__					)

						if os.path.isfile(lowerCamelCase__					):
												__SCREAMING_SNAKE_CASE:				Optional[int]           =   module_file_or_url
												__SCREAMING_SNAKE_CASE:				Tuple           =   '''local'''
						elif pretrained_model_name_or_path.count('''/'''					) == 0:
												__SCREAMING_SNAKE_CASE:				Optional[int]           =   get_diffusers_versions()
												# cut ".dev0"
												__SCREAMING_SNAKE_CASE:				Union[str, Any]           =   '''v''' + '''.'''.join(__version__.split('''.'''					)[:3]					)

												# retrieve github version that matches
												if revision is None:
																		__SCREAMING_SNAKE_CASE:				List[str]           =   latest_version if latest_version[1:] in available_versions else '''main'''
																		logger.info(F"""Defaulting to latest_version: {revision}."""					)
												elif revision in available_versions:
																		__SCREAMING_SNAKE_CASE:				Any           =   F"""v{revision}"""
												elif revision == "main":
																		__SCREAMING_SNAKE_CASE:				str           =   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
												__SCREAMING_SNAKE_CASE:				Union[str, Any]           =   COMMUNITY_PIPELINES_URL.format(revision=lowerCamelCase__       ,			pipeline=lowerCamelCase__					)
												try:
																		__SCREAMING_SNAKE_CASE:				Union[str, Any]           =   cached_download(
																		    lowerCamelCase__       ,			cache_dir=lowerCamelCase__       ,			force_download=lowerCamelCase__       ,			proxies=lowerCamelCase__       ,			resume_download=lowerCamelCase__       ,			local_files_only=lowerCamelCase__       ,			use_auth_token=lowerCamelCase__       ,			)
																		__SCREAMING_SNAKE_CASE:				str           =   '''git'''
																		__SCREAMING_SNAKE_CASE:				Optional[int]           =   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
																		__SCREAMING_SNAKE_CASE:				Union[str, Any]           =   hf_hub_download(
																		    lowerCamelCase__       ,			lowerCamelCase__       ,			cache_dir=lowerCamelCase__       ,			force_download=lowerCamelCase__       ,			proxies=lowerCamelCase__       ,			resume_download=lowerCamelCase__       ,			local_files_only=lowerCamelCase__       ,			use_auth_token=lowerCamelCase__       ,			)
																		__SCREAMING_SNAKE_CASE:				Dict           =   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
						__SCREAMING_SNAKE_CASE:				str           =   check_imports(lowerCamelCase__					)

						# Now we move the module inside our cached dynamic modules.
						__SCREAMING_SNAKE_CASE:				int           =   DIFFUSERS_DYNAMIC_MODULE_NAME + os.path.sep + submodule
						create_dynamic_module(lowerCamelCase__					)
						__SCREAMING_SNAKE_CASE:				Dict           =   Path(lowerCamelCase__					) / 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(lowerCamelCase__       ,			submodule_path / module_file					)
												for module_needed in modules_needed:
																		__SCREAMING_SNAKE_CASE:				List[Any]           =   F"""{module_needed}.py"""
																		shutil.copy(os.path.join(lowerCamelCase__       ,			lowerCamelCase__					)       ,			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(lowerCamelCase__       ,			lowerCamelCase__					):
																		__SCREAMING_SNAKE_CASE:				Union[str, Any]           =   use_auth_token
												elif use_auth_token is True:
																		__SCREAMING_SNAKE_CASE:				List[Any]           =   HfFolder.get_token()
												else:
																		__SCREAMING_SNAKE_CASE:				Optional[int]           =   None

												__SCREAMING_SNAKE_CASE:				List[Any]           =   model_info(lowerCamelCase__       ,			revision=lowerCamelCase__       ,			token=lowerCamelCase__					).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.
												__SCREAMING_SNAKE_CASE:				Union[str, Any]           =   submodule_path / commit_hash
												__SCREAMING_SNAKE_CASE:				Union[str, Any]           =   full_submodule + os.path.sep + commit_hash
												create_dynamic_module(lowerCamelCase__					)

												if not (submodule_path / module_file).exists():
																		shutil.copy(lowerCamelCase__       ,			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(
																								    lowerCamelCase__       ,			F"""{module_needed}.py"""       ,			cache_dir=lowerCamelCase__       ,			force_download=lowerCamelCase__       ,			resume_download=lowerCamelCase__       ,			proxies=lowerCamelCase__       ,			use_auth_token=lowerCamelCase__       ,			revision=lowerCamelCase__       ,			local_files_only=lowerCamelCase__       ,			)
						return os.path.join(lowerCamelCase__       ,			lowerCamelCase__					)






def 		lowerCAmelCase  (	UpperCamelCase__   :   Union[str, os.PathLike]       ,			UpperCamelCase__   :   str       ,			UpperCamelCase__   :   Optional[str] = None       ,			UpperCamelCase__   :   Optional[Union[str, os.PathLike]] = None       ,			UpperCamelCase__   :   bool = False       ,			UpperCamelCase__   :   bool = False       ,			UpperCamelCase__   :   Optional[Dict[str, str]] = None       ,			UpperCamelCase__   :   Optional[Union[bool, str]] = None       ,			UpperCamelCase__   :   Optional[str] = None       ,			UpperCamelCase__   :   bool = False       ,			**UpperCamelCase__   :   Dict       ,			)      ->    Tuple:

						"""simple docstring"""





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







# limitations under the License.


from typing import Optional, Tuple, Union

import torch

from diffusers import DiffusionPipeline, ImagePipelineOutput







class        __lowercase       (							a_  ):



		"""simple docstring"""

		def __init__( self		,      A		,      A	)  -> List[Any]:







					'''simple docstring'''







					super().__init__()
					self.register_modules(unet=A		,      scheduler=A	)







		@torch.no_grad()
		def __call__( self		,      A = 1		,      A = None		,      A = 50		,      A = "pil"		,      A = True		,      **A		,      )  -> Union[ImagePipelineOutput, Tuple]:







					'''simple docstring'''







					lowerCamelCase    		=  torch.randn(
					    (batch_size, self.unet.config.in_channels, self.unet.config.sample_size, self.unet.config.sample_size)		,      generator=A		,      )
					lowerCamelCase    		=  image.to(self.device	)

					# set step values
					self.scheduler.set_timesteps(A	)

					for t in self.progress_bar(self.scheduler.timesteps	):
								# 1. predict noise model_output
								lowerCamelCase    		=  self.unet(A		,      A	).sample

								# 2. predict previous mean of image x_t-1 and add variance depending on eta
								# eta corresponds to η in paper and should be between [0, 1]
								# do x_t -> x_t-1
								lowerCamelCase    		=  self.scheduler.step(A		,      A		,      A	).prev_sample

					lowerCamelCase    		=  (image / 2 + 0.5).clamp(0		,      1	)
					lowerCamelCase    		=  image.cpu().permute(0		,      2		,      3		,      1	).numpy()
					if output_type == "pil":
								lowerCamelCase    		=  self.numpy_to_pil(A	)

					if not return_dict:
								return (image,), "This is a local test"

					return ImagePipelineOutput(images=A	), "This is a local test"

