


import argparse
import logging
import os
import sys
import tempfile
from pathlib import Path

import lightning_base
import pytest
import pytorch_lightning as pl
import torch
from convert_pl_checkpoint_to_hf import convert_pl_to_hf
from distillation import distill_main
from finetune import SummarizationModule, main
from huggingface_hub import list_models
from parameterized import parameterized
from run_eval import generate_summaries_or_translations
from torch import nn

from transformers import AutoConfig, AutoModelForSeqaSeqLM
from transformers.testing_utils import CaptureStderr, CaptureStdout, TestCasePlus, require_torch_gpu, slow
from utils import label_smoothed_nll_loss, lmap, load_json


logging.basicConfig(level=logging.DEBUG)

UpperCAmelCase__            =							logging.getLogger()
UpperCAmelCase__            =							torch.cuda.is_available()
UpperCAmelCase__            =							{
    "max_tokens_per_batch": None,
    "supervise_forward": True,
    "normalize_hidden": True,
    "label_smoothing": 0.2,
    "eval_max_gen_length": None,
    "eval_beams": 1,
    "val_metric": "loss",
    "save_top_k": 1,
    "adafactor": True,
    "early_stopping_patience": 2,
    "logger_name": "default",
    "length_penalty": 0.5,
    "cache_dir": "",
    "task": "summarization",
    "num_workers": 2,
    "alpha_hid": 0,
    "freeze_embeds": True,
    "enc_only": False,
    "tgt_suffix": "",
    "resume_from_checkpoint": None,
    "sortish_sampler": True,
    "student_decoder_layers": 1,
    "val_check_interval": 1.0,
    "output_dir": "",
    "fp16": False,  # TODO(SS): set this to CUDA_AVAILABLE if ci installs apex or start using native amp
    "no_teacher": False,
    "fp16_opt_level": "O1",
    "gpus": 1 if CUDA_AVAILABLE else 0,
    "n_tpu_cores": 0,
    "max_grad_norm": 1.0,
    "do_train": True,
    "do_predict": True,
    "accumulate_grad_batches": 1,
    "server_ip": "",
    "server_port": "",
    "seed": 42,
    "model_name_or_path": "sshleifer/bart-tiny-random",
    "config_name": "",
    "tokenizer_name": "facebook/bart-large",
    "do_lower_case": False,
    "learning_rate": 0.3,
    "lr_scheduler": "linear",
    "weight_decay": 0.0,
    "adam_epsilon": 1e-08,
    "warmup_steps": 0,
    "max_epochs": 1,
    "train_batch_size": 2,
    "eval_batch_size": 2,
    "max_source_length": 12,
    "max_target_length": 12,
    "val_max_target_length": 12,
    "test_max_target_length": 12,
    "fast_dev_run": False,
    "no_cache": False,
    "n_train": -1,
    "n_val": -1,
    "n_test": -1,
    "student_encoder_layers": 1,
    "freeze_encoder": False,
    "auto_scale_batch_size": False,
    "overwrite_output_dir": False,
    "student": None,
}
def     UpperCAmelCase_ (							__snake_case					,						__snake_case						)					->      Tuple:







 """simple docstring"""


 _lowercase											='''\n'''.join(__lowercase						)
 Path(__lowercase						).open('''w'''						).writelines(__lowercase						)


UpperCAmelCase__            =							[" Sam ate lunch today.", "Sams lunch ingredients."]
UpperCAmelCase__            =							["A very interesting story about what I ate for lunch.", "Avocado, celery, turkey, coffee"]
UpperCAmelCase__            =							"patrickvonplaten/t5-tiny-random"
UpperCAmelCase__            =							"sshleifer/t5-tinier-random"
UpperCAmelCase__            =							"sshleifer/bart-tiny-random"
UpperCAmelCase__            =							"sshleifer/tiny-mbart"
UpperCAmelCase__            =							"sshleifer/tiny-marian-en-de"
UpperCAmelCase__            =							"stas/tiny-wmt19-en-de"


UpperCAmelCase__            =							logging.StreamHandler(sys.stdout)
logger.addHandler(stream_handler)
logging.disable(logging.CRITICAL)  # remove noisy download output from tracebacks
def     UpperCAmelCase_ (							__snake_case						)					->      Optional[Any]:







 """simple docstring"""


 for split in ["train", "val", "test"]:
  _dump_articles(os.path.join(__lowercase					,						F"{split}.source"						)					,						__lowercase						)
  _dump_articles(os.path.join(__lowercase					,						F"{split}.target"						)					,						__lowercase						)
 return tmp_dir







class 				lowerCamelCase__       (	lowercase_):





     @classmethod
     def 					__A (cls	)					->       str:
      logging.disable(logging.CRITICAL	)  # remove noisy download output from tracebacks
      return cls





     @slow
     @require_torch_gpu
     def 					__A (self	)					->       Union[str, Any]:
      _lowercase											=list_models()
      _lowercase											='''sshleifer'''
      _lowercase											=[x.modelId for x in model_list if x.modelId.startswith(UpperCamelCase_	)]
      _lowercase											=['''sshleifer/blenderbot-3B''', '''sshleifer/blenderbot-90M''']
      _lowercase											=[]
      for m in model_ids:
       if m in allowed_to_be_broken:
        continue
       try:
        AutoConfig.from_pretrained(UpperCamelCase_	)
       except Exception:
        failures.append(UpperCamelCase_	)
      assert not failures, f"The following models could not be loaded through AutoConfig: {failures}"





     def 					__A (self	)					->       List[str]:
      _lowercase											={'''student_encoder_layers''': 2, '''student_decoder_layers''': 1, '''no_teacher''': True}
      self._test_distiller_cli(UpperCamelCase_	)





     def 					__A (self	)					->       List[Any]:
      _lowercase											={
          '''student_encoder_layers''': 2,
          '''student_decoder_layers''': 1,
          '''max_epochs''': 4,
          '''val_check_interval''': 0.25,
          '''alpha_hid''': 2.0,
          '''model_name_or_path''': '''IGNORE_THIS_IT_DOESNT_GET_USED''',
      }
      _lowercase											=self._test_distiller_cli(UpperCamelCase_		,					check_contents=UpperCamelCase_	)

      _lowercase											=list(Path(model.output_dir	).glob('''*.ckpt'''	)	)
      self.assertEqual(1		,					len(UpperCamelCase_	)	)
      _lowercase											=list(Path(model.output_dir	).glob('''**/*.bin'''	)	)
      self.assertEqual(len(UpperCamelCase_	)		,					2	)
      _lowercase											=lmap(str.strip		,					Path(model.hparams.data_dir	).joinpath('''test.source'''	).open().readlines()	)
      _lowercase											=tempfile.mktemp()  # XXX: not being cleaned up
      generate_summaries_or_translations(UpperCamelCase_		,					UpperCamelCase_		,					str(model.output_dir / '''best_tfmr'''	)	)
      self.assertTrue(Path(UpperCamelCase_	).exists()	)

      _lowercase											=self.get_auto_remove_tmp_dir()
      convert_pl_to_hf(ckpts[0]		,					transformer_ckpts[0].parent		,					UpperCamelCase_	)
      assert os.path.exists(os.path.join(UpperCamelCase_		,					'''pytorch_model.bin'''	)	)





     def 					__A (self	)					->       int:
      _lowercase											=AutoModelForSeqaSeqLM.from_pretrained(UpperCamelCase_	)
      _lowercase				,   _lowercase											=model.dummy_inputs['''input_ids'''], model.dummy_inputs['''attention_mask''']
      _lowercase											=torch.tensor([[0, 4, 8, 2], [0, 8, 2, 1]]		,					dtype=torch.long		,					device=model.device	)
      _lowercase											=target_ids[:, :-1].contiguous()  # Why this line?
      _lowercase											=target_ids[:, 1:].clone()  # why clone?
      _lowercase											=model(
          UpperCamelCase_		,					attention_mask=UpperCamelCase_		,					decoder_input_ids=UpperCamelCase_		,					labels=UpperCamelCase_		,					use_cache=UpperCamelCase_	).loss

      _lowercase											=model(UpperCamelCase_		,					attention_mask=UpperCamelCase_		,					decoder_input_ids=UpperCamelCase_		,					use_cache=UpperCamelCase_	).logits

      _lowercase											=nn.functional.log_softmax(UpperCamelCase_		,					dim=-1	)
      _lowercase				,   _lowercase											=label_smoothed_nll_loss(
          UpperCamelCase_		,					UpperCamelCase_		,					0.1		,					ignore_index=model.config.pad_token_id	)
      with self.assertRaises(UpperCamelCase_	):
       # TODO: understand why this breaks
       self.assertEqual(UpperCamelCase_		,					UpperCamelCase_	)





     def 					__A (self	)					->       Optional[Any]:
      _lowercase											={
          '''student_encoder_layers''': 2,
          '''student_decoder_layers''': 1,
          '''num_train_epochs''': 4,
          '''val_check_interval''': 0.25,
          '''alpha_hid''': 2.0,
          '''task''': '''translation''',
          '''model_name_or_path''': '''IGNORE_THIS_IT_DOESNT_GET_USED''',
          '''tokenizer_name''': MBART_TINY,
          '''teacher''': MBART_TINY,
          '''src_lang''': '''en_XX''',
          '''tgt_lang''': '''ro_RO''',
      }
      _lowercase											=self._test_distiller_cli(UpperCamelCase_		,					check_contents=UpperCamelCase_	)
      assert model.model.config.model_type == "mbart"

      _lowercase											=list(Path(model.output_dir	).glob('''*.ckpt'''	)	)
      self.assertEqual(1		,					len(UpperCamelCase_	)	)
      _lowercase											=list(Path(model.output_dir	).glob('''**/*.bin'''	)	)
      _lowercase											=list(Path(model.output_dir	).glob('''best_tfmr/*'''	)	)
      assert len(UpperCamelCase_	) > 2
      self.assertEqual(len(UpperCamelCase_	)		,					2	)





     def 					__A (self	)					->       int:
      _lowercase											={
          '''student_encoder_layers''': 1,
          '''student_decoder_layers''': 1,
          '''alpha_hid''': 2.0,
          '''teacher''': T5_TINY,
          '''model_name_or_path''': T5_TINY,
          '''tokenizer_name''': T5_TINY,
      }
      self._test_distiller_cli(UpperCamelCase_	)





     def 					__A (self	)					->       Tuple:
      _lowercase											={
          '''teacher''': T5_TINY,
          '''student''': T5_TINIER,
          '''model_name_or_path''': T5_TINIER,
          '''tokenizer_name''': T5_TINIER,
      }
      self._test_distiller_cli(UpperCamelCase_	)





     def 					__A (self		,					UpperCAmelCase		,					UpperCAmelCase=True	)					->       int:
      _lowercase											={
          '''label_smoothing''': 0.0,
          '''early_stopping_patience''': -1,
          '''train_batch_size''': 1,
          '''eval_batch_size''': 2,
          '''max_epochs''': 2,
          '''alpha_mlm''': 0.2,
          '''alpha_ce''': 0.8,
          '''do_predict''': True,
          '''model_name_or_path''': '''sshleifer/tinier_bart''',
          '''teacher''': CHEAP_ARGS['''model_name_or_path'''],
          '''val_check_interval''': 0.5,
      }
      default_updates.update(UpperCamelCase_	)
      _lowercase											=CHEAP_ARGS.copy()
      _lowercase											=make_test_data_dir(tmp_dir=self.get_auto_remove_tmp_dir()	)
      _lowercase											=self.get_auto_remove_tmp_dir()

      args_d.update(data_dir=UpperCamelCase_		,					output_dir=UpperCamelCase_		,					**UpperCamelCase_	)
      _lowercase											=distill_main(argparse.Namespace(**UpperCamelCase_	)	)
      if not check_contents:
       return model
      _lowercase											=os.listdir(UpperCamelCase_	)
      _lowercase											={os.path.basename(UpperCamelCase_	) for p in contents}
      _lowercase											=[p for p in contents if p.endswith('''ckpt'''	)]
      assert len(UpperCamelCase_	) > 0

      self.assertIn('''test_generations.txt'''		,					UpperCamelCase_	)
      self.assertIn('''test_results.txt'''		,					UpperCamelCase_	)

      _lowercase											=load_json(model.metrics_save_path	)
      _lowercase											=metrics['''val'''][-1]
      self.assertGreaterEqual(last_step_stats['''val_avg_gen_time''']		,					0.01	)
      self.assertGreaterEqual(1.0		,					last_step_stats['''val_avg_gen_time''']	)
      self.assertIsInstance(last_step_stats[f"val_avg_{model.val_metric}"]		,					UpperCamelCase_	)
      _lowercase											=int(args_d['''max_epochs'''] * (1 / args_d['''val_check_interval''']) + 1	)
      self.assertEqual(len(metrics['''val''']	)		,					UpperCamelCase_	)
      self.assertEqual(len(metrics['''test''']	)		,					1	)
      return model







class 				lowerCamelCase__       (	lowercase_):





     @parameterized.expand(
         [T5_TINY, BART_TINY, MBART_TINY, MARIAN_TINY, FSMT_TINY]		,					)
     def 					__A (self		,					UpperCAmelCase	)					->       Optional[Any]:
      _lowercase											=CHEAP_ARGS.copy()
      _lowercase											='''translation''' if model in [MBART_TINY, MARIAN_TINY, FSMT_TINY] else '''summarization'''
      _lowercase											=0.1 if task == '''translation''' else 0

      _lowercase											=make_test_data_dir(tmp_dir=self.get_auto_remove_tmp_dir()	)
      _lowercase											=self.get_auto_remove_tmp_dir()
      args_d.update(
          data_dir=UpperCamelCase_		,					model_name_or_path=UpperCamelCase_		,					tokenizer_name=UpperCamelCase_		,					train_batch_size=2		,					eval_batch_size=2		,					output_dir=UpperCamelCase_		,					do_predict=UpperCamelCase_		,					task=UpperCamelCase_		,					src_lang='''en_XX'''		,					tgt_lang='''ro_RO'''		,					freeze_encoder=UpperCamelCase_		,					freeze_embeds=UpperCamelCase_		,					)
      assert "n_train" in args_d
      _lowercase											=argparse.Namespace(**UpperCamelCase_	)
      _lowercase											=main(UpperCamelCase_	)

      _lowercase											=module.model.get_input_embeddings()
      assert not input_embeds.weight.requires_grad
      if model == T5_TINY:
       _lowercase											=module.model.lm_head
       assert not lm_head.weight.requires_grad
       assert (lm_head.weight == input_embeds.weight).all().item()
      elif model == FSMT_TINY:
       _lowercase											=module.model.model
       _lowercase											=fsmt.decoder.embed_positions
       assert not embed_pos.weight.requires_grad
       assert not fsmt.decoder.embed_tokens.weight.requires_grad
       # check that embeds are not the same
       assert fsmt.decoder.embed_tokens != fsmt.encoder.embed_tokens
      else:
       _lowercase											=module.model.model
       _lowercase											=bart.decoder.embed_positions
       assert not embed_pos.weight.requires_grad
       assert not bart.shared.weight.requires_grad
       # check that embeds are the same
       assert bart.decoder.embed_tokens == bart.encoder.embed_tokens
       assert bart.decoder.embed_tokens == bart.shared

      _lowercase											=load_json(module.output_dir / '''text_batch.json'''	)
      assert isinstance(UpperCamelCase_		,					UpperCamelCase_	)
      assert len(UpperCamelCase_	) >= 4





     def 					__A (self	)					->       Dict:
      _lowercase											=CHEAP_ARGS.copy()

      _lowercase											='''summarization'''
      _lowercase											=make_test_data_dir(tmp_dir=self.get_auto_remove_tmp_dir()	)

      args_d.update(
          data_dir=UpperCamelCase_		,					tokenizer_name=UpperCamelCase_		,					train_batch_size=2		,					eval_batch_size=2		,					do_predict=UpperCamelCase_		,					task=UpperCamelCase_		,					src_lang='''en_XX'''		,					tgt_lang='''ro_RO'''		,					freeze_encoder=UpperCamelCase_		,					freeze_embeds=UpperCamelCase_		,					)

      # test models whose config includes the extra_model_args
      _lowercase											=BART_TINY
      _lowercase											=self.get_auto_remove_tmp_dir()
      _lowercase											=args_d.copy()
      args_da.update(
          model_name_or_path=UpperCamelCase_		,					output_dir=UpperCamelCase_		,					)
      _lowercase											=('''encoder_layerdrop''', '''decoder_layerdrop''', '''dropout''', '''attention_dropout''')
      for p in extra_model_params:
       _lowercase											=0.5
      _lowercase											=argparse.Namespace(**UpperCamelCase_	)
      _lowercase											=main(UpperCamelCase_	)
      for p in extra_model_params:
       assert getattr(model.config		,					UpperCamelCase_	) == 0.5, f"failed to override the model config for param {p}"

      # test models whose config doesn't include the extra_model_args
      _lowercase											=T5_TINY
      _lowercase											=self.get_auto_remove_tmp_dir()
      _lowercase											=args_d.copy()
      args_da.update(
          model_name_or_path=UpperCamelCase_		,					output_dir=UpperCamelCase_		,					)
      _lowercase											='''encoder_layerdrop'''
      _lowercase											=0.5
      _lowercase											=argparse.Namespace(**UpperCamelCase_	)
      with pytest.raises(UpperCamelCase_	) as excinfo:
       _lowercase											=main(UpperCamelCase_	)
      assert str(excinfo.value	) == f"model config doesn't have a `{unsupported_param}` attribute"





     def 					__A (self	)					->       Any:
      _lowercase											=CHEAP_ARGS.copy()

      _lowercase											='''summarization'''
      _lowercase											=make_test_data_dir(tmp_dir=self.get_auto_remove_tmp_dir()	)

      _lowercase											=BART_TINY
      _lowercase											=self.get_auto_remove_tmp_dir()

      args_d.update(
          data_dir=UpperCamelCase_		,					model_name_or_path=UpperCamelCase_		,					output_dir=UpperCamelCase_		,					tokenizer_name=UpperCamelCase_		,					train_batch_size=2		,					eval_batch_size=2		,					do_predict=UpperCamelCase_		,					task=UpperCamelCase_		,					src_lang='''en_XX'''		,					tgt_lang='''ro_RO'''		,					freeze_encoder=UpperCamelCase_		,					freeze_embeds=UpperCamelCase_		,					)

      # emulate finetune.py
      _lowercase											=argparse.ArgumentParser()
      _lowercase											=pl.Trainer.add_argparse_args(UpperCamelCase_	)
      _lowercase											=SummarizationModule.add_model_specific_args(UpperCamelCase_		,					os.getcwd()	)
      _lowercase											={'''--help''': True}

      # --help test
      with pytest.raises(UpperCamelCase_	) as excinfo:
       with CaptureStdout() as cs:
        _lowercase											=parser.parse_args(UpperCamelCase_	)
       assert False, "--help is expected to sys.exit"
      assert excinfo.type == SystemExit
      _lowercase											=lightning_base.arg_to_scheduler_metavar
      assert expected in cs.out, "--help is expected to list the supported schedulers"

      # --lr_scheduler=non_existing_scheduler test
      _lowercase											='''non_existing_scheduler'''
      _lowercase											={f"--lr_scheduler={unsupported_param}"}
      with pytest.raises(UpperCamelCase_	) as excinfo:
       with CaptureStderr() as cs:
        _lowercase											=parser.parse_args(UpperCamelCase_	)
       assert False, "invalid argument is expected to sys.exit"
      assert excinfo.type == SystemExit
      _lowercase											=f"invalid choice: '{unsupported_param}'"
      assert expected in cs.err, f"should have bailed on invalid choice of scheduler {unsupported_param}"

      # --lr_scheduler=existing_scheduler test
      _lowercase											='''cosine'''
      _lowercase											=args_d.copy()
      _lowercase											=supported_param
      _lowercase											=argparse.Namespace(**UpperCamelCase_	)
      _lowercase											=main(UpperCamelCase_	)
      assert (
          getattr(model.hparams		,					'''lr_scheduler'''	) == supported_param
      ), f"lr_scheduler={supported_param} shouldn't fail"




UpperCAmelCase__            =							[
    [0, 0, 1, 0, 0, 0, 0, 1, 0, 0, 0, 0, 0],
    [0, 0, 0, 0, 0, 0, 0, 1, 1, 1, 0, 0, 0],
    [0, 1, 1, 0, 1, 0, 0, 0, 0, 0, 0, 0, 0],
    [0, 1, 0, 0, 1, 1, 0, 0, 1, 0, 1, 0, 0],
    [0, 1, 0, 0, 1, 1, 0, 0, 1, 1, 1, 0, 0],
    [0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 1, 0, 0],
    [0, 0, 0, 0, 0, 0, 0, 1, 1, 1, 0, 0, 0],
    [0, 0, 0, 0, 0, 0, 0, 1, 1, 0, 0, 0, 0],
]
def     UpperCAmelCase_ (							__snake_case					,						__snake_case					,						__snake_case					,						__snake_case						)					->      bool:







 """simple docstring"""


 return 0 <= row < rows and 0 <= col < cols
def     UpperCAmelCase_ (							__snake_case					,						__snake_case					,						__snake_case					,						__snake_case						)					->      int:







 """simple docstring"""


 _lowercase											=len(__snake_case						)
 _lowercase											=len(mat[0]						)
 if is_safe(__snake_case					,						__snake_case					,						__snake_case					,						__snake_case						) and (row, col) not in seen and mat[row][col] == 1:
  seen.add((row, col)						)
  return (
      1
      + depth_first_search(row + 1					,						__snake_case					,						__snake_case					,						__snake_case						)
      + depth_first_search(row - 1					,						__snake_case					,						__snake_case					,						__snake_case						)
      + depth_first_search(__snake_case					,						col + 1					,						__snake_case					,						__snake_case						)
      + depth_first_search(__snake_case					,						col - 1					,						__snake_case					,						__snake_case						)
  )
 else:
  return 0
def     UpperCAmelCase_ (							__snake_case						)					->      int:







 """simple docstring"""


 _lowercase											=set()

 _lowercase											=0
 for row, line in enumerate(__snake_case						):
  for col, item in enumerate(__snake_case						):
   if item == 1 and (row, col) not in seen:
    # Maximizing the area
    _lowercase											=max(__snake_case					,						depth_first_search(__snake_case					,						__snake_case					,						__snake_case					,						__snake_case						)						)
 return max_area


if __name__ == "__main__":
  import doctest

  print(find_max_area(matrix))  # Output -> 6
  doctest.testmod()

