


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 							__lowerCamelCase		(					UpperCAmelCase_		:						Any		):




      """simple docstring"""





      a		:Optional[int]									=			filter(lambda UpperCAmelCase_		: p.requires_grad					,    model.parameters()		)
      a		:Dict									=			sum([np.prod(p.size()		) for p in model_parameters]		)
      return params


snake_case							:       Optional[Any]										=	logging.getLogger(__name__)
def 							__lowerCamelCase		(					UpperCAmelCase_		:						Union[str, Any]					,    UpperCAmelCase_		:						List[str]		):




      """simple docstring"""





      if metric == "rouge2":
            a		:Optional[Any]									=			'''{val_avg_rouge2:.4f}-{step_count}'''
      elif metric == "bleu":
            a		:List[Any]									=			'''{val_avg_bleu:.4f}-{step_count}'''
      elif metric == "em":
            a		:List[Any]									=			'''{val_avg_em:.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.'''		)

      a		:Dict									=			ModelCheckpoint(
          dirpath=UpperCAmelCase_					,    filename=UpperCAmelCase_					,    monitor=F'''val_{metric}'''					,    mode='''max'''					,    save_top_k=3					,    every_n_epochs=1					,    )
      return checkpoint_callback
def 							__lowerCamelCase		(					UpperCAmelCase_		:						List[Any]					,    UpperCAmelCase_		:						str		):




      """simple docstring"""





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







class 						_snake_case    ( pl.Callback				):




 def 			SCREAMING_SNAKE_CASE__							(						self   ,    _lowerCamelCase   ,    _lowerCamelCase       ):
       a		:List[str]									=			{F'''lr_group_{i}''': param['''lr'''] for i, param in enumerate(pl_module.trainer.optimizers[0].param_groups       )}
       pl_module.logger.log_metrics(_lowerCamelCase       )




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

       if not save_generations:
             return

       if "preds" in metrics:
             a		:Tuple									=			'''\n'''.join(metrics['''preds''']       )
             generations_file.open('''w+'''       ).write(_lowerCamelCase       )




 @rank_zero_only
 def 			SCREAMING_SNAKE_CASE__							(						self   ,    _lowerCamelCase   ,    _lowerCamelCase       ):
       try:
             a		:Optional[Any]									=			pl_module.model.model.num_parameters()
       except AttributeError:
             a		:str									=			pl_module.model.num_parameters()

       a		:List[Any]									=			count_trainable_parameters(_lowerCamelCase       )
       # 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 			SCREAMING_SNAKE_CASE__							(						self   ,    _lowerCamelCase   ,    _lowerCamelCase       ):
       save_json(pl_module.metrics   ,    pl_module.metrics_save_path       )
       return self._write_logs(_lowerCamelCase   ,    _lowerCamelCase   ,    '''test'''       )




 @rank_zero_only
 def 			SCREAMING_SNAKE_CASE__							(						self   ,    _lowerCamelCase   ,    _lowerCamelCase       ):
       save_json(pl_module.metrics   ,    pl_module.metrics_save_path       )
       # Uncommenting this will save val generations
       # return self._write_logs(trainer, pl_module, "valid")









import qiskit
def 							__lowerCamelCase		(					UpperCAmelCase_		:						int					,    UpperCAmelCase_		:						int		):




      """simple docstring"""





      a		:Tuple									=			qiskit.Aer.get_backend('''aer_simulator'''		)

      # Create a Quantum Circuit acting on the q register
      a		:str									=			qiskit.QuantumCircuit(UpperCAmelCase_					,    UpperCAmelCase_		)

      # Apply X (NOT) Gate to Qubits 0 & 1
      circuit.x(0		)
      circuit.x(1		)

      # Map the quantum measurement to the classical bits
      circuit.measure([0, 1]					,    [0, 1]		)

      # Execute the circuit on the qasm simulator
      a		:Optional[Any]									=			qiskit.execute(UpperCAmelCase_					,    UpperCAmelCase_					,    shots=1000		)

      # Return the histogram data of the results of the experiment.
      return job.result().get_counts(UpperCAmelCase_		)


if __name__ == "__main__":
     snake_case							:       Any										=	single_qubit_measure(2, 2)
     print(F"""Total count for various states are: {counts}""")





