




'''simple docstring'''




import argparse
import glob
import importlib.util
import os
import re

import black
from doc_builder.style_doc import style_docstrings_in_code


# All paths are set with the intent you should run this script from the root of the repo with the command
# python utils/check_copies.py
a_      :		Optional[int]							      =       """src/diffusers"""
a_      :		str							      =       """."""


# This is to make sure the diffusers module imported is the one in the repo.
a_      :		List[str]							      =       importlib.util.spec_from_file_location(
    """diffusers""",
    os.path.join(DIFFUSERS_PATH, """__init__.py"""),
    submodule_search_locations=[DIFFUSERS_PATH],
)
a_      :		List[Any]							      =       spec.loader.load_module()





def      a_							(      __snake_case							:			Any   ,       __snake_case							:			Any   )							->						int:



       """simple docstring"""
       return line.startswith(__snake_case   ) or len(__snake_case   ) <= 1 or re.search(r'''^\s*\)(\s*->.*:|:)\s*$'''   ,       __snake_case   ) is not None





def      a_							(      __snake_case							:			Any   )							->						List[Any]:



       """simple docstring"""
       lowerCamelCase_							=object_name.split('''.'''   )
       lowerCamelCase_							=0

       # First let's find the module where our object lives.
       lowerCamelCase_							=parts[i]
       while i < len(__snake_case   ) and not os.path.isfile(os.path.join(__snake_case   ,       F'''{module}.py'''   )   ):
              i += 1
              if i < len(__snake_case   ):
                     lowerCamelCase_							=os.path.join(__snake_case   ,       parts[i]   )
       if i >= len(__snake_case   ):
              raise ValueError(F'''`object_name` should begin with the name of a module of diffusers but got {object_name}.'''   )

       with open(os.path.join(__snake_case   ,       F'''{module}.py'''   )   ,       '''r'''   ,       encoding='''utf-8'''   ,       newline='''\n'''   ) as f:
              lowerCamelCase_							=f.readlines()

       # Now let's find the class / func in the code!
       lowerCamelCase_							=''''''
       lowerCamelCase_							=0
       for name in parts[i + 1 :]:
              while (
                  line_index < len(__snake_case   ) and re.search(rF'''^{indent}(class|def)\s+{name}(\(|\:)'''   ,       lines[line_index]   ) is None
              ):
                     line_index += 1
              indent += "    "
              line_index += 1

       if line_index >= len(__snake_case   ):
              raise ValueError(F''' {object_name} does not match any function or class in {module}.'''   )

       # We found the beginning of the class / func, now let's find the end (when the indent diminishes).
       lowerCamelCase_							=line_index
       while line_index < len(__snake_case   ) and _should_continue(lines[line_index]   ,       __snake_case   ):
              line_index += 1
       # Clean up empty lines at the end (if any).
       while len(lines[line_index - 1]   ) <= 1:
              line_index -= 1

       lowerCamelCase_							=lines[start_index:line_index]
       return "".join(__snake_case   )


a_      :		Tuple							      =       re.compile(R"""^(\s*)#\s*Copied from\s+diffusers\.(\S+\.\S+)\s*($|\S.*$)""")
a_      :		Optional[int]							      =       re.compile(R"""^\s*(\S+)->(\S+)(\s+.*|$)""")
a_      :		Union[str, Any]							      =       re.compile(R"""<FILL\s+[^>]*>""")





def      a_							(      __snake_case							:			Dict   )							->						Tuple:



       """simple docstring"""
       lowerCamelCase_							=code.split('''\n'''   )
       lowerCamelCase_							=0
       while idx < len(__snake_case   ) and len(lines[idx]   ) == 0:
              idx += 1
       if idx < len(__snake_case   ):
              return re.search(r'''^(\s*)\S'''   ,       lines[idx]   ).groups()[0]
       return ""





def      a_							(      __snake_case							:			Any   )							->						int:



       """simple docstring"""
       lowerCamelCase_							=len(get_indent(__snake_case   )   ) > 0
       if has_indent:
              lowerCamelCase_							=F'''class Bla:\n{code}'''
       lowerCamelCase_							=black.Mode(target_versions={black.TargetVersion.PYaa}   ,       line_length=119   ,       preview=__snake_case   )
       lowerCamelCase_							=black.format_str(__snake_case   ,       mode=__snake_case   )
       lowerCamelCase_,      lowerCamelCase_							=style_docstrings_in_code(__snake_case   )
       return result[len('''class Bla:\n'''   ) :] if has_indent else result





def      a_							(      __snake_case							:			str   ,       __snake_case							:			Union[str, Any]=False   )							->						Optional[int]:



       """simple docstring"""
       with open(__snake_case   ,       '''r'''   ,       encoding='''utf-8'''   ,       newline='''\n'''   ) as f:
              lowerCamelCase_							=f.readlines()
       lowerCamelCase_							=[]
       lowerCamelCase_							=0
       # Not a for loop cause `lines` is going to change (if `overwrite=True`).
       while line_index < len(__snake_case   ):
              lowerCamelCase_							=_re_copy_warning.search(lines[line_index]   )
              if search is None:
                     line_index += 1
                     continue

              # There is some copied code here, let's retrieve the original.
              lowerCamelCase_,      lowerCamelCase_,      lowerCamelCase_							=search.groups()
              lowerCamelCase_							=find_code_in_diffusers(__snake_case   )
              lowerCamelCase_							=get_indent(__snake_case   )

              lowerCamelCase_							=line_index + 1 if indent == theoretical_indent else line_index + 2
              lowerCamelCase_							=theoretical_indent
              lowerCamelCase_							=start_index

              # Loop to check the observed code, stop when indentation diminishes or if we see a End copy comment.
              lowerCamelCase_							=True
              while line_index < len(__snake_case   ) and should_continue:
                     line_index += 1
                     if line_index >= len(__snake_case   ):
                            break
                     lowerCamelCase_							=lines[line_index]
                     lowerCamelCase_							=_should_continue(__snake_case   ,       __snake_case   ) and re.search(F'''^{indent}# End copy'''   ,       __snake_case   ) is None
              # Clean up empty lines at the end (if any).
              while len(lines[line_index - 1]   ) <= 1:
                     line_index -= 1

              lowerCamelCase_							=lines[start_index:line_index]
              lowerCamelCase_							=''''''.join(__snake_case   )

              # Remove any nested `Copied from` comments to avoid circular copies
              lowerCamelCase_							=[line for line in theoretical_code.split('''\n'''   ) if _re_copy_warning.search(__snake_case   ) is None]
              lowerCamelCase_							='''\n'''.join(__snake_case   )

              # Before comparing, use the `replace_pattern` on the original code.
              if len(__snake_case   ) > 0:
                     lowerCamelCase_							=replace_pattern.replace('''with'''   ,       ''''''   ).split(''','''   )
                     lowerCamelCase_							=[_re_replace_pattern.search(__snake_case   ) for p in patterns]
                     for pattern in patterns:
                            if pattern is None:
                                   continue
                            lowerCamelCase_,      lowerCamelCase_,      lowerCamelCase_							=pattern.groups()
                            lowerCamelCase_							=re.sub(__snake_case   ,       __snake_case   ,       __snake_case   )
                            if option.strip() == "all-casing":
                                   lowerCamelCase_							=re.sub(obja.lower()   ,       obja.lower()   ,       __snake_case   )
                                   lowerCamelCase_							=re.sub(obja.upper()   ,       obja.upper()   ,       __snake_case   )

            # Blackify after replacement. To be able to do that, we need the header (class or function definition)
            # from the previous line
                     lowerCamelCase_							=blackify(lines[start_index - 1] + theoretical_code   )
                     lowerCamelCase_							=theoretical_code[len(lines[start_index - 1]   ) :]

              # Test for a diff and act accordingly.
              if observed_code != theoretical_code:
                     diffs.append([object_name, start_index]   )
                     if overwrite:
                            lowerCamelCase_							=lines[:start_index] + [theoretical_code] + lines[line_index:]
                            lowerCamelCase_							=start_index + 1

       if overwrite and len(__snake_case   ) > 0:
              # Warn the user a file has been modified.
              print(F'''Detected changes, rewriting {filename}.'''   )
              with open(__snake_case   ,       '''w'''   ,       encoding='''utf-8'''   ,       newline='''\n'''   ) as f:
                     f.writelines(__snake_case   )
       return diffs





def      a_							(      __snake_case							:			bool = False   )							->						Dict:



       """simple docstring"""
       lowerCamelCase_							=glob.glob(os.path.join(__snake_case   ,       '''**/*.py'''   )   ,       recursive=__snake_case   )
       lowerCamelCase_							=[]
       for filename in all_files:
              lowerCamelCase_							=is_copy_consistent(__snake_case   ,       __snake_case   )
              diffs += [F'''- {filename}: copy does not match {d[0]} at line {d[1]}''' for d in new_diffs]
       if not overwrite and len(__snake_case   ) > 0:
              lowerCamelCase_							='''\n'''.join(__snake_case   )
              raise Exception(
                  '''Found the following copy inconsistencies:\n'''
                  + diff
                  + '''\nRun `make fix-copies` or `python utils/check_copies.py --fix_and_overwrite` to fix them.'''   )


if __name__ == "__main__":
     a_      :		str							      =       argparse.ArgumentParser()
     parser.add_argument("""--fix_and_overwrite""", action="""store_true""", help="""Whether to fix inconsistencies.""")
     a_      :		Union[str, Any]							      =       parser.parse_args()

     check_copies(args.fix_and_overwrite)












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						:    Dict,  *lowerCamelCase						:    Union[str, Any],  lowerCamelCase						:    Union[str, Any]=None,  lowerCamelCase						:    Any=None,  **lowerCamelCase						:    str  ):

								'''simple docstring'''
								super().__init__(*lowerCamelCase,  **lowerCamelCase  )
								lowercase__    =			eval_examples
								lowercase__    =			post_process_function




				def 				lowercase__   (     self						:    int,  lowerCamelCase						:    str=None,  lowerCamelCase						:    Optional[Any]=None,  lowerCamelCase						:    Union[str, Any]=None,  lowerCamelCase						:    str = "eval"  ):

								'''simple docstring'''
								lowercase__    =			self.eval_dataset if eval_dataset is None else eval_dataset
								lowercase__    =			self.get_eval_dataloader(lowerCamelCase  )
								lowercase__    =			self.eval_examples if eval_examples is None else eval_examples

								# Temporarily disable metric computation, we will do it in the loop here.
								lowercase__    =			self.compute_metrics
								lowercase__    =			None
								lowercase__    =			self.prediction_loop if self.args.use_legacy_prediction_loop else self.evaluation_loop
								lowercase__    =			time.time()
								try:
												lowercase__    =			eval_loop(
												    lowerCamelCase,  description='''Evaluation''',  prediction_loss_only=True if compute_metrics is None else None,  ignore_keys=lowerCamelCase,  metric_key_prefix=lowerCamelCase,  )
								finally:
												lowercase__    =			compute_metrics
								lowercase__    =			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
												lowercase__    =			self.post_process_function(lowerCamelCase,  lowerCamelCase,  output.predictions  )
												lowercase__    =			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}_"""  ):
																				lowercase__    =			metrics.pop(lowerCamelCase  )
												metrics.update(output.metrics  )
								else:
												lowercase__    =			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()  )

								lowercase__    =			self.callback_handler.on_evaluate(self.args,  self.state,  self.control,  lowerCamelCase  )
								return metrics




				def 				lowercase__   (     self						:    List[Any],  lowerCamelCase						:    Any,  lowerCamelCase						:    Dict,  lowerCamelCase						:    int=None,  lowerCamelCase						:    str = "test"  ):

								'''simple docstring'''
								lowercase__    =			self.get_test_dataloader(lowerCamelCase  )

								# Temporarily disable metric computation, we will do it in the loop here.
								lowercase__    =			self.compute_metrics
								lowercase__    =			None
								lowercase__    =			self.prediction_loop if self.args.use_legacy_prediction_loop else self.evaluation_loop
								lowercase__    =			time.time()
								try:
												lowercase__    =			eval_loop(
												    lowerCamelCase,  description='''Prediction''',  prediction_loss_only=True if compute_metrics is None else None,  ignore_keys=lowerCamelCase,  metric_key_prefix=lowerCamelCase,  )
								finally:
												lowercase__    =			compute_metrics
								lowercase__    =			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

								lowercase__    =			self.post_process_function(lowerCamelCase,  lowerCamelCase,  output.predictions,  '''predict'''  )
								lowercase__    =			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}_"""  ):
																lowercase__    =			metrics.pop(lowerCamelCase  )
								metrics.update(output.metrics  )
								return PredictionOutput(predictions=predictions.predictions,  label_ids=predictions.label_ids,  metrics=lowerCamelCase  )




