






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 : List[Any]						   =       (
    '''https://raw.githubusercontent.com/huggingface/diffusers/{revision}/examples/community/{pipeline}.py'''
)


lowerCamelCase : Any						   =       logging.get_logger(__name__)  # pylint: disable=invalid-name
def       snake_case_		(     ):
							__lowercase      :							List[Any]				=     """https://pypi.org/pypi/diffusers/json"""
							__lowercase      :							Tuple				=     json.loads(request.urlopen(lowerCAmelCase_      ).read()      )["""releases"""].keys()
							return sorted(lowerCAmelCase_       ,       key=lambda lowerCAmelCase_      : version.Version(lowerCAmelCase_      )      )
def       snake_case_		(     ):
							# This function has already been executed if HF_MODULES_CACHE already is in the Python path.
							if HF_MODULES_CACHE in sys.path:
														return

							sys.path.append(lowerCAmelCase_      )
							os.makedirs(lowerCAmelCase_       ,       exist_ok=lowerCAmelCase_      )
							__lowercase      :							Any				=     Path(lowerCAmelCase_      ) / """__init__.py"""
							if not init_path.exists():
														init_path.touch()
def       snake_case_		(     lowerCAmelCase_       :						Union[str, os.PathLike]      ):
							init_hf_modules()
							__lowercase      :							List[str]				=     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      :							Union[str, Any]				=     dynamic_module_path / """__init__.py"""
							if not init_path.exists():
														init_path.touch()
def       snake_case_		(     lowerCAmelCase_       :						Union[str, Any]      ):
							with open(lowerCAmelCase_       ,       """r"""       ,       encoding="""utf-8"""      ) as f:
														__lowercase      :							List[Any]				=     f.read()

							# Imports of the form `import .xxx`
							__lowercase      :							Tuple				=     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_       :						List[Any]      ):
							__lowercase      :							List[str]				=     False
							__lowercase      :							Union[str, Any]				=     [module_file]
							__lowercase      :							int				=     []

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

														__lowercase      :							List[Any]				=     Path(lowerCAmelCase_      ).parent
														__lowercase      :							Tuple				=     [str(module_path / m      ) for m in new_imports]
														__lowercase      :							Union[str, Any]				=     [f for f in new_import_files if f not in all_relative_imports]
														__lowercase      :							str				=     [F"{f}.py" for f in new_import_files]

														__lowercase      :							Union[str, Any]				=     len(lowerCAmelCase_      ) == 0
														all_relative_imports.extend(lowerCAmelCase_      )

							return all_relative_imports
def       snake_case_		(     lowerCAmelCase_       :						List[Any]      ):
							with open(lowerCAmelCase_       ,       """r"""       ,       encoding="""utf-8"""      ) as f:
														__lowercase      :							List[Any]				=     f.read()

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

							# Unique-ify and test we got them all
							__lowercase      :							Optional[int]				=     list(set(lowerCAmelCase_      )      )
							__lowercase      :							Tuple				=     []
							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_       :						Optional[Any]       ,       lowerCAmelCase_       :						List[Any]      ):
							__lowercase      :							str				=     module_path.replace(os.path.sep       ,       """."""      )
							__lowercase      :							Optional[Any]				=     importlib.import_module(lowerCAmelCase_      )

							if class_name is None:
														return find_pipeline_class(lowerCAmelCase_      )
							return getattr(lowerCAmelCase_       ,       lowerCAmelCase_      )
def       snake_case_		(     lowerCAmelCase_       :						Tuple      ):
							from ..pipelines import DiffusionPipeline

							__lowercase      :							Optional[int]				=     dict(inspect.getmembers(lowerCAmelCase_       ,       inspect.isclass      )      )

							__lowercase      :							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}."      )
																					__lowercase      :							Tuple				=     cls

							return pipeline_class
def       snake_case_		(     lowerCAmelCase_       :						Union[str, os.PathLike]       ,       lowerCAmelCase_       :						str       ,       lowerCAmelCase_       :						Optional[Union[str, os.PathLike]] = None       ,       lowerCAmelCase_       :						bool = False       ,       lowerCAmelCase_       :						bool = False       ,       lowerCAmelCase_       :						Optional[Dict[str, str]] = None       ,       lowerCAmelCase_       :						Optional[Union[bool, str]] = None       ,       lowerCAmelCase_       :						Optional[str] = None       ,       lowerCAmelCase_       :						bool = False       ,       ):
							__lowercase      :							Dict				=     str(lowerCAmelCase_      )

							__lowercase      :							Any				=     os.path.join(lowerCAmelCase_       ,       lowerCAmelCase_      )

							if os.path.isfile(lowerCAmelCase_      ):
														__lowercase      :							Dict				=     module_file_or_url
														__lowercase      :							Union[str, Any]				=     """local"""
							elif pretrained_model_name_or_path.count("""/"""      ) == 0:
														__lowercase      :							List[str]				=     get_diffusers_versions()
														# cut ".dev0"
														__lowercase      :							List[str]				=     """v""" + """.""".join(__version__.split("""."""      )[:3]      )

														# retrieve github version that matches
														if revision is None:
																					__lowercase      :							Optional[int]				=     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      :							List[Any]				=     F"v{revision}"
														elif revision == "main":
																					__lowercase      :							Tuple				=     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      :							Optional[Any]				=     COMMUNITY_PIPELINES_URL.format(revision=lowerCAmelCase_       ,       pipeline=lowerCAmelCase_      )
														try:
																					__lowercase      :							Optional[int]				=     cached_download(
																					    lowerCAmelCase_       ,       cache_dir=lowerCAmelCase_       ,       force_download=lowerCAmelCase_       ,       proxies=lowerCAmelCase_       ,       resume_download=lowerCAmelCase_       ,       local_files_only=lowerCAmelCase_       ,       use_auth_token=lowerCAmelCase_       ,       )
																					__lowercase      :							Any				=     """git"""
																					__lowercase      :							Tuple				=     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      :							Optional[int]				=     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      :							List[str]				=     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      :							str				=     check_imports(lowerCAmelCase_      )

							# Now we move the module inside our cached dynamic modules.
							__lowercase      :							Tuple				=     DIFFUSERS_DYNAMIC_MODULE_NAME + os.path.sep + submodule
							create_dynamic_module(lowerCAmelCase_      )
							__lowercase      :							List[Any]				=     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      :							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_      ):
																					__lowercase      :							int				=     use_auth_token
														elif use_auth_token is True:
																					__lowercase      :							Optional[int]				=     HfFolder.get_token()
														else:
																					__lowercase      :							Union[str, Any]				=     None

														__lowercase      :							Optional[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.
														__lowercase      :							Union[str, Any]				=     submodule_path / commit_hash
														__lowercase      :							int				=     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_       :						Union[str, os.PathLike]       ,       lowerCAmelCase_       :						str       ,       lowerCAmelCase_       :						Optional[str] = None       ,       lowerCAmelCase_       :						Optional[Union[str, os.PathLike]] = None       ,       lowerCAmelCase_       :						bool = False       ,       lowerCAmelCase_       :						bool = False       ,       lowerCAmelCase_       :						Optional[Dict[str, str]] = None       ,       lowerCAmelCase_       :						Optional[Union[bool, str]] = None       ,       lowerCAmelCase_       :						Optional[str] = None       ,       lowerCAmelCase_       :						bool = False       ,       **lowerCAmelCase_       :						str       ,       ):
							__lowercase      :							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"""       ,       """"""      )      )







import warnings

from ...utils import logging
from .image_processing_beit import BeitImageProcessor


lowerCamelCase : Union[str, Any]						   =       logging.get_logger(__name__)

class 		lowerCAmelCase     ( __a			):

					'''simple docstring'''

					def __init__(  self				:	int		,     *__a				:	Dict		,     **__a				:	Optional[Any]							)			->  None:


												"""simple docstring"""

												warnings.warn(
												    """The class BeitFeatureExtractor is deprecated and will be removed in version 5 of Transformers. Please"""
												    """ use BeitImageProcessor instead."""		,     __a		,     )
												super().__init__(*__a		,     **__a							)