

import json
import os
import subprocess
import unittest
from ast import literal_eval

import pytest
from parameterized import parameterized, parameterized_class

from . import is_sagemaker_available


if is_sagemaker_available():
 from sagemaker import Session, TrainingJobAnalytics
 from sagemaker.huggingface import HuggingFace







@pytest.mark.skipif(
    literal_eval(os.getenv("""TEST_SAGEMAKER"""		,   """False""" ) ) is not True		,   reason="""Skipping test because should only be run when releasing minor transformers version"""		,   )
@pytest.mark.usefixtures("""sm_env""" )
@parameterized_class(
    [
        {
            """framework""": """pytorch""",
            """script""": """run_glue_model_parallelism.py""",
            """model_name_or_path""": """roberta-large""",
            """instance_type""": """ml.p3dn.24xlarge""",
            """results""": {"""train_runtime""": 1600, """eval_accuracy""": 0.3, """eval_loss""": 1.2},
        },
        {
            """framework""": """pytorch""",
            """script""": """run_glue.py""",
            """model_name_or_path""": """roberta-large""",
            """instance_type""": """ml.p3dn.24xlarge""",
            """results""": {"""train_runtime""": 1600, """eval_accuracy""": 0.3, """eval_loss""": 1.2},
        },
    ] )
class   __lowercase      (unittest.TestCase ):
 def 	UpperCamelCase__  (							self					)					->Any:



       '''simple docstring'''





       if self.framework == "pytorch":
             subprocess.run(
                 f"""cp ./examples/pytorch/text-classification/run_glue.py {self.env.test_path}/run_glue.py""".split()      ,							encoding='''utf-8'''      ,							check=A_      ,							)
       assert hasattr(self      ,							'''env'''					)
 def 	UpperCamelCase__  (							self      ,							A_					)					->Tuple:



       '''simple docstring'''





       __lowerCAmelCase					:						List[Any]					=					{
           '''enabled''': True,
           '''processes_per_host''': 8,
       }
       __lowerCAmelCase					:						List[Any]					=					{
           '''enabled''': True,
           '''parameters''': {
               '''microbatches''': 4,
               '''placement_strategy''': '''spread''',
               '''pipeline''': '''interleaved''',
               '''optimize''': '''speed''',
               '''partitions''': 4,
               '''ddp''': True,
           },
       }

       __lowerCAmelCase					:						int					=					{'''smdistributed''': {'''modelparallel''': smp_options}, '''mpi''': mpi_options}

       __lowerCAmelCase					:						str					=					'''trainer''' if self.script == '''run_glue.py''' else '''smtrainer'''
       # creates estimator
       return HuggingFace(
           entry_point=self.script      ,							source_dir=self.env.test_path      ,							role=self.env.role      ,							image_uri=self.env.image_uri      ,							base_job_name=f"""{self.env.base_job_name}-{instance_count}-smp-{name_extension}"""      ,							instance_count=A_      ,							instance_type=self.instance_type      ,							debugger_hook_config=A_      ,							hyperparameters={
               **self.env.hyperparameters,
               '''model_name_or_path''': self.model_name_or_path,
               '''max_steps''': 500,
           }      ,							metric_definitions=self.env.metric_definitions      ,							distribution=A_      ,							py_version='''py36'''      ,							)
 def 	UpperCamelCase__  (							self      ,							A_					)					->Union[str, Any]:



       '''simple docstring'''





       TrainingJobAnalytics(A_					).export_csv(f"""{self.env.test_path}/{job_name}_metrics.csv"""					)





 @parameterized.expand([(1,)]					)
 def 	UpperCamelCase__  (							self      ,							A_					)					->List[Any]:



       '''simple docstring'''





       __lowerCAmelCase					:						Optional[int]					=					self.create_estimator(A_					)

       # run training
       estimator.fit()

       # result dataframe
       __lowerCAmelCase					:						Optional[Any]					=					TrainingJobAnalytics(estimator.latest_training_job.name					).dataframe()

       # extract kpis
       __lowerCAmelCase					:						List[str]					=					list(result_metrics_df[result_metrics_df.metric_name == '''eval_accuracy''']['''value''']					)
       __lowerCAmelCase					:						Optional[Any]					=					list(result_metrics_df[result_metrics_df.metric_name == '''eval_loss''']['''value''']					)
       # get train time from SageMaker job, this includes starting, preprocessing, stopping
       __lowerCAmelCase					:						Optional[int]					=					(
           Session().describe_training_job(estimator.latest_training_job.name					).get('''TrainingTimeInSeconds'''      ,							99_9999					)
       )

       # assert kpis
       assert train_runtime <= self.results["train_runtime"]
       assert all(t >= self.results['''eval_accuracy'''] for t in eval_accuracy					)
       assert all(t <= self.results['''eval_loss'''] for t in eval_loss					)

       # dump tests result into json file to share in PR
       with open(f"""{estimator.latest_training_job.name}.json"""      ,							'''w'''					) as outfile:
             json.dump({'''train_time''': train_runtime, '''eval_accuracy''': eval_accuracy, '''eval_loss''': eval_loss}      ,							A_					)





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 (     lowercase__						):
      __lowerCAmelCase					:						str					=					[]
      __lowerCAmelCase					:						List[Any]					=					[]
      __lowerCAmelCase					:						str					=					[]

      for rt in rc.restypes:
            __lowerCAmelCase					:						List[Any]					=					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]						)
            __lowerCAmelCase					:						List[str]					=					{name: i for i, name in enumerate(lowercase__						)}
            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] * 1_4						)
      restype_atomaa_to_atomaa_list.append([0] * 3_7						)
      restype_atomaa_mask_list.append([0.0] * 1_4						)

      __lowerCAmelCase					:						List[Any]					=					torch.tensor(
          lowercase__   ,	dtype=torch.intaa   ,	device=protein['''aatype'''].device   ,	)
      __lowerCAmelCase					:						Optional[Any]					=					torch.tensor(
          lowercase__   ,	dtype=torch.intaa   ,	device=protein['''aatype'''].device   ,	)
      __lowerCAmelCase					:						Tuple					=					torch.tensor(
          lowercase__   ,	dtype=torch.floataa   ,	device=protein['''aatype'''].device   ,	)
      __lowerCAmelCase					:						List[Any]					=					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
      __lowerCAmelCase					:						Any					=					restype_atomaa_to_atomaa[protein_aatype]
      __lowerCAmelCase					:						Union[str, Any]					=					restype_atomaa_mask[protein_aatype]

      __lowerCAmelCase					:						int					=					residx_atomaa_mask
      __lowerCAmelCase					:						List[str]					=					residx_atomaa_to_atomaa.long()

      # create the gather indices for mapping back
      __lowerCAmelCase					:						int					=					restype_atomaa_to_atomaa[protein_aatype]
      __lowerCAmelCase					:						Union[str, Any]					=					residx_atomaa_to_atomaa.long()

      # create the corresponding mask
      __lowerCAmelCase					:						str					=					torch.zeros([2_1, 3_7]   ,	dtype=torch.floataa   ,	device=protein['''aatype'''].device						)
      for restype, restype_letter in enumerate(rc.restypes						):
            __lowerCAmelCase					:						Optional[int]					=					rc.restype_atoa[restype_letter]
            __lowerCAmelCase					:						Optional[Any]					=					rc.residue_atoms[restype_name]
            for atom_name in atom_names:
                  __lowerCAmelCase					:						str					=					rc.atom_order[atom_name]
                  __lowerCAmelCase					:						List[Any]					=					1

      __lowerCAmelCase					:						Union[str, Any]					=					restype_atomaa_mask[protein_aatype]
      __lowerCAmelCase					:						Any					=					residx_atomaa_mask

      return protein







def 		_lowercase (     lowercase__						):
      __lowerCAmelCase					:						Dict					=					tree_map(lambda lowercase__						: torch.tensor(lowercase__   ,	device=batch['''aatype'''].device						)   ,	lowercase__   ,	np.ndarray						)
      __lowerCAmelCase					:						Tuple					=					tensor_tree_map(lambda lowercase__						: np.array(lowercase__						)   ,	make_atomaa_masks(lowercase__						)						)
      return out


