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)

a_					      = logging.getLogger()
a_					      = torch.cuda.is_available()
a_					      = {
    '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      __lowercase  (		lowerCamelCase      :  Path						,						lowerCamelCase      :  list			):
 UpperCamelCase_      :		int				=      '\n'.join(lowerCamelCase			)
 Path(lowerCamelCase			).open('w'			).writelines(lowerCamelCase			)


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


a_					      = logging.StreamHandler(sys.stdout)
logger.addHandler(stream_handler)
logging.disable(logging.CRITICAL)  # remove noisy download output from tracebacks





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





class 	_lowercase		(      snake_case_ ):

 @classmethod
 def 		SCREAMING_SNAKE_CASE__     (				cls			:							Optional[int]    )     ->		Any:

  """simple docstring"""





  logging.disable(logging.CRITICAL    )  # remove noisy download output from tracebacks
  return cls

 @slow
 @require_torch_gpu
 def 		SCREAMING_SNAKE_CASE__     (				self			:							List[Any]    )     ->		Dict:

  """simple docstring"""





  UpperCamelCase_      :		Dict				=      list_models()
  UpperCamelCase_      :		List[str]				=      'sshleifer'
  UpperCamelCase_      :		Optional[int]				=      [x.modelId for x in model_list if x.modelId.startswith(snake_case    )]
  UpperCamelCase_      :		List[Any]				=      ['sshleifer/blenderbot-3B', 'sshleifer/blenderbot-90M']
  UpperCamelCase_      :		Optional[Any]				=      []
  for m in model_ids:
   if m in allowed_to_be_broken:
    continue
   try:
    AutoConfig.from_pretrained(snake_case    )
   except Exception:
    failures.append(snake_case    )
  assert not failures, f"The following models could not be loaded through AutoConfig: {failures}"

 def 		SCREAMING_SNAKE_CASE__     (				self			:							Optional[int]    )     ->		Dict:

  """simple docstring"""





  UpperCamelCase_      :		List[Any]				=      {'student_encoder_layers': 2, 'student_decoder_layers': 1, 'no_teacher': True}
  self._test_distiller_cli(snake_case    )

 def 		SCREAMING_SNAKE_CASE__     (				self			:							int    )     ->		int:

  """simple docstring"""





  UpperCamelCase_      :		List[Any]				=      {
      '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',
  }
  UpperCamelCase_      :		List[Any]				=      self._test_distiller_cli(snake_case     ,   check_contents=snake_case    )

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

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

 def 		SCREAMING_SNAKE_CASE__     (				self			:							Any    )     ->		str:

  """simple docstring"""





  UpperCamelCase_      :		Any				=      AutoModelForSeqaSeqLM.from_pretrained(snake_case    )
  UpperCamelCase_,				UpperCamelCase_      :		Dict				=      model.dummy_inputs['input_ids'], model.dummy_inputs['attention_mask']
  UpperCamelCase_      :		List[Any]				=      torch.tensor([[0, 4, 8, 2], [0, 8, 2, 1]]     ,   dtype=torch.long     ,   device=model.device    )
  UpperCamelCase_      :		List[str]				=      target_ids[:, :-1].contiguous()  # Why this line?
  UpperCamelCase_      :		Tuple				=      target_ids[:, 1:].clone()  # why clone?
  UpperCamelCase_      :		int				=      model(
      snake_case     ,   attention_mask=snake_case     ,   decoder_input_ids=snake_case     ,   labels=snake_case     ,   use_cache=snake_case    ).loss

  UpperCamelCase_      :		Optional[int]				=      model(snake_case     ,   attention_mask=snake_case     ,   decoder_input_ids=snake_case     ,   use_cache=snake_case    ).logits

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

 def 		SCREAMING_SNAKE_CASE__     (				self			:							Any    )     ->		Dict:

  """simple docstring"""





  UpperCamelCase_      :		List[Any]				=      {
      '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',
  }
  UpperCamelCase_      :		Optional[Any]				=      self._test_distiller_cli(snake_case     ,   check_contents=snake_case    )
  assert model.model.config.model_type == "mbart"

  UpperCamelCase_      :		List[str]				=      list(Path(model.output_dir    ).glob('*.ckpt'    )    )
  self.assertEqual(1     ,   len(snake_case    )    )
  UpperCamelCase_      :		List[Any]				=      list(Path(model.output_dir    ).glob('**/*.bin'    )    )
  UpperCamelCase_      :		Optional[Any]				=      list(Path(model.output_dir    ).glob('best_tfmr/*'    )    )
  assert len(snake_case    ) > 2
  self.assertEqual(len(snake_case    )     ,   2    )

 def 		SCREAMING_SNAKE_CASE__     (				self			:							Optional[int]    )     ->		int:

  """simple docstring"""





  UpperCamelCase_      :		int				=      {
      '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(snake_case    )

 def 		SCREAMING_SNAKE_CASE__     (				self			:							Dict    )     ->		Dict:

  """simple docstring"""





  UpperCamelCase_      :		Dict				=      {
      'teacher': T5_TINY,
      'student': T5_TINIER,
      'model_name_or_path': T5_TINIER,
      'tokenizer_name': T5_TINIER,
  }
  self._test_distiller_cli(snake_case    )


 def 		SCREAMING_SNAKE_CASE__     (				self			:							str     ,   snake_case			:							str     ,   snake_case			:							str=True    )     ->		Optional[int]:

  """simple docstring"""





  UpperCamelCase_      :		Optional[Any]				=      {
      '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(snake_case    )
  UpperCamelCase_      :		dict				=      CHEAP_ARGS.copy()
  UpperCamelCase_      :		Dict				=      make_test_data_dir(tmp_dir=self.get_auto_remove_tmp_dir()    )
  UpperCamelCase_      :		List[str]				=      self.get_auto_remove_tmp_dir()

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

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

  UpperCamelCase_      :		str				=      load_json(model.metrics_save_path    )
  UpperCamelCase_      :		Optional[int]				=      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}"]     ,   snake_case    )
  UpperCamelCase_      :		Union[str, Any]				=      int(args_d['max_epochs'] * (1 / args_d['val_check_interval']) + 1    )
  self.assertEqual(len(metrics['val']    )     ,   snake_case    )
  self.assertEqual(len(metrics['test']    )     ,   1    )
  return model





class 	_lowercase		(      snake_case_ ):

 @parameterized.expand(
     [T5_TINY, BART_TINY, MBART_TINY, MARIAN_TINY, FSMT_TINY]     ,   )
 def 		SCREAMING_SNAKE_CASE__     (				self			:							List[Any]     ,   snake_case			:							Optional[int]    )     ->		Optional[int]:

  """simple docstring"""





  UpperCamelCase_      :		dict				=      CHEAP_ARGS.copy()
  UpperCamelCase_      :		str				=      'translation' if model in [MBART_TINY, MARIAN_TINY, FSMT_TINY] else 'summarization'
  UpperCamelCase_      :		List[Any]				=      0.1 if task == 'translation' else 0

  UpperCamelCase_      :		Optional[int]				=      make_test_data_dir(tmp_dir=self.get_auto_remove_tmp_dir()    )
  UpperCamelCase_      :		Optional[Any]				=      self.get_auto_remove_tmp_dir()
  args_d.update(
      data_dir=snake_case     ,   model_name_or_path=snake_case     ,   tokenizer_name=snake_case     ,   train_batch_size=2     ,   eval_batch_size=2     ,   output_dir=snake_case     ,   do_predict=snake_case     ,   task=snake_case     ,   src_lang='en_XX'     ,   tgt_lang='ro_RO'     ,   freeze_encoder=snake_case     ,   freeze_embeds=snake_case     ,   )
  assert "n_train" in args_d
  UpperCamelCase_      :		List[Any]				=      argparse.Namespace(**snake_case    )
  UpperCamelCase_      :		Union[str, Any]				=      main(snake_case    )

  UpperCamelCase_      :		List[Any]				=      module.model.get_input_embeddings()
  assert not input_embeds.weight.requires_grad
  if model == T5_TINY:
   UpperCamelCase_      :		Optional[int]				=      module.model.lm_head
   assert not lm_head.weight.requires_grad
   assert (lm_head.weight == input_embeds.weight).all().item()
  elif model == FSMT_TINY:
   UpperCamelCase_      :		str				=      module.model.model
   UpperCamelCase_      :		Optional[int]				=      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:
   UpperCamelCase_      :		Optional[int]				=      module.model.model
   UpperCamelCase_      :		List[str]				=      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

  UpperCamelCase_      :		Optional[Any]				=      load_json(module.output_dir / 'text_batch.json'    )
  assert isinstance(snake_case     ,   snake_case    )
  assert len(snake_case    ) >= 4

 def 		SCREAMING_SNAKE_CASE__     (				self			:							str    )     ->		str:

  """simple docstring"""





  UpperCamelCase_      :		dict				=      CHEAP_ARGS.copy()

  UpperCamelCase_      :		Tuple				=      'summarization'
  UpperCamelCase_      :		Optional[int]				=      make_test_data_dir(tmp_dir=self.get_auto_remove_tmp_dir()    )

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

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

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


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

  """simple docstring"""





  UpperCamelCase_      :		dict				=      CHEAP_ARGS.copy()

  UpperCamelCase_      :		Optional[int]				=      'summarization'
  UpperCamelCase_      :		str				=      make_test_data_dir(tmp_dir=self.get_auto_remove_tmp_dir()    )

  UpperCamelCase_      :		Optional[int]				=      BART_TINY
  UpperCamelCase_      :		Union[str, Any]				=      self.get_auto_remove_tmp_dir()

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

  # emulate finetune.py
  UpperCamelCase_      :		Optional[Any]				=      argparse.ArgumentParser()
  UpperCamelCase_      :		Dict				=      pl.Trainer.add_argparse_args(snake_case    )
  UpperCamelCase_      :		Any				=      SummarizationModule.add_model_specific_args(snake_case     ,   os.getcwd()    )
  UpperCamelCase_      :		List[str]				=      {'--help': True}

  # --help test
  with pytest.raises(snake_case    ) as excinfo:
   with CaptureStdout() as cs:
    UpperCamelCase_      :		Any				=      parser.parse_args(snake_case    )
   assert False, "--help is expected to sys.exit"
  assert excinfo.type == SystemExit
  UpperCamelCase_      :		Union[str, Any]				=      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
  UpperCamelCase_      :		Any				=      'non_existing_scheduler'
  UpperCamelCase_      :		int				=      {f"--lr_scheduler={unsupported_param}"}
  with pytest.raises(snake_case    ) as excinfo:
   with CaptureStderr() as cs:
    UpperCamelCase_      :		List[Any]				=      parser.parse_args(snake_case    )
   assert False, "invalid argument is expected to sys.exit"
  assert excinfo.type == SystemExit
  UpperCamelCase_      :		List[str]				=      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
  UpperCamelCase_      :		int				=      'cosine'
  UpperCamelCase_      :		Tuple				=      args_d.copy()
  UpperCamelCase_      :		int				=      supported_param
  UpperCamelCase_      :		Dict				=      argparse.Namespace(**snake_case    )
  UpperCamelCase_      :		Optional[Any]				=      main(snake_case    )
  assert (
      getattr(model.hparams     ,   'lr_scheduler'    ) == supported_param
  ), f"lr_scheduler={supported_param} shouldn't fail"




def      __lowercase  (		lowerCamelCase      :  int = 998001			):
 UpperCamelCase_      :		int				=      0
 for i in range(999						,						99						,						-1			):  # 3 digit numbers range from 999 down to 100
  for j in range(999						,						99						,						-1			):
   UpperCamelCase_      :		str				=      str(i * j			)
   if product_string == product_string[::-1] and i * j < n:
    UpperCamelCase_      :		str				=      max(lowerCamelCase						,						i * j			)
 return answer


if __name__ == "__main__":
   print(F"""{solution() = }""")



