


import math
import time
from typing import Dict, List, Optional

from torch.utils.data import Dataset

from transformers import SeqaSeqTrainer, is_torch_tpu_available
from transformers.trainer_utils import PredictionOutput, speed_metrics


if is_torch_tpu_available(check_device=False):
						import torch_xla.core.xla_model as xm
						import torch_xla.debug.metrics as met
class        _lowercase      (			__A       ):






				'''simple docstring'''



				def __init__(       self			:List[Any]			,			*lowerCAmelCase__			:List[Any]			,			lowerCAmelCase__			:Union[str, Any]=None			,			lowerCAmelCase__			:Tuple=None			,			**lowerCAmelCase__			:str	)		->			Union[str, Any]:
										super().__init__(*lowercase_			,			**lowercase_	)
										__SCREAMING_SNAKE_CASE						:     Tuple					=		eval_examples
										__SCREAMING_SNAKE_CASE						:     Optional[int]					=		post_process_function


				def 		__magic_name__(       self			:List[Any]			,			lowerCAmelCase__			:Optional[Dataset] = None			,			lowerCAmelCase__			:Any=None			,			lowerCAmelCase__			:Optional[List[str]] = None			,			lowerCAmelCase__			:str = "eval"			,			**lowerCAmelCase__			:Any			,			)		->			Dict[str, float]:
										__SCREAMING_SNAKE_CASE						:     List[Any]					=		gen_kwargs.copy()
										__SCREAMING_SNAKE_CASE						:     Optional[int]					=		(
										    gen_kwargs["max_length"] if gen_kwargs.get('''max_length'''	) is not None else self.args.generation_max_length
										)
										__SCREAMING_SNAKE_CASE						:     List[str]					=		(
										    gen_kwargs["num_beams"] if gen_kwargs.get('''num_beams'''	) is not None else self.args.generation_num_beams
										)
										__SCREAMING_SNAKE_CASE						:     Tuple					=		gen_kwargs

										__SCREAMING_SNAKE_CASE						:     Tuple					=		self.eval_dataset if eval_dataset is None else eval_dataset
										__SCREAMING_SNAKE_CASE						:     str					=		self.get_eval_dataloader(lowercase_	)
										__SCREAMING_SNAKE_CASE						:     Any					=		self.eval_examples if eval_examples is None else eval_examples

										# Temporarily disable metric computation, we will do it in the loop here.
										__SCREAMING_SNAKE_CASE						:     List[str]					=		self.compute_metrics
										__SCREAMING_SNAKE_CASE						:     str					=		None
										__SCREAMING_SNAKE_CASE						:     Any					=		time.time()
										__SCREAMING_SNAKE_CASE						:     List[Any]					=		self.prediction_loop if self.args.use_legacy_prediction_loop else self.evaluation_loop
										try:
																__SCREAMING_SNAKE_CASE						:     Optional[int]					=		eval_loop(
																    lowercase_			,			description='''Evaluation'''			,			prediction_loss_only=True if compute_metrics is None else None			,			ignore_keys=lowercase_			,			metric_key_prefix=lowercase_			,			)
										finally:
																__SCREAMING_SNAKE_CASE						:     int					=		compute_metrics
										__SCREAMING_SNAKE_CASE						:     Tuple					=		self.args.eval_batch_size * self.args.world_size
										if f'''{metric_key_prefix}_jit_compilation_time''' in output.metrics:
																start_time += output.metrics[f'''{metric_key_prefix}_jit_compilation_time''']
										output.metrics.update(
										    speed_metrics(
										        lowercase_			,			lowercase_			,			num_samples=output.num_samples			,			num_steps=math.ceil(output.num_samples / total_batch_size	)			,			)	)

										if self.post_process_function is not None and self.compute_metrics is not None and self.args.should_save:
																# Only the main node write the results by default
																__SCREAMING_SNAKE_CASE						:     Optional[int]					=		self.post_process_function(lowercase_			,			lowercase_			,			lowercase_	)
																__SCREAMING_SNAKE_CASE						:     str					=		self.compute_metrics(lowercase_	)

																# Prefix all keys with metric_key_prefix + '_'
																for key in list(metrics.keys()	):
																						if not key.startswith(f'''{metric_key_prefix}_'''	):
																												__SCREAMING_SNAKE_CASE						:     List[str]					=		metrics.pop(lowercase_	)

																metrics.update(output.metrics	)
										else:
																__SCREAMING_SNAKE_CASE						:     Union[str, Any]					=		output.metrics

										if self.args.should_log:
																# Only the main node log the results by default
																self.log(lowercase_	)

										if self.args.tpu_metrics_debug or self.args.debug:
																# tpu-comment: Logging debug metrics for PyTorch/XLA (compile, execute times, ops, etc.)
																xm.master_print(met.metrics_report()	)

										__SCREAMING_SNAKE_CASE						:     List[str]					=		self.callback_handler.on_evaluate(self.args			,			self.state			,			self.control			,			lowercase_	)
										return metrics



				def 		__magic_name__(       self			:str			,			lowerCAmelCase__			:str			,			lowerCAmelCase__			:List[Any]			,			lowerCAmelCase__			:Optional[int]=None			,			lowerCAmelCase__			:str = "test"			,			**lowerCAmelCase__			:Any	)		->			Tuple:
										__SCREAMING_SNAKE_CASE						:     Any					=		gen_kwargs.copy()

										__SCREAMING_SNAKE_CASE						:     Tuple					=		self.get_test_dataloader(lowercase_	)

										# Temporarily disable metric computation, we will do it in the loop here.
										__SCREAMING_SNAKE_CASE						:     List[Any]					=		self.compute_metrics
										__SCREAMING_SNAKE_CASE						:     Optional[Any]					=		None
										__SCREAMING_SNAKE_CASE						:     Any					=		time.time()
										__SCREAMING_SNAKE_CASE						:     Optional[int]					=		self.prediction_loop if self.args.use_legacy_prediction_loop else self.evaluation_loop
										try:
																__SCREAMING_SNAKE_CASE						:     Dict					=		eval_loop(
																    lowercase_			,			description='''Prediction'''			,			prediction_loss_only=True if compute_metrics is None else None			,			ignore_keys=lowercase_			,			metric_key_prefix=lowercase_			,			)
										finally:
																__SCREAMING_SNAKE_CASE						:     Optional[int]					=		compute_metrics

										__SCREAMING_SNAKE_CASE						:     Union[str, Any]					=		self.args.eval_batch_size * self.args.world_size
										if f'''{metric_key_prefix}_jit_compilation_time''' in output.metrics:
																start_time += output.metrics[f'''{metric_key_prefix}_jit_compilation_time''']
										output.metrics.update(
										    speed_metrics(
										        lowercase_			,			lowercase_			,			num_samples=output.num_samples			,			num_steps=math.ceil(output.num_samples / total_batch_size	)			,			)	)
										if self.post_process_function is None or self.compute_metrics is None:
																return output

										__SCREAMING_SNAKE_CASE						:     int					=		self.post_process_function(lowercase_			,			lowercase_			,			lowercase_			,			'''predict'''	)
										__SCREAMING_SNAKE_CASE						:     List[Any]					=		self.compute_metrics(lowercase_	)

										# Prefix all keys with metric_key_prefix + '_'
										for key in list(metrics.keys()	):
																if not key.startswith(f'''{metric_key_prefix}_'''	):
																						__SCREAMING_SNAKE_CASE						:     Tuple					=		metrics.pop(lowercase_	)
										metrics.update(output.metrics	)
										return PredictionOutput(predictions=predictions.predictions			,			label_ids=predictions.label_ids			,			metrics=lowercase_	)

from typing import Dict

import numpy as np
import torch

from . import residue_constants as rc
from .tensor_utils import tensor_tree_map, tree_map

def 			lowercase_				(			_lowerCamelCase							:       Dict[str, torch.Tensor]):
    lowercase__ :   Any    					=				[]
    lowercase__ :   Optional[int]    					=				[]
    lowercase__ :   Tuple    					=				[]

    for rt in rc.restypes:
        lowercase__ :   Dict    					=				rc.restype_name_to_atomaa_names[rc.restype_atoa[rt]]
        restype_atomaa_to_atomaa_list.append([(rc.atom_order[name] if name else 0) for name in atom_names])
        lowercase__ :   str    					=				{name: i for i, name in enumerate(_lowerCamelCase)}
        restype_atomaa_to_atomaa_list.append(
            [(atom_name_to_idxaa[name] if name in atom_name_to_idxaa else 0) for name in rc.atom_types])

        restype_atomaa_mask_list.append([(1.0 if name else 0.0) for name in atom_names])

    # Add dummy mapping for restype 'UNK'
    restype_atomaa_to_atomaa_list.append([0] * 14)
    restype_atomaa_to_atomaa_list.append([0] * 37)
    restype_atomaa_mask_list.append([0.0] * 14)

    lowercase__ :   Union[str, Any]    					=				torch.tensor(
        _lowerCamelCase		,	dtype=torch.intaa		,	device=protein["aatype"].device		,	)
    lowercase__ :   str    					=				torch.tensor(
        _lowerCamelCase		,	dtype=torch.intaa		,	device=protein["aatype"].device		,	)
    lowercase__ :   List[str]    					=				torch.tensor(
        _lowerCamelCase		,	dtype=torch.floataa		,	device=protein["aatype"].device		,	)
    lowercase__ :   str    					=				protein["aatype"].to(torch.long)

    # create the mapping for (residx, atom14) --> atom37, i.e. an array
    # with shape (num_res, 14) containing the atom37 indices for this protein
    lowercase__ :   Dict    					=				restype_atomaa_to_atomaa[protein_aatype]
    lowercase__ :   str    					=				restype_atomaa_mask[protein_aatype]

    lowercase__ :   List[Any]    					=				residx_atomaa_mask
    lowercase__ :   Optional[Any]    					=				residx_atomaa_to_atomaa.long()

    # create the gather indices for mapping back
    lowercase__ :   str    					=				restype_atomaa_to_atomaa[protein_aatype]
    lowercase__ :   str    					=				residx_atomaa_to_atomaa.long()

    # create the corresponding mask
    lowercase__ :   Optional[Any]    					=				torch.zeros([21, 37]		,	dtype=torch.floataa		,	device=protein["aatype"].device)
    for restype, restype_letter in enumerate(rc.restypes):
        lowercase__ :   Tuple    					=				rc.restype_atoa[restype_letter]
        lowercase__ :   List[Any]    					=				rc.residue_atoms[restype_name]
        for atom_name in atom_names:
            lowercase__ :   Optional[int]    					=				rc.atom_order[atom_name]
            lowercase__ :   Tuple    					=				1

    lowercase__ :   Dict    					=				restype_atomaa_mask[protein_aatype]
    lowercase__ :   Any    					=				residx_atomaa_mask

    return protein

def 			lowercase_				(			_lowerCamelCase							:       Dict[str, torch.Tensor]):
    lowercase__ :   Tuple    					=				tree_map(lambda _lowerCamelCase: torch.tensor(_lowerCamelCase		,	device=batch["aatype"].device)		,	_lowerCamelCase		,	np.ndarray)
    lowercase__ :   List[str]    					=				tensor_tree_map(lambda _lowerCamelCase: np.array(_lowerCamelCase)		,	make_atomaa_masks(_lowerCamelCase))
    return out



