



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__								=						'''src/diffusers'''
A__								=						'''.'''


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



def 			_lowerCAmelCase       (					__lowerCAmelCase  ,  __lowerCAmelCase      )   ->					str:


				"""simple docstring"""

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



def 			_lowerCAmelCase       (					__lowerCAmelCase      )   ->					int:


				"""simple docstring"""

				snake_case__	:      List[str]     		=	object_name.split('''.'''      )
				snake_case__	:      int     		=	0

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

				with open(os.path.join(__lowerCAmelCase  ,  f"""{module}.py"""      )  ,  '''r'''  ,  encoding='''utf-8'''  ,  newline='''\n'''      ) as f:
								snake_case__	:      Optional[Any]     		=	f.readlines()

				# Now let's find the class / func in the code!
				snake_case__	:      Dict     		=	''''''
				snake_case__	:      List[str]     		=	0
				for name in parts[i + 1 :]:
								while (
								    line_index < len(__lowerCAmelCase      ) 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(__lowerCAmelCase      ):
								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).
				snake_case__	:      Dict     		=	line_index
				while line_index < len(__lowerCAmelCase      ) and _should_continue(lines[line_index]  ,  __lowerCAmelCase      ):
								line_index += 1
				# Clean up empty lines at the end (if any).
				while len(lines[line_index - 1]      ) <= 1:
								line_index -= 1

				snake_case__	:      List[Any]     		=	lines[start_index:line_index]
				return "".join(__lowerCAmelCase      )


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



def 			_lowerCAmelCase       (					__lowerCAmelCase      )   ->					List[Any]:


				"""simple docstring"""

				snake_case__	:      Tuple     		=	code.split('''\n'''      )
				snake_case__	:      List[Any]     		=	0
				while idx < len(__lowerCAmelCase      ) and len(lines[idx]      ) == 0:
								idx += 1
				if idx < len(__lowerCAmelCase      ):
								return re.search(r'''^(\s*)\S'''  ,  lines[idx]      ).groups()[0]
				return ""



def 			_lowerCAmelCase       (					__lowerCAmelCase      )   ->					Optional[int]:


				"""simple docstring"""

				snake_case__	:      str     		=	len(get_indent(__lowerCAmelCase      )      ) > 0
				if has_indent:
								snake_case__	:      Dict     		=	f"""class Bla:\n{code}"""
				snake_case__	:      str     		=	black.Mode(target_versions={black.TargetVersion.PYaa}  ,  line_length=119  ,  preview=__lowerCAmelCase      )
				snake_case__	:      Tuple     		=	black.format_str(__lowerCAmelCase  ,  mode=__lowerCAmelCase      )
				snake_case__		, snake_case__	:      Tuple     		=	style_docstrings_in_code(__lowerCAmelCase      )
				return result[len('''class Bla:\n'''      ) :] if has_indent else result



def 			_lowerCAmelCase       (					__lowerCAmelCase  ,  __lowerCAmelCase=False      )   ->					Dict:


				"""simple docstring"""

				with open(__lowerCAmelCase  ,  '''r'''  ,  encoding='''utf-8'''  ,  newline='''\n'''      ) as f:
								snake_case__	:      Optional[Any]     		=	f.readlines()
				snake_case__	:      List[str]     		=	[]
				snake_case__	:      Dict     		=	0
				# Not a for loop cause `lines` is going to change (if `overwrite=True`).
				while line_index < len(__lowerCAmelCase      ):
								snake_case__	:      Dict     		=	_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.
								snake_case__		, snake_case__		, snake_case__	:      List[Any]     		=	search.groups()
								snake_case__	:      int     		=	find_code_in_diffusers(__lowerCAmelCase      )
								snake_case__	:      int     		=	get_indent(__lowerCAmelCase      )

								snake_case__	:      Tuple     		=	line_index + 1 if indent == theoretical_indent else line_index + 2
								snake_case__	:      List[Any]     		=	theoretical_indent
								snake_case__	:      str     		=	start_index

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

								snake_case__	:      Union[str, Any]     		=	lines[start_index:line_index]
								snake_case__	:      Union[str, Any]     		=	''''''.join(__lowerCAmelCase      )

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

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

            # Blackify after replacement. To be able to do that, we need the header (class or function definition)
            # from the previous line
												snake_case__	:      Dict     		=	blackify(lines[start_index - 1] + theoretical_code      )
												snake_case__	:      str     		=	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:
																snake_case__	:      Tuple     		=	lines[:start_index] + [theoretical_code] + lines[line_index:]
																snake_case__	:      int     		=	start_index + 1

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



def 			_lowerCAmelCase       (					__lowerCAmelCase = False      )   ->					str:


				"""simple docstring"""

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

				check_copies(args.fix_and_overwrite)






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


A__								=						['''\nclass''', '''\ndef''', '''\n#''', '''\n@''', '''\nprint''', '''\nif''']


class       a							(			__lowerCamelCase				):



		def __init__(     self     :str   ,__lowercase     :Dict   ,__lowercase     :str   ,__lowercase     :Any=None   ,__lowercase     :Union[str, Any]=1  ):
						snake_case__	:      Optional[int]     		=	tokenizer
						snake_case__	:      List[Any]     		=	dataset
						snake_case__	:      Any     		=	len(__lowercase  ) if n_tasks is None else n_tasks
						snake_case__	:      List[Any]     		=	n_copies






		def __iter__(     self     :Tuple  ):
						snake_case__	:      Union[str, Any]     		=	[]
						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()  )
						snake_case__	:      Optional[int]     		=	self.tokenizer(__lowercase   ,padding=__lowercase   ,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							(			__lowerCamelCase				):



		def __init__(     self     :Union[str, Any]   ,__lowercase     :List[Any]   ,__lowercase     :int   ,__lowercase     :List[str]  ):
						snake_case__	:      List[str]     		=	start_length
						snake_case__	:      Any     		=	eof_strings
						snake_case__	:      Optional[Any]     		=	tokenizer






		def __call__(     self     :str   ,__lowercase     :str   ,__lowercase     :Any   ,**__lowercase     :Optional[int]  ):
						snake_case__	:      Tuple     		=	self.tokenizer.batch_decode(input_ids[:, self.start_length :]  )
						snake_case__	:      Union[str, Any]     		=	[]
						for decoded_generation in decoded_generations:
										done.append(any(stop_string in decoded_generation for stop_string in self.eof_strings  )  )
						return all(__lowercase  )



def 			_lowerCAmelCase       (					__lowerCAmelCase      )   ->					List[Any]:


				"""simple docstring"""

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



def 			_lowerCAmelCase       (					__lowerCAmelCase  ,  __lowerCAmelCase  ,  __lowerCAmelCase  ,  __lowerCAmelCase  ,  __lowerCAmelCase  ,  __lowerCAmelCase=20  ,  **__lowerCAmelCase      )   ->					List[Any]:


				"""simple docstring"""

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

												snake_case__		, snake_case__	:      Any     		=	accelerator.gather((generated_tokens, generated_tasks)      )
												snake_case__	:      Union[str, Any]     		=	generated_tokens.cpu().numpy()
												snake_case__	:      Optional[Any]     		=	generated_tasks.cpu().numpy()

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

				snake_case__	:      List[Any]     		=	[[] for _ in range(__lowerCAmelCase      )]
				for task, generated_tokens in gen_token_dict.items():
								for s in generated_tokens:
												snake_case__	:      int     		=	tokenizer.decode(__lowerCAmelCase  ,  skip_special_tokens=__lowerCAmelCase  ,  clean_up_tokenization_spaces=__lowerCAmelCase      )
												code_gens[task].append(remove_last_block(__lowerCAmelCase      )      )
				return code_gens



def 			_lowerCAmelCase       (					)   ->					Optional[int]:


				"""simple docstring"""

				snake_case__	:      List[str]     		=	HfArgumentParser(__lowerCAmelCase      )
				snake_case__	:      Optional[Any]     		=	parser.parse_args()

				transformers.logging.set_verbosity_error()
				# enables code execution in code_eval metric
				snake_case__	:      Union[str, Any]     		=	args.HF_ALLOW_CODE_EVAL
				# make sure tokenizer plays nice with multiprocessing
				snake_case__	:      Union[str, Any]     		=	'''false'''

				if args.num_workers is None:
								snake_case__	:      Union[str, Any]     		=	multiprocessing.cpu_count()

				# Use dataset load to feed to accelerate
				snake_case__	:      Dict     		=	Accelerator()
				set_seed(args.seed  ,  device_specific=__lowerCAmelCase      )

				# Load model and tokenizer
				snake_case__	:      str     		=	AutoTokenizer.from_pretrained(args.model_ckpt      )
				snake_case__	:      List[str]     		=	tokenizer.eos_token
				snake_case__	:      Optional[Any]     		=	AutoModelForCausalLM.from_pretrained(args.model_ckpt      )

				# Generation settings
				snake_case__	:      List[Any]     		=	{
				    '''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  ,  __lowerCAmelCase  ,  __lowerCAmelCase      )]      ),
				}

				# Load evaluation dataset and metric
				snake_case__	:      List[str]     		=	load_dataset('''openai_humaneval'''      )
				snake_case__	:      str     		=	load_metric('''code_eval'''      )

				snake_case__	:      List[Any]     		=	args.num_tasks if args.num_tasks is not None else len(human_eval['''test''']      )
				snake_case__	:      str     		=	args.n_samples // args.batch_size

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

				# Run a quick test to see if code evaluation is enabled
				try:
								snake_case__	:      Optional[Any]     		=	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

				snake_case__		, snake_case__	:      Optional[Any]     		=	accelerator.prepare(__lowerCAmelCase  ,  __lowerCAmelCase      )

				snake_case__	:      Dict     		=	complete_code(
				    __lowerCAmelCase  ,  __lowerCAmelCase  ,  __lowerCAmelCase  ,  __lowerCAmelCase  ,  n_tasks=__lowerCAmelCase  ,  batch_size=args.batch_size  ,  **__lowerCAmelCase  ,  )

				if accelerator.is_main_process:
								snake_case__	:      Optional[Any]     		=	[]

								for task in tqdm(range(__lowerCAmelCase      )      ):
												snake_case__	:      str     		=	human_eval['''test'''][task]['''test''']
												snake_case__	:      int     		=	f"""check({human_eval['test'][task]['entry_point']})"""
												references.append('''\n''' + test_func + '''\n''' + entry_point      )

								# Evaluate completions with "code_eval" metric
								snake_case__		, snake_case__	:      Any     		=	code_eval_metric.compute(
								    references=__lowerCAmelCase  ,  predictions=__lowerCAmelCase  ,  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(__lowerCAmelCase  ,  __lowerCAmelCase      )


# 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()

