





import functools
import operator

from ...configuration_utils import PretrainedConfig
from ...utils import logging


_lowerCAmelCase				:   Any						       =						logging.get_logger(__name__)

_lowerCAmelCase				:   Optional[Any]						       =						{
    '''microsoft/unispeech-sat-base-100h-libri-ft''': (
        '''https://huggingface.co/microsoft/unispeech-sat-base-100h-libri-ft/resolve/main/config.json'''
    ),
    # See all UniSpeechSat models at https://huggingface.co/models?filter=unispeech_sat
}


class        __magic_name__	(				lowerCamelCase__				):





     """simple docstring"""



     __UpperCamelCase            =  'unispeech-sat'



     def __init__(				self							:Any      ,   snake_case							:Any=32      ,   snake_case							:Any=768      ,   snake_case							:Optional[Any]=12      ,   snake_case							:int=12      ,   snake_case							:str=3_072      ,   snake_case							:Dict="gelu"      ,   snake_case							:int=0.1      ,   snake_case							:Any=0.1      ,   snake_case							:Tuple=0.1      ,   snake_case							:Tuple=0.0      ,   snake_case							:Union[str, Any]=0.0      ,   snake_case							:Tuple=0.1      ,   snake_case							:Dict=0.1      ,   snake_case							:List[str]=0.02      ,   snake_case							:Optional[Any]=1e-5      ,   snake_case							:Union[str, Any]="group"      ,   snake_case							:Optional[Any]="gelu"      ,   snake_case							:List[str]=(512, 512, 512, 512, 512, 512, 512)      ,   snake_case							:Tuple=(5, 2, 2, 2, 2, 2, 2)      ,   snake_case							:List[str]=(10, 3, 3, 3, 3, 2, 2)      ,   snake_case							:Optional[Any]=False      ,   snake_case							:Any=128      ,   snake_case							:Union[str, Any]=16      ,   snake_case							:Dict=False      ,   snake_case							:List[Any]=True      ,   snake_case							:Tuple=0.05      ,   snake_case							:Union[str, Any]=10      ,   snake_case							:List[Any]=2      ,   snake_case							:Tuple=0.0      ,   snake_case							:Optional[Any]=10      ,   snake_case							:int=0      ,   snake_case							:Tuple=320      ,   snake_case							:Dict=2      ,   snake_case							:Dict=0.1      ,   snake_case							:List[str]=100      ,   snake_case							:str=256      ,   snake_case							:List[str]=256      ,   snake_case							:Union[str, Any]=0.1      ,   snake_case							:Dict="mean"      ,   snake_case							:Dict=False      ,   snake_case							:Dict=False      ,   snake_case							:Tuple=256      ,   snake_case							:Optional[int]=(512, 512, 512, 512, 1_500)      ,   snake_case							:List[Any]=(5, 3, 3, 1, 1)      ,   snake_case							:Dict=(1, 2, 3, 1, 1)      ,   snake_case							:Optional[Any]=512      ,   snake_case							:List[Any]=0      ,   snake_case							:List[Any]=1      ,   snake_case							:Optional[Any]=2      ,   snake_case							:List[str]=504      ,   **snake_case							:Dict      ,   ):





        '''simple docstring'''




        super().__init__(**SCREAMING_SNAKE_CASE_      ,   pad_token_id=SCREAMING_SNAKE_CASE_      ,   bos_token_id=SCREAMING_SNAKE_CASE_      ,   eos_token_id=SCREAMING_SNAKE_CASE_	)
        A_       :					List[Any]        =      hidden_size
        A_       :					Any        =      feat_extract_norm
        A_       :					Union[str, Any]        =      feat_extract_activation
        A_       :					List[str]        =      list(SCREAMING_SNAKE_CASE_	)
        A_       :					Union[str, Any]        =      list(SCREAMING_SNAKE_CASE_	)
        A_       :					Optional[int]        =      list(SCREAMING_SNAKE_CASE_	)
        A_       :					List[str]        =      conv_bias
        A_       :					List[Any]        =      num_conv_pos_embeddings
        A_       :					List[Any]        =      num_conv_pos_embedding_groups
        A_       :					Optional[Any]        =      len(self.conv_dim	)
        A_       :					List[Any]        =      num_hidden_layers
        A_       :					Any        =      intermediate_size
        A_       :					Union[str, Any]        =      hidden_act
        A_       :					Any        =      num_attention_heads
        A_       :					Optional[int]        =      hidden_dropout
        A_       :					Tuple        =      attention_dropout
        A_       :					Tuple        =      activation_dropout
        A_       :					List[Any]        =      feat_proj_dropout
        A_       :					List[str]        =      final_dropout
        A_       :					Optional[int]        =      layerdrop
        A_       :					List[Any]        =      layer_norm_eps
        A_       :					int        =      initializer_range
        A_       :					int        =      vocab_size
        A_       :					int        =      num_clusters
        A_       :					Optional[Any]        =      do_stable_layer_norm
        A_       :					Any        =      use_weighted_layer_sum

        if (
            (len(self.conv_stride	) != self.num_feat_extract_layers)
            or (len(self.conv_kernel	) != self.num_feat_extract_layers)
            or (len(self.conv_dim	) != self.num_feat_extract_layers)
        ):
           raise ValueError(
               "Configuration for convolutional layers is incorrect. It is required that `len(config.conv_dim)` =="
               " `len(config.conv_stride)` == `len(config.conv_kernel)`, but is `len(config.conv_dim) ="
               f" {len(self.conv_dim	)}`, `len(config.conv_stride) = {len(self.conv_stride	)}`,"
               f" `len(config.conv_kernel) = {len(self.conv_kernel	)}`."	)

        # fine-tuning config parameters for SpecAugment: https://arxiv.org/abs/1904.08779
        A_       :					List[Any]        =      apply_spec_augment
        A_       :					Union[str, Any]        =      mask_time_prob
        A_       :					Optional[Any]        =      mask_time_length
        A_       :					Any        =      mask_time_min_masks
        A_       :					List[str]        =      mask_feature_prob
        A_       :					Dict        =      mask_feature_length
        A_       :					List[Any]        =      mask_feature_min_masks

        # parameters for pretraining with codevector quantized representations
        A_       :					int        =      num_codevectors_per_group
        A_       :					List[Any]        =      num_codevector_groups
        A_       :					Any        =      contrastive_logits_temperature
        A_       :					int        =      feat_quantizer_dropout
        A_       :					Union[str, Any]        =      num_negatives
        A_       :					str        =      codevector_dim
        A_       :					Optional[int]        =      proj_codevector_dim
        A_       :					Tuple        =      diversity_loss_weight

        # ctc loss
        A_       :					Any        =      ctc_loss_reduction
        A_       :					List[Any]        =      ctc_zero_infinity

        # SequenceClassification-specific parameter. Feel free to ignore for other classes.
        A_       :					Dict        =      classifier_proj_size

        # XVector-specific parameters. Feel free to ignore for other classes.
        A_       :					Any        =      list(SCREAMING_SNAKE_CASE_	)
        A_       :					str        =      list(SCREAMING_SNAKE_CASE_	)
        A_       :					Union[str, Any]        =      list(SCREAMING_SNAKE_CASE_	)
        A_       :					Optional[int]        =      xvector_output_dim





     @property
     def     SCREAMING_SNAKE_CASE		(				self							:Optional[Any]	):





        '''simple docstring'''




        return functools.reduce(operator.mul      ,   self.conv_stride      ,   1	)


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 							a		(				snake_case__:   Any   ):


				'''simple docstring'''
				lowercase_	    =   filter(lambda snake_case__   : p.requires_grad		,  model.parameters()   )
				lowercase_	    =   sum([np.prod(p.size()   ) for p in model_parameters]   )
				return params


__a         =   logging.getLogger(__name__)




def 							a		(				snake_case__:   int		,  snake_case__:   Any   ):


				'''simple docstring'''
				if metric == "rouge2":
								lowercase_	    =   '''{val_avg_rouge2:.4f}-{step_count}'''
				elif metric == "bleu":
								lowercase_	    =   '''{val_avg_bleu:.4f}-{step_count}'''
				elif metric == "em":
								lowercase_	    =   '''{val_avg_em:.4f}-{step_count}'''
				elif metric == "loss":
								lowercase_	    =   '''{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.'''   )

				lowercase_	    =   ModelCheckpoint(
				    dirpath=snake_case__		,  filename=snake_case__		,  monitor=F'''val_{metric}'''		,  mode='''max'''		,  save_top_k=1		,  every_n_epochs=1		,  )

				return checkpoint_callback




def 							a		(				snake_case__:   str		,  snake_case__:   List[str]   ):


				'''simple docstring'''
				return EarlyStopping(
				    monitor=F'''val_{metric}'''		,  mode='''min''' if '''loss''' in metric else '''max'''		,  patience=snake_case__		,  verbose=snake_case__		,  )


class 					lowercase__(							pl.Callback				):



	"""simple docstring"""




	def 		_lowercase    (			self    :     List[str]			,				SCREAMING_SNAKE_CASE_    :     Optional[int]			,				SCREAMING_SNAKE_CASE_    :     Any			)			->	Optional[Any]:
					lowercase_	    =   {f'''lr_group_{i}''': param['''lr'''] for i, param in enumerate(pl_module.trainer.optimizers[0].param_groups			)}
					pl_module.logger.log_metrics(SCREAMING_SNAKE_CASE_			)
	@rank_zero_only
	def 		_lowercase    (			self    :     Optional[int]			,				SCREAMING_SNAKE_CASE_    :     pl.Trainer			,				SCREAMING_SNAKE_CASE_    :     pl.LightningModule			,				SCREAMING_SNAKE_CASE_    :     str			,				SCREAMING_SNAKE_CASE_    :     Tuple=True			)			->	None:
					logger.info(f'''***** {type_path} results at step {trainer.global_step:05d} *****'''			)
					lowercase_	    =   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
					lowercase_	    =   Path(pl_module.hparams.output_dir			)
					if type_path == "test":
									lowercase_	    =   od / '''test_results.txt'''
									lowercase_	    =   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.
									lowercase_	    =   od / f'''{type_path}_results/{trainer.global_step:05d}.txt'''
									lowercase_	    =   od / f'''{type_path}_generations/{trainer.global_step:05d}.txt'''
									results_file.parent.mkdir(exist_ok=SCREAMING_SNAKE_CASE_			)
									generations_file.parent.mkdir(exist_ok=SCREAMING_SNAKE_CASE_			)
					with open(SCREAMING_SNAKE_CASE_			,				'''a+'''			) as writer:
									for key in sorted(SCREAMING_SNAKE_CASE_			):
													if key in ["log", "progress_bar", "preds"]:
																	continue
													lowercase_	    =   metrics[key]
													if isinstance(SCREAMING_SNAKE_CASE_			,				torch.Tensor			):
																	lowercase_	    =   val.item()
													lowercase_	    =   f'''{key}: {val:.6f}\n'''
													writer.write(SCREAMING_SNAKE_CASE_			)

					if not save_generations:
									return

					if "preds" in metrics:
									lowercase_	    =   '''\n'''.join(metrics['''preds''']			)
									generations_file.open('''w+'''			).write(SCREAMING_SNAKE_CASE_			)
	@rank_zero_only
	def 		_lowercase    (			self    :     Any			,				SCREAMING_SNAKE_CASE_    :     int			,				SCREAMING_SNAKE_CASE_    :     Any			)			->	Union[str, Any]:
					try:
									lowercase_	    =   pl_module.model.model.num_parameters()
					except AttributeError:
									lowercase_	    =   pl_module.model.num_parameters()

					lowercase_	    =   count_trainable_parameters(SCREAMING_SNAKE_CASE_			)
					# 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 		_lowercase    (			self    :     Tuple			,				SCREAMING_SNAKE_CASE_    :     pl.Trainer			,				SCREAMING_SNAKE_CASE_    :     pl.LightningModule			)			->	Any:
					save_json(pl_module.metrics			,				pl_module.metrics_save_path			)
					return self._write_logs(SCREAMING_SNAKE_CASE_			,				SCREAMING_SNAKE_CASE_			,				'''test'''			)




	@rank_zero_only
	def 		_lowercase    (			self    :     str			,				SCREAMING_SNAKE_CASE_    :     pl.Trainer			,				SCREAMING_SNAKE_CASE_    :     Optional[Any]			)			->	Dict:
					save_json(pl_module.metrics			,				pl_module.metrics_save_path			)
					# Uncommenting this will save val generations
					# return self._write_logs(trainer, pl_module, "valid")




