

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    __UpperCAmelCase (	)			->     Tuple:

					"""simple docstring"""

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



def    __UpperCAmelCase (	)			->     Tuple:

					"""simple docstring"""

					if HF_MODULES_CACHE in sys.path:
										return

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



def    __UpperCAmelCase (	__a      :			Union[str, os.PathLike]	)			->     List[Any]:

					"""simple docstring"""

					init_hf_modules()
					_a	:    Any         =		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	)
					_a	:    Optional[int]         =		dynamic_module_path / '''__init__.py'''
					if not init_path.exists():
										init_path.touch()



def    __UpperCAmelCase (	__a      :			Optional[Any]	)			->     List[Any]:

					"""simple docstring"""

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

					# Imports of the form `import .xxx`
					_a	:    List[Any]         =		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    __UpperCAmelCase (	__a      :			List[str]	)			->     str:

					"""simple docstring"""

					_a	:    Tuple         =		False
					_a	:    Union[str, Any]         =		[module_file]
					_a	:    Optional[int]         =		[]

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

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

										_a	:    str         =		len(__a	) == 0
										all_relative_imports.extend(__a	)

					return all_relative_imports



def    __UpperCAmelCase (	__a      :			Optional[int]	)			->     Tuple:

					"""simple docstring"""

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

					# Imports of the form `import xxx`
					_a	:    Optional[int]         =		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
					_a	:    Tuple         =		[imp.split('''.'''	)[0] for imp in imports if not imp.startswith('''.'''	)]

					# Unique-ify and test we got them all
					_a	:    str         =		list(set(__a	)	)
					_a	:    List[Any]         =		[]
					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    __UpperCAmelCase (	__a      :			Optional[int]		,__a      :			Tuple	)			->     int:

					"""simple docstring"""

					_a	:    int         =		module_path.replace(os.path.sep		,'''.'''	)
					_a	:    Dict         =		importlib.import_module(__a	)

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



def    __UpperCAmelCase (	__a      :			Optional[Any]	)			->     Optional[int]:

					"""simple docstring"""

					from ..pipelines import DiffusionPipeline

					_a	:    Tuple         =		dict(inspect.getmembers(__a		,inspect.isclass	)	)

					_a	:    Union[str, Any]         =		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}."""	)
															_a	:    int         =		cls

					return pipeline_class



def    __UpperCAmelCase (	__a      :			Union[str, os.PathLike]		,__a      :			str		,__a      :			Optional[Union[str, os.PathLike]] = None		,__a      :			bool = False		,__a      :			bool = False		,__a      :			Optional[Dict[str, str]] = None		,__a      :			Optional[Union[bool, str]] = None		,__a      :			Optional[str] = None		,__a      :			bool = False		,)			->     List[Any]:

					"""simple docstring"""

					_a	:    str         =		str(__a	)

					_a	:    Any         =		os.path.join(__a		,__a	)

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

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

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

										_a	:    Dict         =		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.
										_a	:    Optional[Any]         =		submodule_path / commit_hash
										_a	:    Any         =		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    __UpperCAmelCase (	__a      :			Union[str, os.PathLike]		,__a      :			str		,__a      :			Optional[str] = None		,__a      :			Optional[Union[str, os.PathLike]] = None		,__a      :			bool = False		,__a      :			bool = False		,__a      :			Optional[Dict[str, str]] = None		,__a      :			Optional[Union[bool, str]] = None		,__a      :			Optional[str] = None		,__a      :			bool = False		,**__a      :			Optional[int]		,)			->     str:

					"""simple docstring"""

					_a	:    Any         =		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'''		,''''''	)	)






from ...configuration_utils import PretrainedConfig
from ...utils import logging
from ...utils.backbone_utils import BackboneConfigMixin, get_aligned_output_features_output_indices


a__				=		logging.get_logger(__name__)

a__				=		{
    '''google/bit-50''': '''https://huggingface.co/google/bit-50/resolve/main/config.json''',
}

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







							UpperCAmelCase__       :						Any	    =    "bit"
							UpperCAmelCase__       :						Optional[int]	    =    ["preactivation", "bottleneck"]
							UpperCAmelCase__       :						Optional[Any]	    =    ["SAME", "VALID"]


							def __init__(			self       ,							_a=3       ,							_a=6_4       ,							_a=[2_5_6, 5_1_2, 1_0_2_4, 2_0_4_8]       ,							_a=[3, 4, 6, 3]       ,							_a="preactivation"       ,							_a="relu"       ,							_a=None       ,							_a=3_2       ,							_a=0.0       ,							_a=False       ,							_a=3_2       ,							_a=1       ,							_a=None       ,							_a=None       ,							**_a       ,							)    ->    Union[str, Any]:
												super().__init__(**_a       )
												if layer_type not in self.layer_types:
																	raise ValueError(F"""layer_type={layer_type} is not one of {','.join(self.layer_types       )}"""       )
												if global_padding is not None:
																	if global_padding.upper() in self.supported_padding:
																						_a	:    Any         =		global_padding.upper()
																	else:
																						raise ValueError(F"""Padding strategy {global_padding} not supported"""       )
												_a	:    Optional[int]         =		num_channels
												_a	:    List[Any]         =		embedding_size
												_a	:    Any         =		hidden_sizes
												_a	:    int         =		depths
												_a	:    Dict         =		layer_type
												_a	:    int         =		hidden_act
												_a	:    Optional[Any]         =		global_padding
												_a	:    Optional[Any]         =		num_groups
												_a	:    Union[str, Any]         =		drop_path_rate
												_a	:    Tuple         =		embedding_dynamic_padding
												_a	:    Union[str, Any]         =		output_stride
												_a	:    Any         =		width_factor

												_a	:    Any         =		['''stem'''] + [F"""stage{idx}""" for idx in range(1       ,							len(_a       ) + 1       )]
												_a      ,					_a	:    List[str]         =		get_aligned_output_features_output_indices(
												    out_features=_a       ,							out_indices=_a       ,							stage_names=self.stage_names       )



