





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''': 1_6_0_0, '''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''': 1_6_0_0, '''eval_accuracy''': 0.3, '''eval_loss''': 1.2},
        },
    ]						)
class 					A_					(   unittest.TestCase						):







       """simple docstring"""





       def  __UpperCAmelCase			(     self						:						Any	)    ->       Optional[int]:
         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						:						str    ,__A						:						Tuple	)    ->       int:
         # configuration for running training on smdistributed Model Parallel
         _lowercase						    =			{
             'enabled': True,
             'processes_per_host': 8,
         }
         _lowercase						    =			{
             'enabled': True,
             'parameters': {
                 'microbatches': 4,
                 'placement_strategy': 'spread',
                 'pipeline': 'interleaved',
                 'optimize': 'speed',
                 'partitions': 4,
                 'ddp': True,
             },
         }

         _lowercase						    =			{'smdistributed': {'modelparallel': smp_options}, 'mpi': mpi_options}

         _lowercase						    =			'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						:						List[Any]    ,__A						:						Any	)    ->       Optional[Any]:
         TrainingJobAnalytics(__A	).export_csv(F"""{self.env.test_path}/{job_name}_metrics.csv"""	)







       @parameterized.expand([(1,)]	)
       def  __UpperCAmelCase			(     self						:						Tuple    ,__A						:						Union[str, Any]	)    ->       Optional[Any]:
         # create estimator
         _lowercase						    =			self.create_estimator(__A	)

         # run training
         estimator.fit()

         # result dataframe
         _lowercase						    =			TrainingJobAnalytics(estimator.latest_training_job.name	).dataframe()

         # extract kpis
         _lowercase						    =			list(result_metrics_df[result_metrics_df.metric_name == 'eval_accuracy']['value']	)
         _lowercase						    =			list(result_metrics_df[result_metrics_df.metric_name == 'eval_loss']['value']	)
         # get train time from SageMaker job, this includes starting, preprocessing, stopping
         _lowercase						    =			(
             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	)


"""simple docstring"""


import qiskit



def   _A					(	_a	:						int     ,   _a	:						int				):



      """simple docstring"""






      A				  =		qiskit.Aer.get_backend("""aer_simulator"""				)

      # Create a Quantum Circuit acting on the q register
      A				  =		qiskit.QuantumCircuit(_a     ,   _a				)

      # 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				  =		qiskit.execute(_a     ,   _a     ,   shots=1_0_0_0				)

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


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




