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


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


def 			UpperCamelCase					(    ):
       snake_case       :   Optional[Any]							=      "https://pypi.org/pypi/diffusers/json"
       snake_case       :   Optional[int]							=      json.loads(request.urlopen(UpperCamelCase_    ).read()    )["releases"].keys()
       return sorted(UpperCamelCase_				,					key=lambda __lowerCamelCase    : version.Version(UpperCamelCase_    )    )


def 			UpperCamelCase					(    ):


       if HF_MODULES_CACHE in sys.path:
              return

       sys.path.append(UpperCamelCase_    )
       os.makedirs(UpperCamelCase_				,					exist_ok=UpperCamelCase_    )
       snake_case       :   Union[str, Any]							=      Path(UpperCamelCase_    ) / "__init__.py"
       if not init_path.exists():
              init_path.touch()


def 			UpperCamelCase					(    __lowerCamelCase     : Union[str, os.PathLike]    ):
       init_hf_modules()
       snake_case       :   List[Any]							=      Path(UpperCamelCase_    ) / 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(UpperCamelCase_				,					exist_ok=UpperCamelCase_    )
       snake_case       :   Any							=      dynamic_module_path / "__init__.py"
       if not init_path.exists():
              init_path.touch()


def 			UpperCamelCase					(    __lowerCamelCase     : Dict    ):


       with open(UpperCamelCase_				,					"r"				,					encoding="utf-8"    ) as f:
              snake_case       :   Union[str, Any]							=      f.read()

       # Imports of the form `import .xxx`
       snake_case       :   Union[str, Any]							=      re.findall("^\s*import\s+\.(\S+)\s*$"				,					UpperCamelCase_				,					flags=re.MULTILINE    )
       # Imports of the form `from .xxx import yyy`
       relative_imports += re.findall("^\s*from\s+\.(\S+)\s+import"				,					UpperCamelCase_				,					flags=re.MULTILINE    )
       # Unique-ify
       return list(set(UpperCamelCase_    )    )


def 			UpperCamelCase					(    __lowerCamelCase     : List[str]    ):
       snake_case       :   List[Any]							=      False
       snake_case       :   str							=      [module_file]
       snake_case       :   str							=      []

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

              snake_case       :   str							=      Path(UpperCamelCase_    ).parent
              snake_case       :   List[str]							=      [str(module_path / m    ) for m in new_imports]
              snake_case       :   int							=      [f for f in new_import_files if f not in all_relative_imports]
              snake_case       :   Dict							=      [f"""{f}.py""" for f in new_import_files]

              snake_case       :   int							=      len(UpperCamelCase_    ) == 0
              all_relative_imports.extend(UpperCamelCase_    )

       return all_relative_imports


def 			UpperCamelCase					(    __lowerCamelCase     : Optional[int]    ):


       with open(UpperCamelCase_				,					"r"				,					encoding="utf-8"    ) as f:
              snake_case       :   List[Any]							=      f.read()

       # Imports of the form `import xxx`
       snake_case       :   Any							=      re.findall("^\s*import\s+(\S+)\s*$"				,					UpperCamelCase_				,					flags=re.MULTILINE    )
       # Imports of the form `from xxx import yyy`
       imports += re.findall("^\s*from\s+(\S+)\s+import"				,					UpperCamelCase_				,					flags=re.MULTILINE    )
       # Only keep the top-level module
       snake_case       :   Tuple							=      [imp.split("."    )[0] for imp in imports if not imp.startswith("."    )]

       # Unique-ify and test we got them all
       snake_case       :   Optional[Any]							=      list(set(UpperCamelCase_    )    )
       snake_case       :   Any							=      []
       for imp in imports:
              try:
                     importlib.import_module(UpperCamelCase_    )
              except ImportError:
                     missing_packages.append(UpperCamelCase_    )

       if len(UpperCamelCase_    ) > 0:
              raise ImportError(
                  "This modeling file requires the following packages that were not found in your environment: "
                  f"""{', '.join(UpperCamelCase_    )}. Run `pip install {' '.join(UpperCamelCase_    )}`"""    )

       return get_relative_imports(UpperCamelCase_    )


def 			UpperCamelCase					(    __lowerCamelCase     : Optional[int]				,					__lowerCamelCase     : int    ):
       snake_case       :   Tuple							=      module_path.replace(os.path.sep				,					"."    )
       snake_case       :   int							=      importlib.import_module(UpperCamelCase_    )

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


def 			UpperCamelCase					(    __lowerCamelCase     : int    ):
       from ..pipelines import DiffusionPipeline

       snake_case       :   Tuple							=      dict(inspect.getmembers(UpperCamelCase_				,					inspect.isclass    )    )

       snake_case       :   List[str]							=      None
       for cls_name, cls in cls_members.items():
              if (
                  cls_name != DiffusionPipeline.__name__
                  and issubclass(cls				,					UpperCamelCase_    )
                  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}."""    )
                     snake_case       :   List[str]							=      cls

       return pipeline_class


def 			UpperCamelCase					(    __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				,					):
       snake_case       :   Optional[int]							=      str(UpperCamelCase_    )

       snake_case       :   Any							=      os.path.join(UpperCamelCase_				,					UpperCamelCase_    )

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

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

       # Now we move the module inside our cached dynamic modules.
       snake_case       :   str							=      DIFFUSERS_DYNAMIC_MODULE_NAME + os.path.sep + submodule
       create_dynamic_module(UpperCamelCase_    )
       snake_case       :   Any							=      Path(UpperCamelCase_    ) / 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(UpperCamelCase_				,					submodule_path / module_file    )
              for module_needed in modules_needed:
                     snake_case       :   Optional[int]							=      f"""{module_needed}.py"""
                     shutil.copy(os.path.join(UpperCamelCase_				,					UpperCamelCase_    )				,					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(UpperCamelCase_				,					UpperCamelCase_    ):
                     snake_case       :   List[str]							=      use_auth_token
              elif use_auth_token is True:
                     snake_case       :   Union[str, Any]							=      HfFolder.get_token()
              else:
                     snake_case       :   List[Any]							=      None

              snake_case       :   Optional[Any]							=      model_info(UpperCamelCase_				,					revision=UpperCamelCase_				,					token=UpperCamelCase_    ).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.
              snake_case       :   int							=      submodule_path / commit_hash
              snake_case       :   Optional[Any]							=      full_submodule + os.path.sep + commit_hash
              create_dynamic_module(UpperCamelCase_    )

              if not (submodule_path / module_file).exists():
                     shutil.copy(UpperCamelCase_				,					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(
                                UpperCamelCase_				,					f"""{module_needed}.py"""				,					cache_dir=UpperCamelCase_				,					force_download=UpperCamelCase_				,					resume_download=UpperCamelCase_				,					proxies=UpperCamelCase_				,					use_auth_token=UpperCamelCase_				,					revision=UpperCamelCase_				,					local_files_only=UpperCamelCase_				,					)
       return os.path.join(UpperCamelCase_				,					UpperCamelCase_    )




def 			UpperCamelCase					(    __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     : List[str]				,					):
       snake_case       :   List[str]							=      get_cached_module_file(
           UpperCamelCase_				,					UpperCamelCase_				,					cache_dir=UpperCamelCase_				,					force_download=UpperCamelCase_				,					resume_download=UpperCamelCase_				,					proxies=UpperCamelCase_				,					use_auth_token=UpperCamelCase_				,					revision=UpperCamelCase_				,					local_files_only=UpperCamelCase_				,					)
       return get_class_in_module(UpperCamelCase_				,					final_module.replace(".py"				,					""    )    )








"""simple docstring"""





# Copyright 2023 The HuggingFace Inc. team. All rights reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
from ..models.whisper import WhisperForConditionalGeneration, WhisperProcessor
from .base import PipelineTool




class 					_lowerCAmelCase  (	lowercase   ):







 """simple docstring"""







 __UpperCAmelCase	:					Tuple   		=	"openai/whisper-base"
 __UpperCAmelCase	:					Union[str, Any]   		=	(
     "This is a tool that transcribes an audio into text. It takes an input named `audio` and returns the "
     "transcribed text."
 )
 __UpperCAmelCase	:					List[str]   		=	"transcriber"
 __UpperCAmelCase	:					Optional[Any]   		=	WhisperProcessor
 __UpperCAmelCase	:					str   		=	WhisperForConditionalGeneration

 __UpperCAmelCase	:					List[str]   		=	["audio"]
 __UpperCAmelCase	:					Tuple   		=	["text"]


 def 					_lowercase (			self							:			str,					UpperCAmelCase__							:			int	):
    return self.pre_processor(UpperCAmelCase__,					return_tensors="pt"	).input_features


 def 					_lowercase (			self							:			Union[str, Any],					UpperCAmelCase__							:			Optional[Any]	):
    return self.model.generate(inputs=UpperCAmelCase__	)



 def 					_lowercase (			self							:			Dict,					UpperCAmelCase__							:			Optional[int]	):
    return self.pre_processor.batch_decode(UpperCAmelCase__,					skip_special_tokens=UpperCAmelCase__	)[0]




