

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


_SCREAMING_SNAKE_CASE  :						int         =    (
    '''https://raw.githubusercontent.com/huggingface/diffusers/{revision}/examples/community/{pipeline}.py'''
)


_SCREAMING_SNAKE_CASE  :						Dict         =    logging.get_logger(__name__)  # pylint: disable=invalid-name
def        UpperCAmelCase_     (		):





     '''simple docstring'''




     SCREAMING_SNAKE_CASE__					=					'''https://pypi.org/pypi/diffusers/json'''
     SCREAMING_SNAKE_CASE__					=					json.loads(request.urlopen(_A       ).read()       )['''releases'''].keys()
     return sorted(_A   ,	key=lambda _A       : version.Version(_A       )       )
def        UpperCAmelCase_     (		):





     '''simple docstring'''




     if HF_MODULES_CACHE in sys.path:
          return

     sys.path.append(_A       )
     os.makedirs(_A   ,	exist_ok=_A       )
     SCREAMING_SNAKE_CASE__					=					Path(_A       ) / '''__init__.py'''
     if not init_path.exists():
          init_path.touch()
def        UpperCAmelCase_     (		_A       ):





     '''simple docstring'''




     init_hf_modules()
     SCREAMING_SNAKE_CASE__					=					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       )
     SCREAMING_SNAKE_CASE__					=					dynamic_module_path / '''__init__.py'''
     if not init_path.exists():
          init_path.touch()
def        UpperCAmelCase_     (		_A       ):





     '''simple docstring'''




     with open(_A   ,	'''r'''   ,	encoding='''utf-8'''       ) as f:
          SCREAMING_SNAKE_CASE__					=					f.read()

     # Imports of the form `import .xxx`
     SCREAMING_SNAKE_CASE__					=					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       ):





     '''simple docstring'''




     SCREAMING_SNAKE_CASE__					=					False
     SCREAMING_SNAKE_CASE__					=					[module_file]
     SCREAMING_SNAKE_CASE__					=					[]

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

          SCREAMING_SNAKE_CASE__					=					Path(_A       ).parent
          SCREAMING_SNAKE_CASE__					=					[str(module_path / m       ) for m in new_imports]
          SCREAMING_SNAKE_CASE__					=					[f for f in new_import_files if f not in all_relative_imports]
          SCREAMING_SNAKE_CASE__					=					[F'''{f}.py''' for f in new_import_files]

          SCREAMING_SNAKE_CASE__					=					len(_A       ) == 0
          all_relative_imports.extend(_A       )

     return all_relative_imports
def        UpperCAmelCase_     (		_A       ):





     '''simple docstring'''




     with open(_A   ,	'''r'''   ,	encoding='''utf-8'''       ) as f:
          SCREAMING_SNAKE_CASE__					=					f.read()

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

     # Unique-ify and test we got them all
     SCREAMING_SNAKE_CASE__					=					list(set(_A       )       )
     SCREAMING_SNAKE_CASE__					=					[]
     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   ,	_A       ):





     '''simple docstring'''




     SCREAMING_SNAKE_CASE__					=					module_path.replace(os.path.sep   ,	'''.'''       )
     SCREAMING_SNAKE_CASE__					=					importlib.import_module(_A       )

     if class_name is None:
          return find_pipeline_class(_A       )
     return getattr(_A   ,	_A       )
def        UpperCAmelCase_     (		_A       ):





     '''simple docstring'''




     from ..pipelines import DiffusionPipeline

     SCREAMING_SNAKE_CASE__					=					dict(inspect.getmembers(_A   ,	inspect.isclass       )       )

     SCREAMING_SNAKE_CASE__					=					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}.'''       )
               SCREAMING_SNAKE_CASE__					=					cls

     return pipeline_class
def        UpperCAmelCase_     (		_A   ,	_A   ,	_A = None   ,	_A = False   ,	_A = False   ,	_A = None   ,	_A = None   ,	_A = None   ,	_A = False   ,	):





     '''simple docstring'''




     SCREAMING_SNAKE_CASE__					=					str(_A       )

     SCREAMING_SNAKE_CASE__					=					os.path.join(_A   ,	_A       )

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

          # retrieve github version that matches
          if revision is None:
               SCREAMING_SNAKE_CASE__					=					latest_version if latest_version[1:] in available_versions else '''main'''
               logger.info(F'''Defaulting to latest_version: {revision}.'''       )
          elif revision in available_versions:
               SCREAMING_SNAKE_CASE__					=					F'''v{revision}'''
          elif revision == "main":
               SCREAMING_SNAKE_CASE__					=					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
          SCREAMING_SNAKE_CASE__					=					COMMUNITY_PIPELINES_URL.format(revision=_A   ,	pipeline=_A       )
          try:
               SCREAMING_SNAKE_CASE__					=					cached_download(
                   _A   ,	cache_dir=_A   ,	force_download=_A   ,	proxies=_A   ,	resume_download=_A   ,	local_files_only=_A   ,	use_auth_token=_A   ,	)
               SCREAMING_SNAKE_CASE__					=					'''git'''
               SCREAMING_SNAKE_CASE__					=					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
               SCREAMING_SNAKE_CASE__					=					hf_hub_download(
                   _A   ,	_A   ,	cache_dir=_A   ,	force_download=_A   ,	proxies=_A   ,	resume_download=_A   ,	local_files_only=_A   ,	use_auth_token=_A   ,	)
               SCREAMING_SNAKE_CASE__					=					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
     SCREAMING_SNAKE_CASE__					=					check_imports(_A       )

     # Now we move the module inside our cached dynamic modules.
     SCREAMING_SNAKE_CASE__					=					DIFFUSERS_DYNAMIC_MODULE_NAME + os.path.sep + submodule
     create_dynamic_module(_A       )
     SCREAMING_SNAKE_CASE__					=					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:
               SCREAMING_SNAKE_CASE__					=					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       ):
               SCREAMING_SNAKE_CASE__					=					use_auth_token
          elif use_auth_token is True:
               SCREAMING_SNAKE_CASE__					=					HfFolder.get_token()
          else:
               SCREAMING_SNAKE_CASE__					=					None

          SCREAMING_SNAKE_CASE__					=					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.
          SCREAMING_SNAKE_CASE__					=					submodule_path / commit_hash
          SCREAMING_SNAKE_CASE__					=					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   ,	_A   ,	_A = None   ,	_A = None   ,	_A = False   ,	_A = False   ,	_A = None   ,	_A = None   ,	_A = None   ,	_A = False   ,	**_A   ,	):





     '''simple docstring'''




     SCREAMING_SNAKE_CASE__					=					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 math
import time

from transformers import Trainer, is_torch_tpu_available
from transformers.trainer_utils import PredictionOutput, speed_metrics


if is_torch_tpu_available(check_device=False):
       import torch_xla.core.xla_model as xm
       import torch_xla.debug.metrics as met






class  UpperCAmelCase__	( A__ ):






  """simple docstring"""


  def __init__(							self		:	Tuple			,     *__lowerCamelCase		:	List[str]			,     __lowerCamelCase		:	Any=None			,     __lowerCamelCase		:	Optional[int]=None			,     **__lowerCamelCase		:	Optional[int]      )	->							List[str]:
       super().__init__(*__lowerCamelCase			,     **__lowerCamelCase      )
       SCREAMING_SNAKE_CASE__					=					eval_examples
       SCREAMING_SNAKE_CASE__					=					post_process_function






  def    lowercase_	(							self		:	Optional[Any]			,     __lowerCamelCase		:	Dict=None			,     __lowerCamelCase		:	int=None			,     __lowerCamelCase		:	str=None			,     __lowerCamelCase		:	str = "eval"      )	->							Optional[Any]:
       SCREAMING_SNAKE_CASE__					=					self.eval_dataset if eval_dataset is None else eval_dataset
       SCREAMING_SNAKE_CASE__					=					self.get_eval_dataloader(__lowerCamelCase      )
       SCREAMING_SNAKE_CASE__					=					self.eval_examples if eval_examples is None else eval_examples

       # Temporarily disable metric computation, we will do it in the loop here.
       SCREAMING_SNAKE_CASE__					=					self.compute_metrics
       SCREAMING_SNAKE_CASE__					=					None
       SCREAMING_SNAKE_CASE__					=					self.prediction_loop if self.args.use_legacy_prediction_loop else self.evaluation_loop
       SCREAMING_SNAKE_CASE__					=					time.time()
       try:
            SCREAMING_SNAKE_CASE__					=					eval_loop(
                __lowerCamelCase			,     description='''Evaluation'''			,     prediction_loss_only=True if compute_metrics is None else None			,     ignore_keys=__lowerCamelCase			,     metric_key_prefix=__lowerCamelCase			,     )
       finally:
            SCREAMING_SNAKE_CASE__					=					compute_metrics
       SCREAMING_SNAKE_CASE__					=					self.args.eval_batch_size * self.args.world_size
       if f'''{metric_key_prefix}_jit_compilation_time''' in output.metrics:
            start_time += output.metrics[f'''{metric_key_prefix}_jit_compilation_time''']
       output.metrics.update(
           speed_metrics(
               __lowerCamelCase			,     __lowerCamelCase			,     num_samples=output.num_samples			,     num_steps=math.ceil(output.num_samples / total_batch_size      )			,     )      )
       if self.post_process_function is not None and self.compute_metrics is not None and self.args.should_save:
            # Only the main node write the results by default
            SCREAMING_SNAKE_CASE__					=					self.post_process_function(__lowerCamelCase			,     __lowerCamelCase			,     output.predictions      )
            SCREAMING_SNAKE_CASE__					=					self.compute_metrics(__lowerCamelCase      )

            # Prefix all keys with metric_key_prefix + '_'
            for key in list(metrics.keys()      ):
                 if not key.startswith(f'''{metric_key_prefix}_'''      ):
                      SCREAMING_SNAKE_CASE__					=					metrics.pop(__lowerCamelCase      )
            metrics.update(output.metrics      )
       else:
            SCREAMING_SNAKE_CASE__					=					output.metrics

       if self.args.should_log:
            # Only the main node log the results by default
            self.log(__lowerCamelCase      )

       if self.args.tpu_metrics_debug or self.args.debug:
            # tpu-comment: Logging debug metrics for PyTorch/XLA (compile, execute times, ops, etc.)
            xm.master_print(met.metrics_report()      )

       SCREAMING_SNAKE_CASE__					=					self.callback_handler.on_evaluate(self.args			,     self.state			,     self.control			,     __lowerCamelCase      )
       return metrics







  def    lowercase_	(							self		:	Dict			,     __lowerCamelCase		:	Dict			,     __lowerCamelCase		:	str			,     __lowerCamelCase		:	str=None			,     __lowerCamelCase		:	str = "test"      )	->							Any:
       SCREAMING_SNAKE_CASE__					=					self.get_test_dataloader(__lowerCamelCase      )

       # Temporarily disable metric computation, we will do it in the loop here.
       SCREAMING_SNAKE_CASE__					=					self.compute_metrics
       SCREAMING_SNAKE_CASE__					=					None
       SCREAMING_SNAKE_CASE__					=					self.prediction_loop if self.args.use_legacy_prediction_loop else self.evaluation_loop
       SCREAMING_SNAKE_CASE__					=					time.time()
       try:
            SCREAMING_SNAKE_CASE__					=					eval_loop(
                __lowerCamelCase			,     description='''Prediction'''			,     prediction_loss_only=True if compute_metrics is None else None			,     ignore_keys=__lowerCamelCase			,     metric_key_prefix=__lowerCamelCase			,     )
       finally:
            SCREAMING_SNAKE_CASE__					=					compute_metrics
       SCREAMING_SNAKE_CASE__					=					self.args.eval_batch_size * self.args.world_size
       if f'''{metric_key_prefix}_jit_compilation_time''' in output.metrics:
            start_time += output.metrics[f'''{metric_key_prefix}_jit_compilation_time''']
       output.metrics.update(
           speed_metrics(
               __lowerCamelCase			,     __lowerCamelCase			,     num_samples=output.num_samples			,     num_steps=math.ceil(output.num_samples / total_batch_size      )			,     )      )

       if self.post_process_function is None or self.compute_metrics is None:
            return output

       SCREAMING_SNAKE_CASE__					=					self.post_process_function(__lowerCamelCase			,     __lowerCamelCase			,     output.predictions			,     '''predict'''      )
       SCREAMING_SNAKE_CASE__					=					self.compute_metrics(__lowerCamelCase      )

       # Prefix all keys with metric_key_prefix + '_'
       for key in list(metrics.keys()      ):
            if not key.startswith(f'''{metric_key_prefix}_'''      ):
                 SCREAMING_SNAKE_CASE__					=					metrics.pop(__lowerCamelCase      )
       metrics.update(output.metrics      )
       return PredictionOutput(predictions=predictions.predictions			,     label_ids=predictions.label_ids			,     metrics=__lowerCamelCase      )


