





from __future__ import annotations

from random import random
from typing import Generic, TypeVar

_SCREAMING_SNAKE_CASE 		=						TypeVar("""KT""")
_SCREAMING_SNAKE_CASE 		=						TypeVar("""VT""")
class SCREAMING_SNAKE_CASE_  (   Generic[KT, VT]   ):






						def __init__( self : List[str]						,				_A : KT | str = "root"						,				_A : VT | None = None		)							->      int:







							"""simple docstring"""


							snake_case_						:     Optional[int]     =		key
							snake_case_						:     Optional[int]     =		value
							snake_case_						:     list[Node[KT, VT]]     =		[]






						def __repr__( self : Tuple		)							->      str:







							"""simple docstring"""


							return F"""Node({self.key}: {self.value})"""






						@property
						def UpperCAmelCase_			( self : Dict		)							->      int:







							"""simple docstring"""


							return len(self.forward		)
class SCREAMING_SNAKE_CASE_  (   Generic[KT, VT]   ):






						def __init__( self : Optional[Any]						,				_A : float = 0.5						,				_A : int = 16		)							->      Optional[int]:







							"""simple docstring"""


							snake_case_						:     Node[KT, VT]     =		Node[KT, VT]()
							snake_case_						:     str     =		0
							snake_case_						:     Optional[int]     =		p
							snake_case_						:     str     =		max_level






						def __str__( self : Optional[Any]		)							->      str:







							"""simple docstring"""


							snake_case_						:     Dict     =		list(self		)

							if len(_A		) == 0:
								return F"""SkipList(level={self.level})"""

							snake_case_						:     Union[str, Any]     =		max((len(str(_A		)		) for item in items)						,				default=4		)
							snake_case_						:     Tuple     =		max(_A						,				4		) + 4

							snake_case_						:     Optional[Any]     =		self.head
							snake_case_						:     Optional[Any]     =		[]

							snake_case_						:     Any     =		node.forward.copy()
							lines.append(F"""[{node.key}]""".ljust(_A						,				'-'		) + '* ' * len(_A		)		)
							lines.append(' ' * label_size + '| ' * len(_A		)		)

							while len(node.forward		) != 0:
								snake_case_						:     Optional[Any]     =		node.forward[0]

								lines.append(
								    F"""[{node.key}]""".ljust(_A						,				'-'		)
								    + ' '.join(str(n.key		) if n.key == node.key else '|' for n in forwards		)		)
								lines.append(' ' * label_size + '| ' * len(_A		)		)
								snake_case_						:     Optional[Any]     =		node.forward

							lines.append('None'.ljust(_A		) + '* ' * len(_A		)		)
							return F"""SkipList(level={self.level})\n""" + "\n".join(_A		)






						def __iter__( self : List[str]		)							->      str:







							"""simple docstring"""


							snake_case_						:     int     =		self.head

							while len(node.forward		) != 0:
								yield node.forward[0].key
								snake_case_						:     str     =		node.forward[0]






						def UpperCAmelCase_			( self : str		)							->      int:







							"""simple docstring"""


							snake_case_						:     Tuple     =		1
							while random() < self.p and level < self.max_level:
								level += 1

							return level






						def UpperCAmelCase_			( self : int						,				_A : str		)							->      tuple[Node[KT, VT] | None, list[Node[KT, VT]]]:







							"""simple docstring"""


							snake_case_						:     Dict     =		[]

							snake_case_						:     Dict     =		self.head

							for i in reversed(range(self.level		)		):
								# i < node.level - When node level is lesser than `i` decrement `i`.
								# node.forward[i].key < key - Jumping to node with key value higher
								#                             or equal to searched key would result
								#                             in skipping searched key.
								while i < node.level and node.forward[i].key < key:
									snake_case_						:     Optional[Any]     =		node.forward[i]
								# Each leftmost node (relative to searched node) will potentially have to
								# be updated.
								update_vector.append(_A		)

							update_vector.reverse()  # Note that we were inserting values in reverse order.

							# len(node.forward) != 0 - If current node doesn't contain any further
							#                          references then searched key is not present.
							# node.forward[0].key == key - Next node key should be equal to search key
							#                              if key is present.
							if len(node.forward		) != 0 and node.forward[0].key == key:
								return node.forward[0], update_vector
							else:
								return None, update_vector






						def UpperCAmelCase_			( self : List[str]						,				_A : KT		)							->      List[str]:







							"""simple docstring"""


							snake_case_  ,snake_case_						:     str     =		self._locate_node(_A		)

							if node is not None:
								for i, update_node in enumerate(_A		):
									# Remove or replace all references to removed node.
									if update_node.level > i and update_node.forward[i].key == key:
										if node.level > i:
											snake_case_						:     Optional[Any]     =		node.forward[i]
										else:
											snake_case_						:     Any     =		update_node.forward[:i]






						def UpperCAmelCase_			( self : Any						,				_A : KT						,				_A : VT		)							->      Any:







							"""simple docstring"""


							snake_case_  ,snake_case_						:     int     =		self._locate_node(_A		)
							if node is not None:
								snake_case_						:     Tuple     =		value
							else:
								snake_case_						:     List[Any]     =		self.random_level()

								if level > self.level:
									# After level increase we have to add additional nodes to head.
									for _ in range(self.level - 1						,				_A		):
										update_vector.append(self.head		)
									snake_case_						:     List[Any]     =		level

								snake_case_						:     Union[str, Any]     =		Node(_A						,				_A		)

								for i, update_node in enumerate(update_vector[:level]		):
									# Change references to pass through new node.
									if update_node.level > i:
										new_node.forward.append(update_node.forward[i]		)

									if update_node.level < i + 1:
										update_node.forward.append(_A		)
									else:
										snake_case_						:     Tuple     =		new_node






						def UpperCAmelCase_			( self : Union[str, Any]						,				_A : VT		)							->      VT | None:







							"""simple docstring"""


							snake_case_  ,snake_case_						:     List[Any]     =		self._locate_node(_A		)

							if node is not None:
								return node.value

							return None


def 							SCREAMING_SNAKE_CASE__  (      ):
	snake_case_						:     Any     =		SkipList()
	skip_list.insert('Key1'				,	3				)
	skip_list.insert('Key2'				,	12				)
	skip_list.insert('Key3'				,	41				)
	skip_list.insert('Key4'				,	-19				)

	snake_case_						:     str     =		skip_list.head
	snake_case_						:     List[str]     =		{}
	while node.level != 0:
		snake_case_						:     List[Any]     =		node.forward[0]
		snake_case_						:     int     =		node.value

	assert len(__a				) == 4
	assert all_values["Key1"] == 3
	assert all_values["Key2"] == 12
	assert all_values["Key3"] == 41
	assert all_values["Key4"] == -19


def 							SCREAMING_SNAKE_CASE__  (      ):
	snake_case_						:     int     =		SkipList()
	skip_list.insert('Key1'				,	10				)
	skip_list.insert('Key1'				,	12				)

	skip_list.insert('Key5'				,	7				)
	skip_list.insert('Key7'				,	10				)
	skip_list.insert('Key10'				,	5				)

	skip_list.insert('Key7'				,	7				)
	skip_list.insert('Key5'				,	5				)
	skip_list.insert('Key10'				,	10				)

	snake_case_						:     Union[str, Any]     =		skip_list.head
	snake_case_						:     Union[str, Any]     =		{}
	while node.level != 0:
		snake_case_						:     List[str]     =		node.forward[0]
		snake_case_						:     str     =		node.value

	if len(__a				) != 4:
		print()
	assert len(__a				) == 4
	assert all_values["Key1"] == 12
	assert all_values["Key7"] == 7
	assert all_values["Key5"] == 5
	assert all_values["Key10"] == 10


def 							SCREAMING_SNAKE_CASE__  (      ):
	snake_case_						:     Dict     =		SkipList()
	assert skip_list.find('Some key'				) is None


def 							SCREAMING_SNAKE_CASE__  (      ):
	snake_case_						:     List[Any]     =		SkipList()

	skip_list.insert('Key2'				,	20				)
	assert skip_list.find('Key2'				) == 20

	skip_list.insert('Some Key'				,	10				)
	skip_list.insert('Key2'				,	8				)
	skip_list.insert('V'				,	13				)

	assert skip_list.find('Y'				) is None
	assert skip_list.find('Key2'				) == 8
	assert skip_list.find('Some Key'				) == 10
	assert skip_list.find('V'				) == 13


def 							SCREAMING_SNAKE_CASE__  (      ):
	snake_case_						:     Union[str, Any]     =		SkipList()
	skip_list.delete('Some key'				)

	assert len(skip_list.head.forward				) == 0


def 							SCREAMING_SNAKE_CASE__  (      ):
	snake_case_						:     Dict     =		SkipList()

	skip_list.insert('Key1'				,	12				)
	skip_list.insert('V'				,	13				)
	skip_list.insert('X'				,	14				)
	skip_list.insert('Key2'				,	15				)

	skip_list.delete('V'				)
	skip_list.delete('Key2'				)

	assert skip_list.find('V'				) is None
	assert skip_list.find('Key2'				) is None


def 							SCREAMING_SNAKE_CASE__  (      ):
	snake_case_						:     Tuple     =		SkipList()

	skip_list.insert('Key1'				,	12				)
	skip_list.insert('V'				,	13				)
	skip_list.insert('X'				,	14				)
	skip_list.insert('Key2'				,	15				)

	skip_list.delete('V'				)
	assert skip_list.find('V'				) is None
	assert skip_list.find('X'				) == 14
	assert skip_list.find('Key1'				) == 12
	assert skip_list.find('Key2'				) == 15

	skip_list.delete('X'				)
	assert skip_list.find('V'				) is None
	assert skip_list.find('X'				) is None
	assert skip_list.find('Key1'				) == 12
	assert skip_list.find('Key2'				) == 15

	skip_list.delete('Key1'				)
	assert skip_list.find('V'				) is None
	assert skip_list.find('X'				) is None
	assert skip_list.find('Key1'				) is None
	assert skip_list.find('Key2'				) == 15

	skip_list.delete('Key2'				)
	assert skip_list.find('V'				) is None
	assert skip_list.find('X'				) is None
	assert skip_list.find('Key1'				) is None
	assert skip_list.find('Key2'				) is None


def 							SCREAMING_SNAKE_CASE__  (      ):
	snake_case_						:     Optional[Any]     =		SkipList()

	skip_list.insert('Key1'				,	12				)
	skip_list.insert('V'				,	13				)
	skip_list.insert('X'				,	1_42				)
	skip_list.insert('Key2'				,	15				)

	skip_list.delete('X'				)

	def traverse_keys(__a				):
		yield node.key
		for forward_node in node.forward:
			yield from traverse_keys(__a				)

	assert len(set(traverse_keys(skip_list.head				)				)				) == 4


def 							SCREAMING_SNAKE_CASE__  (      ):
	def is_sorted(__a				):
		return all(next_item >= item for item, next_item in zip(__a				,	lst[1:]				)				)

	snake_case_						:     Union[str, Any]     =		SkipList()
	for i in range(10				):
		skip_list.insert(__a				,	__a				)
	assert is_sorted(list(__a				)				)
	skip_list.delete(5				)
	skip_list.delete(8				)
	skip_list.delete(2				)
	assert is_sorted(list(__a				)				)
	skip_list.insert(-12				,	-12				)
	skip_list.insert(77				,	77				)
	assert is_sorted(list(__a				)				)


def 							SCREAMING_SNAKE_CASE__  (      ):
	for _ in range(1_00				):
		# Repeat test 100 times due to the probabilistic nature of skip list
		# random values == random bugs
		test_insert()
		test_insert_overrides_existing_value()

		test_searching_empty_list_returns_none()
		test_search()

		test_deleting_item_from_empty_list_do_nothing()
		test_deleted_items_are_not_founded_by_find_method()
		test_delete_removes_only_given_key()
		test_delete_doesnt_leave_dead_nodes()

		test_iter_always_yields_sorted_values()


def 							SCREAMING_SNAKE_CASE__  (      ):
	snake_case_						:     Union[str, Any]     =		SkipList()
	skip_list.insert(2				,	'2'				)
	skip_list.insert(4				,	'4'				)
	skip_list.insert(6				,	'4'				)
	skip_list.insert(4				,	'5'				)
	skip_list.insert(8				,	'4'				)
	skip_list.insert(9				,	'4'				)

	skip_list.delete(4				)

	print(__a				)


if __name__ == "__main__":
						import doctest

						doctest.testmod()
						main()








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


_SCREAMING_SNAKE_CASE 		=						["""\nclass""", """\ndef""", """\n#""", """\n@""", """\nprint""", """\nif"""]
class SCREAMING_SNAKE_CASE_  (   snake_case_   ):






						def __init__( self : Tuple						,				_A : Any						,				_A : str						,				_A : int=None						,				_A : str=1		)							->      List[str]:







							"""simple docstring"""


							snake_case_						:     Union[str, Any]     =		tokenizer
							snake_case_						:     Optional[int]     =		dataset
							snake_case_						:     List[str]     =		len(_A		) if n_tasks is None else n_tasks
							snake_case_						:     List[Any]     =		n_copies






						def __iter__( self : Any		)							->      List[str]:







							"""simple docstring"""


							snake_case_						:     List[str]     =		[]
							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(_A						,				padding=_A						,				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 SCREAMING_SNAKE_CASE_  (   snake_case_   ):






						def __init__( self : List[Any]						,				_A : Optional[int]						,				_A : str						,				_A : Dict		)							->      Any:







							"""simple docstring"""


							snake_case_						:     List[str]     =		start_length
							snake_case_						:     int     =		eof_strings
							snake_case_						:     Dict     =		tokenizer






						def __call__( self : Any						,				_A : Union[str, Any]						,				_A : Dict						,				**_A : List[Any]		)							->      List[str]:







							"""simple docstring"""


							snake_case_						:     Optional[int]     =		self.tokenizer.batch_decode(input_ids[:, self.start_length :]		)
							snake_case_						:     Optional[int]     =		[]
							for decoded_generation in decoded_generations:
								done.append(any(stop_string in decoded_generation for stop_string in self.eof_strings		)		)
							return all(_A		)


def 							SCREAMING_SNAKE_CASE__  (      __a				):
	snake_case_						:     List[str]     =		re.split('(%s)' % '|'.join(__a				)				,	__a				)
	# last string should be ""
	return "".join(string_list[:-2]				)


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

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

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

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


def 							SCREAMING_SNAKE_CASE__  (      ):
	# Setup configuration
	snake_case_						:     Optional[int]     =		HfArgumentParser(__a				)
	snake_case_						:     Tuple     =		parser.parse_args()

	transformers.logging.set_verbosity_error()
	# enables code execution in code_eval metric
	snake_case_						:     Optional[int]     =		args.HF_ALLOW_CODE_EVAL
	# make sure tokenizer plays nice with multiprocessing
	snake_case_						:     int     =		'false'

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

	# Use dataset load to feed to accelerate
	snake_case_						:     List[Any]     =		Accelerator()
	set_seed(args.seed				,	device_specific=__a				)

	# Load model and tokenizer
	snake_case_						:     Any     =		AutoTokenizer.from_pretrained(args.model_ckpt				)
	snake_case_						:     int     =		tokenizer.eos_token
	snake_case_						:     List[str]     =		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				,	__a				,	__a				)]				),
	}

	# Load evaluation dataset and metric
	snake_case_						:     Dict     =		load_dataset('openai_humaneval'				)
	snake_case_						:     Optional[Any]     =		load_metric('code_eval'				)

	snake_case_						:     Union[str, Any]     =		args.num_tasks if args.num_tasks is not None else len(human_eval['test']				)
	snake_case_						:     Optional[int]     =		args.n_samples // args.batch_size

	snake_case_						:     Dict     =		TokenizedDataset(__a				,	human_eval['test']				,	n_copies=__a				,	n_tasks=__a				)
	# do not confuse args.batch_size, which is actually the num_return_sequences
	snake_case_						:     Optional[Any]     =		DataLoader(__a				,	batch_size=1				)

	# Run a quick test to see if code evaluation is enabled
	try:
		snake_case_						:     Union[str, 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_						:     Union[str, Any]     =		accelerator.prepare(__a				,	__a				)

	snake_case_						:     str     =		complete_code(
	    __a				,	__a				,	__a				,	__a				,	n_tasks=__a				,	batch_size=args.batch_size				,	**__a				,	)

	if accelerator.is_main_process:
		snake_case_						:     Tuple     =		[]

		for task in tqdm(range(__a				)				):
			snake_case_						:     Union[str, Any]     =		human_eval['test'][task]['test']
			snake_case_						:     Union[str, Any]     =		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_						:     int     =		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()

