




import logging
from pathlib import Path

import numpy as np
import pytorch_lightning as pl
import torch
from pytorch_lightning.callbacks import EarlyStopping, ModelCheckpoint
from pytorch_lightning.utilities import rank_zero_only

from utils import save_json


def 						lowercase		(     _lowerCAmelCase      ):
  UpperCAmelCase__    						=   filter(lambda _lowerCAmelCase      : p.requires_grad   ,     model.parameters()      )
  UpperCAmelCase__    						=   sum([np.prod(p.size()      ) for p in model_parameters]      )
  return params


snake_case__ :						Optional[Any]     =			logging.getLogger(__name__)





class     snake_case  (	pl.Callback							):


   '''simple docstring'''







   def 	UpperCAmelCase       (					self     :	Dict						,    lowerCamelCase_     :	Any						,    lowerCamelCase_     :	Any			)					->Dict:
     '''simple docstring'''







     UpperCAmelCase__    						=   {f'''lr_group_{i}''': param["""lr"""] for i, param in enumerate(pl_module.trainer.optimizers[0].param_groups			)}
     pl_module.logger.log_metrics(lowerCamelCase_			)




   @rank_zero_only
   def 	UpperCAmelCase       (					self     :	List[Any]						,    lowerCamelCase_     :	pl.Trainer						,    lowerCamelCase_     :	pl.LightningModule						,    lowerCamelCase_     :	str						,    lowerCamelCase_     :	Tuple=True			)					->None:
     '''simple docstring'''







     logger.info(f'''***** {type_path} results at step {trainer.global_step:05d} *****'''			)
     UpperCAmelCase__    						=   trainer.callback_metrics
     trainer.logger.log_metrics({k: v for k, v in metrics.items() if k not in ["""log""", """progress_bar""", """preds"""]}			)
     # Log results
     UpperCAmelCase__    						=   Path(pl_module.hparams.output_dir			)
     if type_path == "test":
       UpperCAmelCase__    						=   od / """test_results.txt"""
       UpperCAmelCase__    						=   od / """test_generations.txt"""
     else:
       # this never gets hit. I prefer not to save intermediate generations, and results are in metrics.json
       # If people want this it will be easy enough to add back.
       UpperCAmelCase__    						=   od / f'''{type_path}_results/{trainer.global_step:05d}.txt'''
       UpperCAmelCase__    						=   od / f'''{type_path}_generations/{trainer.global_step:05d}.txt'''
       results_file.parent.mkdir(exist_ok=lowerCamelCase_			)
       generations_file.parent.mkdir(exist_ok=lowerCamelCase_			)
     with open(lowerCamelCase_						,    """a+"""			) as writer:
       for key in sorted(lowerCamelCase_			):
         if key in ["log", "progress_bar", "preds"]:
           continue
         UpperCAmelCase__    						=   metrics[key]
         if isinstance(lowerCamelCase_						,    torch.Tensor			):
           UpperCAmelCase__    						=   val.item()
         UpperCAmelCase__    						=   f'''{key}: {val:.6f}\n'''
         writer.write(lowerCamelCase_			)

     if not save_generations:
       return

     if "preds" in metrics:
       UpperCAmelCase__    						=   """\n""".join(metrics["""preds"""]			)
       generations_file.open("""w+"""			).write(lowerCamelCase_			)




   @rank_zero_only
   def 	UpperCAmelCase       (					self     :	Dict						,    lowerCamelCase_     :	Optional[int]						,    lowerCamelCase_     :	List[str]			)					->List[Any]:
     '''simple docstring'''







     try:
       UpperCAmelCase__    						=   pl_module.model.model.num_parameters()
     except AttributeError:
       UpperCAmelCase__    						=   pl_module.model.num_parameters()

     UpperCAmelCase__    						=   count_trainable_parameters(lowerCamelCase_			)
     # mp stands for million parameters
     trainer.logger.log_metrics({"""n_params""": npars, """mp""": npars / 1E6, """grad_mp""": n_trainable_pars / 1E6}			)




   @rank_zero_only
   def 	UpperCAmelCase       (					self     :	Optional[int]						,    lowerCamelCase_     :	pl.Trainer						,    lowerCamelCase_     :	pl.LightningModule			)					->Optional[Any]:
     '''simple docstring'''







     save_json(pl_module.metrics						,    pl_module.metrics_save_path			)
     return self._write_logs(lowerCamelCase_						,    lowerCamelCase_						,    """test"""			)







   @rank_zero_only
   def 	UpperCAmelCase       (					self     :	List[Any]						,    lowerCamelCase_     :	pl.Trainer						,    lowerCamelCase_     :	int			)					->str:
     '''simple docstring'''







     save_json(pl_module.metrics						,    pl_module.metrics_save_path			)
     # Uncommenting this will save val generations
     # return self._write_logs(trainer, pl_module, "valid")



def 						lowercase		(     _lowerCAmelCase   ,     _lowerCAmelCase   ,     _lowerCAmelCase=1   ,     _lowerCAmelCase=False      ):
  if metric == "rouge2":
    UpperCAmelCase__    						=   """{val_avg_rouge2:.4f}-{step_count}"""
  elif metric == "bleu":
    UpperCAmelCase__    						=   """{val_avg_bleu:.4f}-{step_count}"""
  elif metric == "loss":
    UpperCAmelCase__    						=   """{val_avg_loss:.4f}-{step_count}"""
  else:
    raise NotImplementedError(
        F'''seq2seq callbacks only support rouge2, bleu and loss, got {metric}, You can make your own by adding to'''
        """ this function."""      )

  UpperCAmelCase__    						=   ModelCheckpoint(
      dirpath=_lowerCAmelCase   ,     filename=_lowerCAmelCase   ,     monitor=F'''val_{metric}'''   ,     mode="""min""" if """loss""" in metric else """max"""   ,     save_top_k=_lowerCAmelCase   ,     )
  return checkpoint_callback





def 						lowercase		(     _lowerCAmelCase   ,     _lowerCAmelCase      ):
  return EarlyStopping(
      monitor=F'''val_{metric}'''   ,     mode="""min""" if """loss""" in metric else """max"""   ,     patience=_lowerCAmelCase   ,     verbose=_lowerCAmelCase   ,     )










import json
import os
from functools import lru_cache
from typing import TYPE_CHECKING, List, Optional, Tuple, Union

import numpy as np
import regex as re

from ...utils import is_tf_available, is_torch_available, logging


if TYPE_CHECKING:
   if is_torch_available():
      import torch
   if is_tf_available():
      import tensorflow as tf

from ...tokenization_utils import AddedToken, PreTrainedTokenizer


snake_case__ :						str     =			logging.get_logger(__name__)

snake_case__ :						List[str]     =			{
    '''vocab_file''': '''vocab.json''',
    '''merges_file''': '''merges.txt''',
}

snake_case__ :						Any     =			{
    '''vocab_file''': {
        '''Salesforce/codegen-350M-mono''': '''https://huggingface.co/Salesforce/codegen-350M-mono/resolve/main/vocab.json''',
    },
    '''merges_file''': {
        '''Salesforce/codegen-350M-mono''': '''https://huggingface.co/Salesforce/codegen-350M-mono/resolve/main/merges.txt''',
    },
}

snake_case__ :						Tuple     =			{
    '''Salesforce/codegen-350M-mono''': 2_0_4_8,
}


@lru_cache()
def 						lowercase		(     ):
  UpperCAmelCase__    						=   (
      list(range(ord("""!"""      )   ,     ord("""~"""      ) + 1      )      ) + list(range(ord("""¡"""      )   ,     ord("""¬"""      ) + 1      )      ) + list(range(ord("""®"""      )   ,     ord("""ÿ"""      ) + 1      )      )
  )
  UpperCAmelCase__    						=   bs[:]
  UpperCAmelCase__    						=   0
  for b in range(2**8      ):
    if b not in bs:
      bs.append(_lowerCAmelCase      )
      cs.append(2**8 + n      )
      n += 1
  UpperCAmelCase__    						=   [chr(_lowerCAmelCase      ) for n in cs]
  return dict(zip(_lowerCAmelCase   ,     _lowerCAmelCase      )      )


def 						lowercase		(     _lowerCAmelCase      ):
  UpperCAmelCase__    						=   set()
  UpperCAmelCase__    						=   word[0]
  for char in word[1:]:
    pairs.add((prev_char, char)      )
    UpperCAmelCase__    						=   char
  return pairs





class     snake_case  (	_snake_case							):


   '''simple docstring'''







   UpperCamelCase__	:       List[Any]     =      VOCAB_FILES_NAMES
   UpperCamelCase__	:       Union[str, Any]     =      PRETRAINED_VOCAB_FILES_MAP
   UpperCamelCase__	:       int     =      PRETRAINED_POSITIONAL_EMBEDDINGS_SIZES
   UpperCamelCase__	:       int     =      ["input_ids", "attention_mask"]




   def __init__(					self     :	Union[str, Any]						,    lowerCamelCase_     :	Tuple						,    lowerCamelCase_     :	Any						,    lowerCamelCase_     :	Union[str, Any]="replace"						,    lowerCamelCase_     :	List[str]="<|endoftext|>"						,    lowerCamelCase_     :	Dict="<|endoftext|>"						,    lowerCamelCase_     :	List[str]="<|endoftext|>"						,    lowerCamelCase_     :	str=None						,    lowerCamelCase_     :	Any=False						,    lowerCamelCase_     :	Optional[int]=False						,    **lowerCamelCase_     :	str						,    )					->Dict:
     '''simple docstring'''







     UpperCAmelCase__    						=   AddedToken(lowerCamelCase_						,    lstrip=lowerCamelCase_						,    rstrip=lowerCamelCase_			) if isinstance(lowerCamelCase_						,    lowerCamelCase_			) else bos_token
     UpperCAmelCase__    						=   AddedToken(lowerCamelCase_						,    lstrip=lowerCamelCase_						,    rstrip=lowerCamelCase_			) if isinstance(lowerCamelCase_						,    lowerCamelCase_			) else eos_token
     UpperCAmelCase__    						=   AddedToken(lowerCamelCase_						,    lstrip=lowerCamelCase_						,    rstrip=lowerCamelCase_			) if isinstance(lowerCamelCase_						,    lowerCamelCase_			) else unk_token
     UpperCAmelCase__    						=   AddedToken(lowerCamelCase_						,    lstrip=lowerCamelCase_						,    rstrip=lowerCamelCase_			) if isinstance(lowerCamelCase_						,    lowerCamelCase_			) else pad_token
     super().__init__(
         errors=lowerCamelCase_						,    unk_token=lowerCamelCase_						,    bos_token=lowerCamelCase_						,    eos_token=lowerCamelCase_						,    pad_token=lowerCamelCase_						,    add_prefix_space=lowerCamelCase_						,    add_bos_token=lowerCamelCase_						,    **lowerCamelCase_						,    )
     UpperCAmelCase__    						=   add_bos_token

     with open(lowerCamelCase_						,    encoding="""utf-8"""			) as vocab_handle:
       UpperCAmelCase__    						=   json.load(lowerCamelCase_			)
     UpperCAmelCase__    						=   {v: k for k, v in self.encoder.items()}
     UpperCAmelCase__    						=   errors  # how to handle errors in decoding
     UpperCAmelCase__    						=   bytes_to_unicode()
     UpperCAmelCase__    						=   {v: k for k, v in self.byte_encoder.items()}
     with open(lowerCamelCase_						,    encoding="""utf-8"""			) as merges_handle:
       UpperCAmelCase__    						=   merges_handle.read().split("""\n"""			)[1:-1]
     UpperCAmelCase__    						=   [tuple(merge.split()			) for merge in bpe_merges]
     UpperCAmelCase__    						=   dict(zip(lowerCamelCase_						,    range(len(lowerCamelCase_			)			)			)			)
     UpperCAmelCase__    						=   {}
     UpperCAmelCase__    						=   add_prefix_space

     # Should have added re.IGNORECASE so BPE merges can happen for capitalized versions of contractions
     UpperCAmelCase__    						=   re.compile(R"""'s|'t|'re|'ve|'m|'ll|'d| ?\p{L}+| ?\p{N}+| ?[^\s\p{L}\p{N}]+|\s+(?!\S)|\s+"""			)




   @property
   def 	UpperCAmelCase       (					self     :	Dict			)					->Any:
     '''simple docstring'''







     return len(self.encoder			)




   def 	UpperCAmelCase       (					self     :	Dict			)					->Optional[int]:
     '''simple docstring'''







     return dict(self.encoder						,    **self.added_tokens_encoder			)




   def 	UpperCAmelCase       (					self     :	Tuple						,    lowerCamelCase_     :	List[str]			)					->Optional[Any]:
     '''simple docstring'''







     if token in self.cache:
       return self.cache[token]
     UpperCAmelCase__    						=   tuple(lowerCamelCase_			)
     UpperCAmelCase__    						=   get_pairs(lowerCamelCase_			)

     if not pairs:
       return token

     while True:
       UpperCAmelCase__    						=   min(lowerCamelCase_						,    key=lambda lowerCamelCase_			: self.bpe_ranks.get(lowerCamelCase_						,    float("""inf"""			)			)			)
       if bigram not in self.bpe_ranks:
         break
       UpperCAmelCase__						,					UpperCAmelCase__    						=   bigram
       UpperCAmelCase__    						=   []
       UpperCAmelCase__    						=   0
       while i < len(lowerCamelCase_			):
         try:
           UpperCAmelCase__    						=   word.index(lowerCamelCase_						,    lowerCamelCase_			)
         except ValueError:
           new_word.extend(word[i:]			)
           break
         else:
           new_word.extend(word[i:j]			)
           UpperCAmelCase__    						=   j

         if word[i] == first and i < len(lowerCamelCase_			) - 1 and word[i + 1] == second:
           new_word.append(first + second			)
           i += 2
         else:
           new_word.append(word[i]			)
           i += 1
       UpperCAmelCase__    						=   tuple(lowerCamelCase_			)
       UpperCAmelCase__    						=   new_word
       if len(lowerCamelCase_			) == 1:
         break
       else:
         UpperCAmelCase__    						=   get_pairs(lowerCamelCase_			)
     UpperCAmelCase__    						=   """ """.join(lowerCamelCase_			)
     UpperCAmelCase__    						=   word
     return word




   def 	UpperCAmelCase       (					self     :	int						,    lowerCamelCase_     :	Optional[int]						,    lowerCamelCase_     :	Optional[Any]=None			)					->str:
     '''simple docstring'''







     if self.add_bos_token:
       UpperCAmelCase__    						=   [self.bos_token_id]
     else:
       UpperCAmelCase__    						=   []

     UpperCAmelCase__    						=   bos_token_ids + token_ids_a

     if token_ids_a is None:
       return output

     return output + bos_token_ids + token_ids_a




   def 	UpperCAmelCase       (					self     :	Any						,    lowerCamelCase_     :	Optional[Any]			)					->Union[str, Any]:
     '''simple docstring'''







     UpperCAmelCase__    						=   []
     for token in re.findall(self.pat						,    lowerCamelCase_			):
       UpperCAmelCase__    						=   """""".join(
           self.byte_encoder[b] for b in token.encode("""utf-8"""			)			)  # Maps all our bytes to unicode strings, avoiding control tokens of the BPE (spaces in our case)
       bpe_tokens.extend(bpe_token for bpe_token in self.bpe(lowerCamelCase_			).split(""" """			)			)
     return bpe_tokens




   def 	UpperCAmelCase       (					self     :	Dict						,    lowerCamelCase_     :	Optional[Any]			)					->int:
     '''simple docstring'''







     return self.encoder.get(lowerCamelCase_						,    self.encoder.get(self.unk_token			)			)




   def 	UpperCAmelCase       (					self     :	Union[str, Any]						,    lowerCamelCase_     :	Dict			)					->Union[str, Any]:
     '''simple docstring'''







     return self.decoder.get(lowerCamelCase_			)




   def 	UpperCAmelCase       (					self     :	int						,    lowerCamelCase_     :	Union[str, Any]			)					->Union[str, Any]:
     '''simple docstring'''







     UpperCAmelCase__    						=   """""".join(lowerCamelCase_			)
     UpperCAmelCase__    						=   bytearray([self.byte_decoder[c] for c in text]			).decode("""utf-8"""						,    errors=self.errors			)
     return text




   def 	UpperCAmelCase       (					self     :	Dict						,    lowerCamelCase_     :	str						,    lowerCamelCase_     :	Optional[str] = None			)					->Tuple[str]:
     '''simple docstring'''







     if not os.path.isdir(lowerCamelCase_			):
       logger.error(f'''Vocabulary path ({save_directory}) should be a directory'''			)
       return
     UpperCAmelCase__    						=   os.path.join(
         lowerCamelCase_						,    (filename_prefix + """-""" if filename_prefix else """""") + VOCAB_FILES_NAMES["""vocab_file"""]			)
     UpperCAmelCase__    						=   os.path.join(
         lowerCamelCase_						,    (filename_prefix + """-""" if filename_prefix else """""") + VOCAB_FILES_NAMES["""merges_file"""]			)

     with open(lowerCamelCase_						,    """w"""						,    encoding="""utf-8"""			) as f:
       f.write(json.dumps(self.encoder						,    indent=2						,    sort_keys=lowerCamelCase_						,    ensure_ascii=lowerCamelCase_			) + """\n"""			)

     UpperCAmelCase__    						=   0
     with open(lowerCamelCase_						,    """w"""						,    encoding="""utf-8"""			) as writer:
       writer.write("""#version: 0.2\n"""			)
       for bpe_tokens, token_index in sorted(self.bpe_ranks.items()						,    key=lambda lowerCamelCase_			: kv[1]			):
         if index != token_index:
           logger.warning(
               f'''Saving vocabulary to {merge_file}: BPE merge indices are not consecutive.'''
               """ Please check that the tokenizer is not corrupted!"""			)
           UpperCAmelCase__    						=   token_index
         writer.write(""" """.join(lowerCamelCase_			) + """\n"""			)
         index += 1

     return vocab_file, merge_file




   def 	UpperCAmelCase       (					self     :	List[Any]						,    lowerCamelCase_     :	int						,    lowerCamelCase_     :	Tuple=False						,    **lowerCamelCase_     :	Any			)					->Any:
     '''simple docstring'''







     UpperCAmelCase__    						=   kwargs.pop("""add_prefix_space"""						,    self.add_prefix_space			)
     if is_split_into_words or add_prefix_space:
       UpperCAmelCase__    						=   """ """ + text
     return (text, kwargs)




   def 	UpperCAmelCase       (					self     :	List[Any]						,    lowerCamelCase_     :	Union[int, List[int], "np.ndarray", "torch.Tensor", "tf.Tensor"]						,    lowerCamelCase_     :	bool = False						,    lowerCamelCase_     :	bool = None						,    lowerCamelCase_     :	Optional[List[str]] = None						,    **lowerCamelCase_     :	Any						,    )					->str:
     '''simple docstring'''







     UpperCAmelCase__    						=   super()._decode(
         token_ids=lowerCamelCase_						,    skip_special_tokens=lowerCamelCase_						,    clean_up_tokenization_spaces=lowerCamelCase_						,    **lowerCamelCase_						,    )

     if truncate_before_pattern is not None and len(lowerCamelCase_			) > 0:
       UpperCAmelCase__    						=   self.truncate(lowerCamelCase_						,    lowerCamelCase_			)

     return decoded_text







   def 	UpperCAmelCase       (					self     :	Tuple						,    lowerCamelCase_     :	Any						,    lowerCamelCase_     :	str			)					->Optional[int]:
     '''simple docstring'''







     def find_re(lowerCamelCase_     :	List[str]						,    lowerCamelCase_     :	Union[str, Any]						,    lowerCamelCase_     :	int			):
       UpperCAmelCase__    						=   pattern.search(lowerCamelCase_						,    lowerCamelCase_			)
       return m.start() if m else -1

     UpperCAmelCase__    						=   [re.compile(lowerCamelCase_						,    re.MULTILINE			) for pattern in truncate_before_pattern]

     UpperCAmelCase__    						=   list(re.finditer("""^print"""						,    lowerCamelCase_						,    re.MULTILINE			)			)

     if len(lowerCamelCase_			) > 1:
       UpperCAmelCase__    						=   completion[: prints[1].start()]

     UpperCAmelCase__    						=   list(re.finditer("""^def"""						,    lowerCamelCase_						,    re.MULTILINE			)			)

     if len(lowerCamelCase_			) > 1:
       UpperCAmelCase__    						=   completion[: defs[1].start()]

     UpperCAmelCase__    						=   0

     UpperCAmelCase__    						=   [
         pos for pos in [find_re(lowerCamelCase_						,    lowerCamelCase_						,    lowerCamelCase_			) for terminal in terminals] if pos != -1
     ]

     if len(lowerCamelCase_			) > 0:
       return completion[: min(lowerCamelCase_			)]
     else:
       return completion




