

'''simple docstring'''

import argparse
import gc
import os
import sys
from pathlib import Path
from typing import List  # noqa: F401

import pytorch_lightning as pl
import torch
from finetune import SummarizationModule, TranslationModule
from finetune import main as ft_main
from make_student import create_student_by_copying_alternating_layers, get_layers_to_supervise
from torch import nn

from transformers import AutoModelForSeqaSeqLM, MBartTokenizer, TaForConditionalGeneration
from transformers.models.bart.modeling_bart import shift_tokens_right
from utils import calculate_bleu, check_output_dir, freeze_params, label_smoothed_nll_loss, use_task_specific_params


# need the parent dir module
sys.path.insert(2, str(Path(__file__).resolve().parents[1]))
from lightning_base import generic_train  # noqa


class   lowerCAmelCase     (      A   ):
       lowerCAmelCase_		= ["loss", "ce_loss", "mlm_loss", "hid_loss_enc", "hid_loss_dec"]







       def __init__(							self			: List[str]						,      __lowercase			: Any      ):







            """simple docstring"""







            assert Path(hparams.data_dir      ).exists()
            __lowercase					=Path(hparams.output_dir      )
            self.output_dir.mkdir(exist_ok=__lowercase      )

            __lowercase					=self.output_dir.joinpath('student'      )

            __lowercase					=str(__lowercase      )  # Tell lightning we are training the student
            __lowercase					=AutoModelForSeqaSeqLM.from_pretrained(hparams.teacher      ).eval()
            use_task_specific_params(__lowercase						,      hparams.task      )  # We copy good generation parameters to student by default
            if hparams.student is not None:
                 __lowercase					=AutoModelForSeqaSeqLM.from_pretrained(hparams.student      )
                 use_task_specific_params(__lowercase						,      hparams.task      )
                 __lowercase		,     __lowercase					=None, None
            else:
                 __lowercase		,     __lowercase		,     __lowercase					=create_student_by_copying_alternating_layers(
                     __lowercase						,      e=hparams.student_encoder_layers						,      d=hparams.student_decoder_layers						,      save_path=__lowercase      )

            if hparams.length_penalty != -1:
                 __lowercase					=hparams.length_penalty
            __lowercase					=hparams.teacher  # Use teacher's tokenizer
            super().__init__(__lowercase						,      model=__lowercase						,      config=student.config      )
            assert student.config.model_type == teacher.config.model_type, (
                f'''teacher, student model types should be the same, got {student.config.model_type} !='''
                f''' {teacher.config.model_type}'''
            )

            if student.config.model_type == "t5":
                 __lowercase					=len(student.get_encoder().block      )
                 __lowercase					=len(student.get_decoder().block      )
                 __lowercase					=len(teacher.get_encoder().block      )
                 __lowercase					=len(teacher.get_decoder().block      )
            else:
                 __lowercase					=student.config.encoder_layers
                 __lowercase					=student.config.decoder_layers
                 __lowercase					=teacher.config.encoder_layers
                 __lowercase					=teacher.config.decoder_layers

            __lowercase					=not (hparams.student is None or hparams.teacher == hparams.student)
            __lowercase					=(not self.different_base_models) and hparams.alpha_hid > 0
            __lowercase					=self.different_base_models or (student_encoder_layers != teacher_encoder_layers)
            # self.different_encoder determines whether we need to run the teacher encoder
            __lowercase					=teacher
            freeze_params(self.teacher      )

            if not self.different_encoder:  # To save RAM, delete teacher encoder and freeze student encoder.
                 try:
                      del self.teacher.model.encoder
                 except AttributeError:  # T5
                      del self.teacher.encoder

            if e_layer_ids is None:
                 __lowercase					=list(range(__lowercase      )      )
            if d_layer_ids is None:
                 __lowercase					=list(range(__lowercase      )      )

            __lowercase		,     __lowercase					=e_layer_ids, d_layer_ids  # type: List[int], List[int]

            if self.do_calc_hidden_loss:  # Intermediate supervision: Decide which layers to supervise
                 if hparams.supervise_forward:
                      __lowercase					=get_layers_to_supervise(
                          n_student=len(self.e_layer_ids      )						,      n_teacher=__lowercase      )
                      __lowercase					=get_layers_to_supervise(
                          n_student=len(self.d_layer_ids      )						,      n_teacher=__lowercase      )
                 else:  # student layer should emulate hidden states of the teacher layer it was copied from
                      __lowercase					=self.e_layer_ids
                      __lowercase					=self.d_layer_ids
            else:
                 __lowercase					=None
                 __lowercase					=None

            __lowercase					=nn.KLDivLoss(reduction='batchmean'      )
            __lowercase					=2.0
            __lowercase					=hparams.alpha_mlm
            __lowercase					=hparams.alpha_ce
            __lowercase					=hparams.alpha_hid
            gc.collect()
            torch.cuda.empty_cache()







       def 			snake_case   (							self			: List[Any]						,      __lowercase			: Optional[Any]						,      __lowercase			: Tuple						,      __lowercase			: str      ):







            """simple docstring"""







            __lowercase					=mask[:, :, None].expand_as(__lowercase      )
            __lowercase					=s_logits.size(-1      )
            __lowercase					=torch.masked_select(__lowercase						,      __lowercase      )  # (bs * seq_length * voc_size) modulo the 1s in mask
            __lowercase					=torch.masked_select(__lowercase						,      __lowercase      )  # (bs * seq_length * voc_size) modulo the 1s in mask
            __lowercase					=s_logits_slct.view(-1						,      __lowercase      )  # (bs * seq_length, voc_size) modulo the 1s in mask
            __lowercase					=t_logits_slct.view(-1						,      __lowercase      )  # (bs * seq_length, voc_size) modulo the 1s in mask
            assert t_logits_slct.size() == s_logits_slct.size()
            __lowercase					=(
                self.ce_loss_fct(
                    nn.functional.log_softmax(s_logits_slct / self.temperature						,      dim=-1      )						,      nn.functional.softmax(t_logits_slct / self.temperature						,      dim=-1      )						,      )
                * (self.temperature) ** 2
            )
            return loss_ce







       @staticmethod
       def 			snake_case   (							__lowercase			: Optional[Any]						,      __lowercase			: List[Any]      ):







            """simple docstring"""







            SummarizationModule.add_model_specific_args(__lowercase						,      __lowercase      )
            add_distill_args(__lowercase      )
            return parser







       def 			snake_case   (							self			: Optional[Any]						,      __lowercase			: dict      ):







            """simple docstring"""







            __lowercase					=self.tokenizer.pad_token_id
            __lowercase		,     __lowercase		,     __lowercase					=batch['input_ids'], batch['attention_mask'], batch['labels']
            if isinstance(self.model						,      __lowercase      ):
                 __lowercase					=self.model._shift_right(__lowercase      )
            else:
                 __lowercase					=shift_tokens_right(__lowercase						,      __lowercase      )

            # noinspection PyCallingNonCallable
            __lowercase					=self(
                __lowercase						,      attention_mask=__lowercase						,      decoder_input_ids=__lowercase						,      output_hidden_states=self.do_calc_hidden_loss						,      output_attentions=__lowercase						,      use_cache=__lowercase						,      )
            __lowercase					=student_outputs['logits']

            # Same cross entropy vs. label smoothing logic as finetune.py
            assert lm_logits.shape[-1] == self.model.config.vocab_size
            if self.hparams.label_smoothing == 0:
                 # Same behavior as modeling_bart.py, besides ignoring pad_token_id
                 __lowercase					=nn.CrossEntropyLoss(ignore_index=__lowercase      )
                 __lowercase					=loss_fct(lm_logits.view(-1						,      lm_logits.shape[-1]      )						,      labels.view(-1      )      )
            else:
                 __lowercase					=nn.functional.log_softmax(__lowercase						,      dim=-1      )
                 __lowercase		,     __lowercase					=label_smoothed_nll_loss(
                     __lowercase						,      __lowercase						,      self.hparams.label_smoothing						,      ignore_index=__lowercase      )

            def zero_tensor():
                 return torch.tensor(0.0      ).type_as(__lowercase      )

            __lowercase					=student_outputs[
                'encoder_last_hidden_state'
            ]  # use this unless self.different_base_models
            __lowercase		,     __lowercase					=zero_tensor(), zero_tensor()
            if self.different_encoder:  # compute encoder hidden state loss
                 __lowercase					=self.teacher.get_encoder()(
                     __lowercase						,      attention_mask=__lowercase						,      output_hidden_states=self.do_calc_hidden_loss						,      )
                 if self.different_base_models:
                      __lowercase					=all_teacher_encoder_outputs['last_hidden_state']
                 elif self.do_calc_hidden_loss:
                      __lowercase					=self.calc_hidden_loss(
                          __lowercase						,      student_outputs['encoder_hidden_states']						,      all_teacher_encoder_outputs['hidden_states']						,      self.e_matches						,      normalize_hidden=self.hparams.normalize_hidden						,      )

            __lowercase					=self.teacher(
                __lowercase						,      attention_mask=__lowercase						,      encoder_outputs=(teacher_enc_outputs,)						,      decoder_input_ids=__lowercase						,      output_hidden_states=self.do_calc_hidden_loss						,      use_cache=__lowercase						,      )
            __lowercase					=decoder_input_ids.ne(__lowercase      )
            __lowercase					=self.calc_ce_loss(__lowercase						,      __lowercase						,      teacher_outputs['logits']      )
            if self.do_calc_hidden_loss:  # Intermediate supervision of decoder hidden states
                 __lowercase					=self.calc_hidden_loss(
                     __lowercase						,      student_outputs['decoder_hidden_states']						,      teacher_outputs['decoder_hidden_states']						,      self.d_matches						,      normalize_hidden=self.hparams.normalize_hidden						,      )

            __lowercase					=(
                self.alpha_ce * loss_ce
                + self.alpha_mlm * student_lm_loss
                + self.hparams.alpha_hid * (hid_loss_enc + hid_loss_dec)
            )
            return blended_loss, loss_ce, student_lm_loss, hid_loss_enc, hid_loss_dec







       @staticmethod
       def 			snake_case   (							__lowercase			: str						,      __lowercase			: Optional[int]						,      __lowercase			: Any						,      __lowercase			: Tuple						,      __lowercase			: int      ):







            """simple docstring"""







            __lowercase					='expected list or tuple for hidden_states, got tensor of shape: '
            assert not isinstance(__lowercase						,      torch.Tensor      ), f'''{msg}{hidden_states.shape}'''
            assert not isinstance(__lowercase						,      torch.Tensor      ), f'''{msg}{hidden_states_T.shape}'''
            __lowercase					=attention_mask.to(hidden_states[0]      )
            __lowercase					=mask.sum() * hidden_states[0].size(-1      )
            __lowercase					=torch.stack([hidden_states[i] for i in range(len(__lowercase      )      )]      )
            __lowercase					=torch.stack([hidden_states_T[j] for j in matches]      )
            assert student_states.shape == teacher_states.shape, f'''{student_states.shape} != {teacher_states.shape}'''
            if normalize_hidden:
                 __lowercase					=nn.functional.layer_norm(__lowercase						,      student_states.shape[1:]      )
                 __lowercase					=nn.functional.layer_norm(__lowercase						,      teacher_states.shape[1:]      )
            __lowercase					=nn.functional.mse_loss(__lowercase						,      __lowercase						,      reduction='none'      )
            __lowercase					=(mse * mask.unsqueeze(0      ).unsqueeze(-1      )).sum() / valid_count
            return masked_mse






def 				__UpperCamelCase							(	lowercase__						:  Union[str, Any] ):






     '''simple docstring'''





     parser.add_argument('--teacher',					type=lowercase__ )
     parser.add_argument('--alpha_ce',					default=0.8,					type=lowercase__ )
     parser.add_argument('--alpha_mlm',					default=0.2,					type=lowercase__ )
     parser.add_argument('--alpha_hid',					default=0.0,					type=lowercase__,					required=lowercase__ )
     parser.add_argument('--student',					type=lowercase__,					required=lowercase__ )
     parser.add_argument('--student_decoder_layers',					default=12,					type=lowercase__,					required=lowercase__ )
     parser.add_argument('--student_encoder_layers',					default=12,					type=lowercase__,					required=lowercase__ )
     parser.add_argument('--no_teacher',					action='store_true',					default=lowercase__ )
     parser.add_argument('--length_penalty',					type=lowercase__,					default=-1 )
     parser.add_argument('--supervise_forward',					action='store_true',					default=lowercase__ )
     parser.add_argument('--normalize_hidden',					action='store_true',					default=lowercase__ )


class   lowerCAmelCase     (      A   ):
       lowerCAmelCase_		= "translation"
       lowerCAmelCase_		= ["bleu"]
       lowerCAmelCase_		= "bleu"







       def __init__(							self			: int						,      __lowercase			: Tuple						,      **__lowercase			: str      ):







            """simple docstring"""







            super().__init__(__lowercase						,      **__lowercase      )
            assert hparams.src_lang is not None
            assert hparams.tgt_lang is not None
            __lowercase					=hparams.src_lang
            __lowercase					=hparams.tgt_lang
            if self.model.config.decoder_start_token_id is None and isinstance(self.tokenizer						,      __lowercase      ):
                 __lowercase					=self.tokenizer.lang_code_to_id[hparams.tgt_lang]







       def 			snake_case   (							self			: Optional[Any]						,      __lowercase			: Optional[Any]						,      __lowercase			: int      ):







            """simple docstring"""







            return calculate_bleu(__lowercase						,      __lowercase      )







       @staticmethod
       def 			snake_case   (							__lowercase			: Union[str, Any]						,      __lowercase			: List[str]      ):







            """simple docstring"""







            TranslationModule.add_model_specific_args(__lowercase						,      __lowercase      )
            add_distill_args(__lowercase      )
            return parser






def 				__UpperCamelCase							(	lowercase__						:  List[Any] ):






     '''simple docstring'''





     if args.no_teacher:
          __lowercase					=TranslationModule if 'translation' in args.task else SummarizationModule
     else:  # DISTILL WITH TEACHER
          __lowercase					=TranslationDistiller if 'translation' in args.task else SummarizationDistiller
     __lowercase					=module_cls.__name__
     print(F'''using module {args.setup_cls}''' )
     __lowercase					=module_cls(lowercase__ )
     return model

def 				__UpperCamelCase							(	lowercase__						:  List[str] ):






     '''simple docstring'''





     Path(args.output_dir ).mkdir(exist_ok=lowercase__ )
     check_output_dir(lowercase__,					expected_items=3 )

     __lowercase					=create_module(lowercase__ )
     return ft_main(lowercase__,					model=lowercase__ )


if __name__ == "__main__":
       UpperCAmelCase				   =  argparse.ArgumentParser()
       UpperCAmelCase				   =  pl.Trainer.add_argparse_args(parser)
       UpperCAmelCase				   =  SummarizationDistiller.add_model_specific_args(parser, os.getcwd())
       UpperCAmelCase				   =  parser.parse_args()

       distill_main(args)





'''simple docstring'''

from __future__ import annotations

# Define the Gravitational Constant G and the function
UpperCAmelCase				   =  6.6_7_4_3E-1_1  # unit of G : m^3 * kg^-1 * s^-2

def 				__UpperCamelCase							(	lowercase__						:  float,					lowercase__						:  float,					lowercase__						:  float,					lowercase__						:  float ):






     '''simple docstring'''





     __lowercase					=mass_a * mass_a

     if (force, mass_a, mass_a, distance).count(0 ) != 1:
          raise ValueError('One and only one argument must be 0' )
     if force < 0:
          raise ValueError('Gravitational force can not be negative' )
     if distance < 0:
          raise ValueError('Distance can not be negative' )
     if mass_a < 0 or mass_a < 0:
          raise ValueError('Mass can not be negative' )
     if force == 0:
          __lowercase					=GRAVITATIONAL_CONSTANT * product_of_mass / (distance**2)
          return {"force": force}
     elif mass_a == 0:
          __lowercase					=(force) * (distance**2) / (GRAVITATIONAL_CONSTANT * mass_a)
          return {"mass_1": mass_a}
     elif mass_a == 0:
          __lowercase					=(force) * (distance**2) / (GRAVITATIONAL_CONSTANT * mass_a)
          return {"mass_2": mass_a}
     elif distance == 0:
          __lowercase					=(GRAVITATIONAL_CONSTANT * product_of_mass / (force)) ** 0.5
          return {"distance": distance}
     raise ValueError('One and only one argument must be 0' )


# Run doctest
if __name__ == "__main__":
       import doctest

       doctest.testmod()


