


"""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
_snake_case          =							"src/diffusers"
_snake_case          =							"."


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



def   snake_case						(				_a:     str		,   _a:     List[Any]							)->		Optional[int]:
      '''simple docstring'''


      return line.startswith(_a							) or len(_a							) <= 1 or re.search(R'^\s*\)(\s*->.*:|:)\s*$'		,   _a							) is not None



def   snake_case						(				_a:     List[Any]							)->		Optional[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(_a							) and not os.path.isfile(os.path.join(_a		,   F'{module}.py'							)							):
            i += 1
            if i < len(_a							):
                  lowerCamelCase__           =		os.path.join(_a		,   parts[i]							)
      if i >= len(_a							):
            raise ValueError(F'`object_name` should begin with the name of a module of diffusers but got {object_name}.'							)

      with open(os.path.join(_a		,   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(_a							) 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(_a							):
            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(_a							) and _should_continue(lines[line_index]		,   _a							):
            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(_a							)


_snake_case          =							re.compile(R"^(\s*)#\s*Copied from\s+diffusers\.(\S+\.\S+)\s*($|\S.*$)")
_snake_case          =							re.compile(R"^\s*(\S+)->(\S+)(\s+.*|$)")
_snake_case          =							re.compile(R"<FILL\s+[^>]*>")



def   snake_case						(				_a:     Tuple							)->		Dict:
      '''simple docstring'''


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



def   snake_case						(				_a:     int							)->		List[Any]:
      '''simple docstring'''


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



def   snake_case						(				_a:     Union[str, Any]		,   _a:     int=False							)->		Any:
      '''simple docstring'''


      with open(_a		,   '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(_a							):
            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(_a							)
            lowerCamelCase__           =		get_indent(_a							)

            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(_a							) and should_continue:
                  line_index += 1
                  if line_index >= len(_a							):
                        break
                  lowerCamelCase__           =		lines[line_index]
                  lowerCamelCase__           =		_should_continue(_a		,   _a							) and re.search(F'^{indent}# End copy'		,   _a							) 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(_a							)

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

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

            # 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(_a							) > 0:
            # Warn the user a file has been modified.
            print(F'Detected changes, rewriting {filename}.'							)
            with open(_a		,   'w'		,   encoding='utf-8'		,   newline='\n'							) as f:
                  f.writelines(_a							)
      return diffs



def   snake_case						(				_a:     bool = False							)->		Union[str, Any]:
      '''simple docstring'''


      lowerCamelCase__           =		glob.glob(os.path.join(_a		,   '**/*.py'							)		,   recursive=_a							)
      lowerCamelCase__           =		[]
      for filename in all_files:
            lowerCamelCase__           =		is_copy_consistent(_a		,   _a							)
            diffs += [F'- {filename}: copy does not match {d[0]} at line {d[1]}' for d in new_diffs]
      if not overwrite and len(_a							) > 0:
            lowerCamelCase__           =		'\n'.join(_a							)
            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__":
   _snake_case          =							argparse.ArgumentParser()
   parser.add_argument("--fix_and_overwrite", action="store_true", help="Whether to fix inconsistencies.")
   _snake_case          =							parser.parse_args()

   check_copies(args.fix_and_overwrite)







"""simple docstring"""


import json
import multiprocessing
import os
import re
from collections import defaultdict

import torch
from accelerate import Accelerator
from accelerate.utils import set_seed
from arguments import HumanEvalArguments
from datasets import load_dataset, load_metric
from torch.utils.data import IterableDataset
from torch.utils.data.dataloader import DataLoader
from tqdm import tqdm

import transformers
from transformers import AutoModelForCausalLM, AutoTokenizer, HfArgumentParser, StoppingCriteria, StoppingCriteriaList


_snake_case          =							["\nclass", "\ndef", "\n#", "\n@", "\nprint", "\nif"]



class   _a      (      SCREAMING_SNAKE_CASE_ ):
    def __init__( self							:					List[str]							,					SCREAMING_SNAKE_CASE__							:					Any							,					SCREAMING_SNAKE_CASE__							:					List[Any]							,					SCREAMING_SNAKE_CASE__							:					int=None							,					SCREAMING_SNAKE_CASE__							:					Optional[int]=1				):
          lowerCamelCase__           =		tokenizer
          lowerCamelCase__           =		dataset
          lowerCamelCase__           =		len(SCREAMING_SNAKE_CASE__				) if n_tasks is None else n_tasks
          lowerCamelCase__           =		n_copies


    def __iter__( self							:					Any				):
          lowerCamelCase__           =		[]
          for task in range(self.n_tasks				):
                # without strip, the model generate commented codes ...
                prompts.append(self.tokenizer.eos_token + self.dataset[task]['prompt'].strip()				)
          lowerCamelCase__           =		self.tokenizer(SCREAMING_SNAKE_CASE__							,					padding=SCREAMING_SNAKE_CASE__							,					return_tensors='pt'				)
          for task in range(self.n_tasks				):
                for _ in range(self.n_copies				):
                      yield {
                          "ids": outputs.input_ids[task],
                          "task_id": task,
                          "input_len": outputs.attention_mask[task].sum(),
                      }





class   _a      (      SCREAMING_SNAKE_CASE_ ):
    def __init__( self							:					Dict							,					SCREAMING_SNAKE_CASE__							:					List[str]							,					SCREAMING_SNAKE_CASE__							:					List[Any]							,					SCREAMING_SNAKE_CASE__							:					Dict				):
          lowerCamelCase__           =		start_length
          lowerCamelCase__           =		eof_strings
          lowerCamelCase__           =		tokenizer


    def __call__( self							:					Any							,					SCREAMING_SNAKE_CASE__							:					Any							,					SCREAMING_SNAKE_CASE__							:					List[str]							,					**SCREAMING_SNAKE_CASE__							:					Dict				):
          lowerCamelCase__           =		self.tokenizer.batch_decode(input_ids[:, self.start_length :]				)
          lowerCamelCase__           =		[]
          for decoded_generation in decoded_generations:
                done.append(any(stop_string in decoded_generation for stop_string in self.eof_strings				)				)
          return all(SCREAMING_SNAKE_CASE__				)





def   snake_case						(				_a:     List[Any]							)->		Dict:
      '''simple docstring'''


      lowerCamelCase__           =		re.split('(%s)' % '|'.join(_a							)		,   _a							)
      # last string should be ""
      return "".join(string_list[:-2]							)



def   snake_case						(				_a:     List[Any]		,   _a:     Optional[int]		,   _a:     str		,   _a:     Union[str, Any]		,   _a:     Dict		,   _a:     Optional[int]=20		,   **_a:     Optional[int]							)->		List[str]:
      '''simple docstring'''


      lowerCamelCase__           =		defaultdict(_a							)  # dict of list of generated tokens
      for step, batch in tqdm(enumerate(_a							)							):
            with torch.no_grad():
                  lowerCamelCase__           =		batch['ids'].shape[-1]
                  lowerCamelCase__           =		accelerator.unwrap_model(_a							).generate(
                      input_ids=batch['ids'][:, : batch['input_len']]		,   num_return_sequences=_a		,   **_a							)
                  # each task is generated batch_size times
                  lowerCamelCase__           =		batch['task_id'].repeat(_a							)
                  lowerCamelCase__           =		accelerator.pad_across_processes(
                      _a		,   dim=1		,   pad_index=tokenizer.pad_token_id							)

                  lowerCamelCase__    ,  lowerCamelCase__           =		accelerator.gather((generated_tokens, generated_tasks)							)
                  lowerCamelCase__           =		generated_tokens.cpu().numpy()
                  lowerCamelCase__           =		generated_tasks.cpu().numpy()

                  for task, generated_tokens in zip(_a		,   _a							):
                        gen_token_dict[task].append(_a							)

      lowerCamelCase__           =		[[] for _ in range(_a							)]
      for task, generated_tokens in gen_token_dict.items():
            for s in generated_tokens:
                  lowerCamelCase__           =		tokenizer.decode(_a		,   skip_special_tokens=_a		,   clean_up_tokenization_spaces=_a							)
                  code_gens[task].append(remove_last_block(_a							)							)
      return code_gens



def   snake_case						(				)->		Union[str, Any]:
      '''simple docstring'''


      lowerCamelCase__           =		HfArgumentParser(_a							)
      lowerCamelCase__           =		parser.parse_args()

      transformers.logging.set_verbosity_error()
      # enables code execution in code_eval metric
      lowerCamelCase__           =		args.HF_ALLOW_CODE_EVAL
      # make sure tokenizer plays nice with multiprocessing
      lowerCamelCase__           =		'false'

      if args.num_workers is None:
            lowerCamelCase__           =		multiprocessing.cpu_count()

      # Use dataset load to feed to accelerate
      lowerCamelCase__           =		Accelerator()
      set_seed(args.seed		,   device_specific=_a							)

      # Load model and tokenizer
      lowerCamelCase__           =		AutoTokenizer.from_pretrained(args.model_ckpt							)
      lowerCamelCase__           =		tokenizer.eos_token
      lowerCamelCase__           =		AutoModelForCausalLM.from_pretrained(args.model_ckpt							)

      # Generation settings
      lowerCamelCase__           =		{
          'do_sample': args.do_sample,
          'temperature': args.temperature,
          'max_new_tokens': args.max_new_tokens,
          'top_p': args.top_p,
          'top_k': args.top_k,
          'stopping_criteria': StoppingCriteriaList([EndOfFunctionCriteria(0		,   _a		,   _a							)]							),
      }

      # Load evaluation dataset and metric
      lowerCamelCase__           =		load_dataset('openai_humaneval'							)
      lowerCamelCase__           =		load_metric('code_eval'							)

      lowerCamelCase__           =		args.num_tasks if args.num_tasks is not None else len(human_eval['test']							)
      lowerCamelCase__           =		args.n_samples // args.batch_size

      lowerCamelCase__           =		TokenizedDataset(_a		,   human_eval['test']		,   n_copies=_a		,   n_tasks=_a							)
      # do not confuse args.batch_size, which is actually the num_return_sequences
      lowerCamelCase__           =		DataLoader(_a		,   batch_size=1							)

      # Run a quick test to see if code evaluation is enabled
      try:
            lowerCamelCase__           =		code_eval_metric.compute(references=['']		,   predictions=[['']]							)
      except ValueError as exception:
            print(
                'Code evaluation not enabled. Read the warning below carefully and then use `--HF_ALLOW_CODE_EVAL="1"`'
                ' flag to enable code evaluation.'							)
            raise exception

      lowerCamelCase__    ,  lowerCamelCase__           =		accelerator.prepare(_a		,   _a							)

      lowerCamelCase__           =		complete_code(
          _a		,   _a		,   _a		,   _a		,   n_tasks=_a		,   batch_size=args.batch_size		,   **_a		,   )

      if accelerator.is_main_process:
            lowerCamelCase__           =		[]

            for task in tqdm(range(_a							)							):
                  lowerCamelCase__           =		human_eval['test'][task]['test']
                  lowerCamelCase__           =		F'check({human_eval["test"][task]["entry_point"]})'
                  references.append('\n' + test_func + '\n' + entry_point							)

            # Evaluate completions with "code_eval" metric
            lowerCamelCase__    ,  lowerCamelCase__           =		code_eval_metric.compute(
                references=_a		,   predictions=_a		,   num_workers=args.num_workers							)
            print(F'Results: {pass_at_k}'							)

            # Save results to json file
            with open(args.output_file		,   'w'							) as fp:
                  json.dump(_a		,   _a							)


# For some reason the folliwng seems to be necessary sometimes for code_eval to work nice with multiprocessing
# https://stackoverflow.com/questions/60804599/python-multiprocessing-keeps-spawning-the-whole-script
if __name__ == "__main__":
   main()



