






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


__UpperCamelCase      :      List[Any]     						=						(
    """https://raw.githubusercontent.com/huggingface/diffusers/{revision}/examples/community/{pipeline}.py"""
)


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





def 				snake_case  (     ):




	'''simple docstring'''







	__lowercase             =  """https://pypi.org/pypi/diffusers/json"""
	__lowercase             =  json.loads(request.urlopen(lowerCamelCase					).read()					)["""releases"""].keys()
	return sorted(lowerCamelCase				,  key=lambda lowerCamelCase					: version.Version(lowerCamelCase					)					)





def 				snake_case  (     ):




	'''simple docstring'''







	if HF_MODULES_CACHE in sys.path:
		return

	sys.path.append(lowerCamelCase					)
	os.makedirs(lowerCamelCase				,  exist_ok=lowerCamelCase					)
	__lowercase             =  Path(lowerCamelCase					) / """__init__.py"""
	if not init_path.exists():
		init_path.touch()





def 				snake_case  (     lowerCamelCase					):




	'''simple docstring'''







	init_hf_modules()
	__lowercase             =  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					)
	__lowercase             =  dynamic_module_path / """__init__.py"""
	if not init_path.exists():
		init_path.touch()





def 				snake_case  (     lowerCamelCase					):




	'''simple docstring'''







	with open(lowerCamelCase				,  """r"""				,  encoding="""utf-8"""					) as f:
		__lowercase             =  f.read()

	# Imports of the form `import .xxx`
	__lowercase             =  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 				snake_case  (     lowerCamelCase					):




	'''simple docstring'''







	__lowercase             =  False
	__lowercase             =  [module_file]
	__lowercase             =  []

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

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

		__lowercase             =  len(lowerCamelCase					) == 0
		all_relative_imports.extend(lowerCamelCase					)

	return all_relative_imports





def 				snake_case  (     lowerCamelCase					):




	'''simple docstring'''







	with open(lowerCamelCase				,  """r"""				,  encoding="""utf-8"""					) as f:
		__lowercase             =  f.read()

	# Imports of the form `import xxx`
	__lowercase             =  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
	__lowercase             =  [imp.split("""."""					)[0] for imp in imports if not imp.startswith("""."""					)]

	# Unique-ify and test we got them all
	__lowercase             =  list(set(lowerCamelCase					)					)
	__lowercase             =  []
	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 				snake_case  (     lowerCamelCase				,  lowerCamelCase					):




	'''simple docstring'''







	__lowercase             =  module_path.replace(os.path.sep				,  """."""					)
	__lowercase             =  importlib.import_module(lowerCamelCase					)

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





def 				snake_case  (     lowerCamelCase					):




	'''simple docstring'''







	from ..pipelines import DiffusionPipeline

	__lowercase             =  dict(inspect.getmembers(lowerCamelCase				,  inspect.isclass					)					)

	__lowercase             =  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}.'					)
			__lowercase             =  cls

	return pipeline_class





def 				snake_case  (     lowerCamelCase				,  lowerCamelCase				,  lowerCamelCase = None				,  lowerCamelCase = False				,  lowerCamelCase = False				,  lowerCamelCase = None				,  lowerCamelCase = None				,  lowerCamelCase = None				,  lowerCamelCase = False				,  ):




	'''simple docstring'''







	__lowercase             =  str(lowerCamelCase					)

	__lowercase             =  os.path.join(lowerCamelCase				,  lowerCamelCase					)

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

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

	# Now we move the module inside our cached dynamic modules.
	__lowercase             =  DIFFUSERS_DYNAMIC_MODULE_NAME + os.path.sep + submodule
	create_dynamic_module(lowerCamelCase					)
	__lowercase             =  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:
			__lowercase             =  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					):
			__lowercase             =  use_auth_token
		elif use_auth_token is True:
			__lowercase             =  HfFolder.get_token()
		else:
			__lowercase             =  None

		__lowercase             =  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.
		__lowercase             =  submodule_path / commit_hash
		__lowercase             =  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 				snake_case  (     lowerCamelCase				,  lowerCamelCase				,  lowerCamelCase = None				,  lowerCamelCase = None				,  lowerCamelCase = False				,  lowerCamelCase = False				,  lowerCamelCase = None				,  lowerCamelCase = None				,  lowerCamelCase = None				,  lowerCamelCase = False				,  **lowerCamelCase				,  ):




	'''simple docstring'''







	__lowercase             =  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"""				,  """"""					)					)









import random
import unittest

import torch

from diffusers import IFInpaintingPipeline
from diffusers.utils import floats_tensor
from diffusers.utils.import_utils import is_xformers_available
from diffusers.utils.testing_utils import skip_mps, torch_device

from ..pipeline_params import (
    TEXT_GUIDED_IMAGE_INPAINTING_BATCH_PARAMS,
    TEXT_GUIDED_IMAGE_INPAINTING_PARAMS,
)
from ..test_pipelines_common import PipelineTesterMixin
from . import IFPipelineTesterMixin


@skip_mps
class        __UpperCamelCase    (     _lowerCAmelCase   ,			_lowerCAmelCase   ,			unittest.TestCase   ):
	__snake_case	:Any   =       IFInpaintingPipeline
	__snake_case	:str   =       TEXT_GUIDED_IMAGE_INPAINTING_PARAMS - {'width', 'height'}
	__snake_case	:Optional[int]   =       TEXT_GUIDED_IMAGE_INPAINTING_BATCH_PARAMS
	__snake_case	:str   =       PipelineTesterMixin.required_optional_params - {'latents'}







	def  _a   (  self      :							Union[str, Any]			)		-> Tuple:
		"""simple docstring"""







		return self._get_dummy_components()







	def  _a   (  self      :							Optional[int]       ,				_lowerCAmelCase      :							List[str]       ,				_lowerCAmelCase      :							Dict=0			)		-> Any:
		"""simple docstring"""







		if str(_lowerCAmelCase			).startswith("""mps"""			):
			__lowercase             =  torch.manual_seed(_lowerCAmelCase			)
		else:
			__lowercase             =  torch.Generator(device=_lowerCAmelCase			).manual_seed(_lowerCAmelCase			)

		__lowercase             =  floats_tensor((1, 3, 32, 32)       ,				rng=random.Random(_lowerCAmelCase			)			).to(_lowerCAmelCase			)
		__lowercase             =  floats_tensor((1, 3, 32, 32)       ,				rng=random.Random(_lowerCAmelCase			)			).to(_lowerCAmelCase			)

		__lowercase             =  {
		    """prompt""": """A painting of a squirrel eating a burger""",
		    """image""": image,
		    """mask_image""": mask_image,
		    """generator""": generator,
		    """num_inference_steps""": 2,
		    """output_type""": """numpy""",
		}

		return inputs







	@unittest.skipIf(
	    torch_device != """cuda""" or not is_xformers_available()       ,				reason="""XFormers attention is only available with CUDA and `xformers` installed"""       ,				)
	def  _a   (  self      :							Tuple			)		-> Dict:
		"""simple docstring"""







		self._test_xformers_attention_forwardGenerator_pass(expected_max_diff=1e-3			)







	def  _a   (  self      :							Union[str, Any]			)		-> Dict:
		"""simple docstring"""







		self._test_save_load_optional_components()







	@unittest.skipIf(torch_device != """cuda"""       ,				reason="""float16 requires CUDA"""			)
	def  _a   (  self      :							List[str]			)		-> Tuple:
		"""simple docstring"""







		super().test_save_load_floataa(expected_max_diff=1e-1			)







	def  _a   (  self      :							Tuple			)		-> Optional[Any]:
		"""simple docstring"""







		self._test_attention_slicing_forward_pass(expected_max_diff=1e-2			)







	def  _a   (  self      :							str			)		-> Optional[int]:
		"""simple docstring"""







		self._test_save_load_local()







	def  _a   (  self      :							List[str]			)		-> List[Any]:
		"""simple docstring"""







		self._test_inference_batch_single_identical(
		    expected_max_diff=1e-2       ,				)

