

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					(      )     ->	Union[str, Any]:

					"""simple docstring"""

					_a      :       Any   =						'''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					(      )     ->	List[str]:

					"""simple docstring"""

					if HF_MODULES_CACHE in sys.path:
										return

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



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

					"""simple docstring"""

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



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

					"""simple docstring"""

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

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

					"""simple docstring"""

					_a      :       Optional[int]   =						False
					_a      :       str   =						[module_file]
					_a      :       Dict   =						[]

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

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

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

					return all_relative_imports



def   __UpperCAmelCase					(      __a :     str      )     ->	Any:

					"""simple docstring"""

					with open(__a						,'''r'''						,encoding='''utf-8'''      ) as f:
										_a      :       List[str]   =						f.read()

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

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

					"""simple docstring"""

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

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



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

					"""simple docstring"""

					from ..pipelines import DiffusionPipeline

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

					_a      :       Optional[int]   =						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      :       Optional[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						,)     ->	Optional[Any]:

					"""simple docstring"""

					_a      :       List[Any]   =						str(__a      )

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

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

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

					# Now we move the module inside our cached dynamic modules.
					_a      :       int   =						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      :       Any   =						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      :       List[str]   =						use_auth_token
										elif use_auth_token is True:
															_a      :       Optional[int]   =						HfFolder.get_token()
										else:
															_a      :       str   =						None

										_a      :       List[Any]   =						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      :       List[str]   =						submodule_path / commit_hash
										_a      :       Tuple   =						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 :     Tuple						,)     ->	int:

					"""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'''						,''''''      )      )






import datetime
import platform
import subprocess
from typing import Optional, Tuple, Union

import numpy as np



def   __UpperCAmelCase					(      __a :     bytes						,__a :     int      )     ->	np.array:

					"""simple docstring"""

					_a      :       int   =						F"""{sampling_rate}"""
					_a      :       str   =						'''1'''
					_a      :       Optional[int]   =						'''f32le'''
					_a      :       Optional[Any]   =						[
					    '''ffmpeg''',
					    '''-i''',
					    '''pipe:0''',
					    '''-ac''',
					    ac,
					    '''-ar''',
					    ar,
					    '''-f''',
					    format_for_conversion,
					    '''-hide_banner''',
					    '''-loglevel''',
					    '''quiet''',
					    '''pipe:1''',
					]

					try:
										with subprocess.Popen(__a						,stdin=subprocess.PIPE						,stdout=subprocess.PIPE      ) as ffmpeg_process:
															_a      :       Any   =						ffmpeg_process.communicate(__a      )
					except FileNotFoundError as error:
										raise ValueError('''ffmpeg was not found but is required to load audio files from filename'''      ) from error
					_a      :       Optional[Any]   =						output_stream[0]
					_a      :       Optional[int]   =						np.frombuffer(__a						,np.floataa      )
					if audio.shape[0] == 0:
										raise ValueError('''Malformed soundfile'''      )
					return audio



def   __UpperCAmelCase					(      __a :     int						,__a :     float						,__a :     str = "f32le"						,)     ->	str:

					"""simple docstring"""

					_a      :       Dict   =						F"""{sampling_rate}"""
					_a      :       Optional[Any]   =						'''1'''
					if format_for_conversion == "s16le":
										_a      :       Dict   =						2
					elif format_for_conversion == "f32le":
										_a      :       Optional[Any]   =						4
					else:
										raise ValueError(F"""Unhandled format `{format_for_conversion}`. Please use `s16le` or `f32le`"""      )

					_a      :       Dict   =						platform.system()
					if system == "Linux":
										_a      :       Dict   =						'''alsa'''
										_a      :       Union[str, Any]   =						'''default'''
					elif system == "Darwin":
										_a      :       Union[str, Any]   =						'''avfoundation'''
										_a      :       List[str]   =						''':0'''
					elif system == "Windows":
										_a      :       Optional[int]   =						'''dshow'''
										_a      :       str   =						'''default'''

					_a      :       Tuple   =						[
					    '''ffmpeg''',
					    '''-f''',
					    format_,
					    '''-i''',
					    input_,
					    '''-ac''',
					    ac,
					    '''-ar''',
					    ar,
					    '''-f''',
					    format_for_conversion,
					    '''-fflags''',
					    '''nobuffer''',
					    '''-hide_banner''',
					    '''-loglevel''',
					    '''quiet''',
					    '''pipe:1''',
					]
					_a      :       Any   =						int(round(sampling_rate * chunk_length_s      )      ) * size_of_sample
					_a      :       str   =						_ffmpeg_stream(__a						,__a      )
					for item in iterator:
										yield item



def   __UpperCAmelCase					(      __a :     int						,__a :     float						,__a :     Optional[int] = None						,__a :     Optional[Union[Tuple[float, float], float]] = None						,__a :     str = "f32le"						,)     ->	Optional[int]:

					"""simple docstring"""

					if stream_chunk_s is not None:
										_a      :       Tuple   =						stream_chunk_s
					else:
										_a      :       Tuple   =						chunk_length_s

					_a      :       Tuple   =						ffmpeg_microphone(__a						,__a						,format_for_conversion=__a      )
					if format_for_conversion == "s16le":
										_a      :       Any   =						np.intaa
										_a      :       Optional[int]   =						2
					elif format_for_conversion == "f32le":
										_a      :       Dict   =						np.floataa
										_a      :       List[Any]   =						4
					else:
										raise ValueError(F"""Unhandled format `{format_for_conversion}`. Please use `s16le` or `f32le`"""      )

					if stride_length_s is None:
										_a      :       List[Any]   =						chunk_length_s / 6
					_a      :       Optional[int]   =						int(round(sampling_rate * chunk_length_s      )      ) * size_of_sample
					if isinstance(__a						,(int, float)      ):
										_a      :       Optional[Any]   =						[stride_length_s, stride_length_s]

					_a      :       Optional[Any]   =						int(round(sampling_rate * stride_length_s[0]      )      ) * size_of_sample
					_a      :       str   =						int(round(sampling_rate * stride_length_s[1]      )      ) * size_of_sample
					_a      :       Optional[Any]   =						datetime.datetime.now()
					_a      :       Tuple   =						datetime.timedelta(seconds=__a      )
					for item in chunk_bytes_iter(__a						,__a						,stride=(stride_left, stride_right)						,stream=__a      ):
										# Put everything back in numpy scale
										_a      :       Dict   =						np.frombuffer(item['''raw''']						,dtype=__a      )
										_a      :       Dict   =						(
										    item['''stride'''][0] // size_of_sample,
										    item['''stride'''][1] // size_of_sample,
										)
										_a      :       str   =						sampling_rate
										audio_time += delta
										if datetime.datetime.now() > audio_time + 10 * delta:
															# We're late !! SKIP
															continue
										yield item



def   __UpperCAmelCase					(      __a :     Optional[int]						,__a :     int						,__a :     Tuple[int, int]						,__a :     bool = False      )     ->	Optional[int]:

					"""simple docstring"""

					_a      :       Any   =						b''''''
					_a ,   _a      :       List[str]   =						stride
					if stride_left + stride_right >= chunk_len:
										raise ValueError(
										    F"""Stride needs to be strictly smaller than chunk_len: ({stride_left}, {stride_right}) vs {chunk_len}"""      )
					_a      :       List[str]   =						0
					for raw in iterator:
										acc += raw
										if stream and len(__a      ) < chunk_len:
															_a      :       Dict   =						(_stride_left, 0)
															yield {"raw": acc[:chunk_len], "stride": stride, "partial": True}
										else:
															while len(__a      ) >= chunk_len:
																				# We are flushing the accumulator
																				_a      :       List[str]   =						(_stride_left, stride_right)
																				_a      :       List[Any]   =						{'''raw''': acc[:chunk_len], '''stride''': stride}
																				if stream:
																									_a      :       List[Any]   =						False
																				yield item
																				_a      :       Optional[Any]   =						stride_left
																				_a      :       Optional[Any]   =						acc[chunk_len - stride_left - stride_right :]
    # Last chunk
					if len(__a      ) > stride_left:
										_a      :       Optional[Any]   =						{'''raw''': acc, '''stride''': (_stride_left, 0)}
										if stream:
															_a      :       Dict   =						False
										yield item







def   __UpperCAmelCase					(      __a :     int						,__a :     int      )     ->	Tuple:

					"""simple docstring"""

					_a      :       Dict   =						2**24  # 16Mo
					try:
										with subprocess.Popen(__a						,stdout=subprocess.PIPE						,bufsize=__a      ) as ffmpeg_process:
															while True:
																				_a      :       int   =						ffmpeg_process.stdout.read(__a      )
																				if raw == b"":
																									break
																				yield raw
					except FileNotFoundError as error:
										raise ValueError('''ffmpeg was not found but is required to stream audio files from filename'''      ) from error



