



'''simple docstring'''





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_rag import save_json
def        _SCREAMING_SNAKE_CASE  (	UpperCamelCase      ):



    """simple docstring"""





    lowerCAmelCase__			:  Tuple											= filter(lambda UpperCamelCase      : p.requires_grad							,   model.parameters()      )
    lowerCAmelCase__			:  Optional[Any]											= sum([np.prod(p.size()      ) for p in model_parameters]      )
    return params


_lowerCAmelCase										=     logging.getLogger(__name__)
def        _SCREAMING_SNAKE_CASE  (	UpperCamelCase							,   UpperCamelCase      ):



    """simple docstring"""





    if metric == "rouge2":
        lowerCAmelCase__			:  Dict											= """{val_avg_rouge2:.4f}-{step_count}"""
    elif metric == "bleu":
        lowerCAmelCase__			:  str											= """{val_avg_bleu:.4f}-{step_count}"""
    elif metric == "em":
        lowerCAmelCase__			:  Tuple											= """{val_avg_em:.4f}-{step_count}"""
    elif metric == "loss":
        lowerCAmelCase__			:  List[Any]											= """{val_avg_loss:.4f}-{step_count}"""
    else:
        raise NotImplementedError(
            f"""seq2seq callbacks only support rouge2 and bleu, got {metric}, You can make your own by adding to this"""
            """ function."""      )

    lowerCAmelCase__			:  Tuple											= ModelCheckpoint(
        dirpath=UpperCamelCase							,   filename=UpperCamelCase							,   monitor=f"""val_{metric}"""							,   mode="""max"""							,   save_top_k=1							,   every_n_epochs=1							,   )

    return checkpoint_callback
def        _SCREAMING_SNAKE_CASE  (	UpperCamelCase							,   UpperCamelCase      ):



    """simple docstring"""





    return EarlyStopping(
        monitor=f"""val_{metric}"""							,   mode="""min""" if """loss""" in metric else """max"""							,   patience=UpperCamelCase							,   verbose=UpperCamelCase							,   )






class 	lowerCAmelCase_(    pl.Callback					):
  '''simple docstring'''




  def 							UpperCAmelCase_     (   self  ,__UpperCAmelCase  ,__UpperCAmelCase				)   ->		str:
      lowerCAmelCase__			:  Tuple											= {F"""lr_group_{i}""": param["""lr"""] for i, param in enumerate(pl_module.trainer.optimizers[0].param_groups				)}
      pl_module.logger.log_metrics(__UpperCAmelCase				)




  @rank_zero_only
  def 							UpperCAmelCase_     (   self  ,__UpperCAmelCase  ,__UpperCAmelCase  ,__UpperCAmelCase  ,__UpperCAmelCase=True				)   ->		None:
      logger.info(F"""***** {type_path} results at step {trainer.global_step:05d} *****"""				)
      lowerCAmelCase__			:  Tuple											= 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
      lowerCAmelCase__			:  Optional[Any]											= Path(pl_module.hparams.output_dir				)
      if type_path == "test":
          lowerCAmelCase__			:  Union[str, Any]											= od / """test_results.txt"""
          lowerCAmelCase__			:  int											= 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.
          lowerCAmelCase__			:  Optional[int]											= od / F"""{type_path}_results/{trainer.global_step:05d}.txt"""
          lowerCAmelCase__			:  int											= od / F"""{type_path}_generations/{trainer.global_step:05d}.txt"""
          results_file.parent.mkdir(exist_ok=__UpperCAmelCase				)
          generations_file.parent.mkdir(exist_ok=__UpperCAmelCase				)
      with open(__UpperCAmelCase  ,"""a+"""				) as writer:
          for key in sorted(__UpperCAmelCase				):
              if key in ["log", "progress_bar", "preds"]:
                  continue
              lowerCAmelCase__			:  Dict											= metrics[key]
              if isinstance(__UpperCAmelCase  ,torch.Tensor				):
                  lowerCAmelCase__			:  Any											= val.item()
              lowerCAmelCase__			:  Union[str, Any]											= F"""{key}: {val:.6f}\n"""
              writer.write(__UpperCAmelCase				)

      if not save_generations:
          return

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




  @rank_zero_only
  def 							UpperCAmelCase_     (   self  ,__UpperCAmelCase  ,__UpperCAmelCase				)   ->		List[str]:
      try:
          lowerCAmelCase__			:  int											= pl_module.model.model.num_parameters()
      except AttributeError:
          lowerCAmelCase__			:  str											= pl_module.model.num_parameters()

      lowerCAmelCase__			:  Union[str, Any]											= count_trainable_parameters(__UpperCAmelCase				)
      # 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  ,__UpperCAmelCase  ,__UpperCAmelCase				)   ->		Tuple:
      save_json(pl_module.metrics  ,pl_module.metrics_save_path				)
      return self._write_logs(__UpperCAmelCase  ,__UpperCAmelCase  ,"""test"""				)






  @rank_zero_only
  def 							UpperCAmelCase_     (   self  ,__UpperCAmelCase  ,__UpperCAmelCase				)   ->		Tuple:
      save_json(pl_module.metrics  ,pl_module.metrics_save_path				)
      # Uncommenting this will save val generations
      # return self._write_logs(trainer, pl_module, "valid")







'''simple docstring'''





import os
def        _SCREAMING_SNAKE_CASE  (	UpperCamelCase      ):



    """simple docstring"""





    lowerCAmelCase__			:  List[str]											= len(grid[0]      )
    lowerCAmelCase__			:  int											= len(UpperCamelCase      )

    lowerCAmelCase__			:  Optional[int]											= 0
    lowerCAmelCase__			:  Tuple											= 0
    lowerCAmelCase__			:  Optional[Any]											= 0

    # Check vertically, horizontally, diagonally at the same time (only works
    # for nxn grid)
    for i in range(UpperCamelCase      ):
        for j in range(n_rows - 3      ):
            lowerCAmelCase__			:  str											= grid[j][i] * grid[j + 1][i] * grid[j + 2][i] * grid[j + 3][i]
            lowerCAmelCase__			:  Optional[int]											= grid[i][j] * grid[i][j + 1] * grid[i][j + 2] * grid[i][j + 3]

            # Left-to-right diagonal (\) product
            if i < n_columns - 3:
                lowerCAmelCase__			:  Optional[int]											= (
                    grid[i][j]
                    * grid[i + 1][j + 1]
                    * grid[i + 2][j + 2]
                    * grid[i + 3][j + 3]
                )

            # Right-to-left diagonal(/) product
            if i > 2:
                lowerCAmelCase__			:  Tuple											= (
                    grid[i][j]
                    * grid[i - 1][j + 1]
                    * grid[i - 2][j + 2]
                    * grid[i - 3][j + 3]
                )

            lowerCAmelCase__			:  Dict											= max(
                UpperCamelCase							,   UpperCamelCase							,   UpperCamelCase							,   UpperCamelCase      )
            if max_product > largest:
                lowerCAmelCase__			:  Any											= max_product

    return largest
def        _SCREAMING_SNAKE_CASE  (	):



    """simple docstring"""





    lowerCAmelCase__			:  List[str]											= []
    with open(os.path.dirname(UpperCamelCase      ) + """/grid.txt"""      ) as file:
        for line in file:
            grid.append(line.strip("""\n"""      ).split(""" """      )      )

    lowerCAmelCase__			:  Dict											= [[int(UpperCamelCase      ) for i in grid[j]] for j in range(len(UpperCamelCase      )      )]

    return largest_product(UpperCamelCase      )


if __name__ == "__main__":
 print(solution())


