


"""simple docstring"""


def   snake_case						(				_a:     Optional[int]		,   _a:     int							)->		tuple[float, float]:
      '''simple docstring'''


      if not len(__UpperCamelCase							) == len(__UpperCamelCase							) == 3:
            raise ValueError('Please enter a valid equation.'							)
      if equationa[0] == equationa[1] == equationa[0] == equationa[1] == 0:
            raise ValueError('Both a & b of two equations can\'t be zero.'							)

      # Extract the coefficients
      lowerCamelCase__    ,  lowerCamelCase__    ,  lowerCamelCase__           =		equationa
      lowerCamelCase__    ,  lowerCamelCase__    ,  lowerCamelCase__           =		equationa

      # Calculate the determinants of the matrices
      lowerCamelCase__           =		aa * ba - aa * ba
      lowerCamelCase__           =		ca * ba - ca * ba
      lowerCamelCase__           =		aa * ca - aa * ca

      # Check if the system of linear equations has a solution (using Cramer's rule)
      if determinant == 0:
            if determinant_x == determinant_y == 0:
                  raise ValueError('Infinite solutions. (Consistent system)'							)
            else:
                  raise ValueError('No solution. (Inconsistent system)'							)
      else:
            if determinant_x == determinant_y == 0:
                  # Trivial solution (Inconsistent system)
                  return (0.0, 0.0)
            else:
                  lowerCamelCase__           =		determinant_x / determinant
                  lowerCamelCase__           =		determinant_y / determinant
                  # Non-Trivial Solution (Consistent system)
                  return (x, y)







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



