


'''simple docstring'''






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


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


def  _snake_case       (					)						->   Union[str, Any]:
							lowerCAmelCase__	       =     '''https://pypi.org/pypi/diffusers/json'''
							lowerCAmelCase__	       =     json.loads(request.urlopen(A	).read()	)['''releases'''].keys()
							return sorted(A			,   key=lambda A	: version.Version(A	)	)


def  _snake_case       (					)						->   Dict:
							# 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(A	)
							os.makedirs(A			,   exist_ok=A	)
							lowerCAmelCase__	       =     Path(A	) / '''__init__.py'''
							if not init_path.exists():
														init_path.touch()


def  _snake_case       (					A	)						->   Optional[int]:
							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  _snake_case       (					A	)						->   Any:
							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  _snake_case       (					A	)						->   Optional[int]:
							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  _snake_case       (					A	)						->   Union[str, Any]:
							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  _snake_case       (					A			,   A	)						->   Tuple:
							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  _snake_case       (					A	)						->   Union[str, Any]:
							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  _snake_case       (					A			,   A			,   A = None			,   A = False			,   A = False			,   A = None			,   A = None			,   A = None			,   A = False			,   )						->   str:
							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  _snake_case       (					A			,   A			,   A = None			,   A = None			,   A = False			,   A = False			,   A = None			,   A = None			,   A = None			,   A = False			,   **A			,   )						->   str:
							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'''			,   ''''''	)	)



'''simple docstring'''






from __future__ import annotations

import numpy as np
from numpy import floataa
from numpy.typing import NDArray


def  _snake_case       (					A			,   A			,   A			,   A			,   )						->   list[float]:
							lowerCAmelCase__       , lowerCAmelCase__	       =     coefficient_matrix.shape
							lowerCAmelCase__       , lowerCAmelCase__	       =     constant_matrix.shape

							if rowsa != colsa:
														lowerCAmelCase__	       =     F"""Coefficient matrix dimensions must be nxn but received {rowsa}x{colsa}"""
														raise ValueError(A	)

							if colsa != 1:
														lowerCAmelCase__	       =     F"""Constant matrix must be nx1 but received {rowsa}x{colsa}"""
														raise ValueError(A	)

							if rowsa != rowsa:
														lowerCAmelCase__	       =     (
														    '''Coefficient and constant matrices dimensions must be nxn and nx1 but '''
														    F"""received {rowsa}x{colsa} and {rowsa}x{colsa}"""
														)
														raise ValueError(A	)

							if len(A	) != rowsa:
														lowerCAmelCase__	       =     (
														    '''Number of initial values must be equal to number of rows in coefficient '''
														    F"""matrix but received {len(A	)} and {rowsa}"""
														)
														raise ValueError(A	)

							if iterations <= 0:
														raise ValueError('''Iterations must be at least 1'''	)

							lowerCAmelCase__	       =     np.concatenate(
							    (coefficient_matrix, constant_matrix)			,   axis=1	)

							lowerCAmelCase__       , lowerCAmelCase__	       =     table.shape

							strictly_diagonally_dominant(A	)

							# Iterates the whole matrix for given number of times
							for _ in range(A	):
														lowerCAmelCase__	       =     []
														for row in range(A	):
																					lowerCAmelCase__	       =     0
																					for col in range(A	):
																												if col == row:
																																			lowerCAmelCase__	       =     table[row][col]
																												elif col == cols - 1:
																																			lowerCAmelCase__	       =     table[row][col]
																												else:
																																			temp += (-1) * table[row][col] * init_val[col]
																					lowerCAmelCase__	       =     (temp + val) / denom
																					new_val.append(A	)
														lowerCAmelCase__	       =     new_val

							return [float(A	) for i in new_val]


def  _snake_case       (					A	)						->   bool:
							lowerCAmelCase__       , lowerCAmelCase__	       =     table.shape

							lowerCAmelCase__	       =     True

							for i in range(0			,   A	):
														lowerCAmelCase__	       =     0
														for j in range(0			,   cols - 1	):
																					if i == j:
																												continue
																					else:
																												total += table[i][j]

														if table[i][i] <= total:
																					raise ValueError('''Coefficient matrix is not strictly diagonally dominant'''	)

							return is_diagonally_dominant


# Test Cases
if __name__ == "__main__":
		import doctest

		doctest.testmod()