




"""simple docstring"""






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.py""",
            """model_name_or_path""": """distilbert-base-cased""",
            """instance_type""": """ml.p3.16xlarge""",
            """results""": {"""train_runtime""": 6_50, """eval_accuracy""": 0.7, """eval_loss""": 0.6},
        },
        {
            """framework""": """pytorch""",
            """script""": """run_ddp.py""",
            """model_name_or_path""": """distilbert-base-cased""",
            """instance_type""": """ml.p3.16xlarge""",
            """results""": {"""train_runtime""": 6_00, """eval_accuracy""": 0.7, """eval_loss""": 0.6},
        },
        {
            """framework""": """tensorflow""",
            """script""": """run_tf_dist.py""",
            """model_name_or_path""": """distilbert-base-cased""",
            """instance_type""": """ml.p3.16xlarge""",
            """results""": {"""train_runtime""": 6_00, """eval_accuracy""": 0.6, """eval_loss""": 0.7},
        },
    ]		)
class lowerCAmelCase__      (	unittest.TestCase		):






			'''simple docstring'''



			def    A_       (    self    ):
							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=lowercase    , )
							assert hasattr(self    , 'env'    )
			def    A_       (    self    , lowercase    ):
							_lowerCamelCase	:     Optional[Any]       =       F'''{self.env.base_job_name}-{instance_count}-{'ddp' if 'ddp' in self.script else 'smd'}'''
							# distributed data settings
							_lowerCamelCase	:     Union[str, Any]       =       {'smdistributed': {'dataparallel': {'enabled': True}}} if self.script != 'run_ddp.py' else None

							# 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=lowercase    , instance_count=lowercase    , instance_type=self.instance_type    , debugger_hook_config=lowercase    , hyperparameters={**self.env.distributed_hyperparameters, 'model_name_or_path': self.model_name_or_path}    , metric_definitions=self.env.metric_definitions    , distribution=lowercase    , py_version='py36'    , )
			def    A_       (    self    , lowercase    ):
							TrainingJobAnalytics(lowercase    ).export_csv(F'''{self.env.test_path}/{job_name}_metrics.csv'''    )






			@parameterized.expand([(2,)]    )
			def    A_       (    self    , lowercase    ):
							# create estimator
							_lowerCamelCase	:     List[Any]       =       self.create_estimator(lowercase    )

							# run training
							estimator.fit()

							# result dataframe
							_lowerCamelCase	:     Tuple       =       TrainingJobAnalytics(estimator.latest_training_job.name    ).dataframe()

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

							# 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}    , lowercase    )





"""simple docstring"""






import qiskit







def 	_snake_case     (							lowercase__ = 2						):
				_lowerCamelCase	:     Optional[Any]       =       qubits

				# Using Aer's simulator
				_lowerCamelCase	:     Dict       =       qiskit.Aer.get_backend('aer_simulator'						)

				# Creating a Quantum Circuit acting on the q register
				_lowerCamelCase	:     Dict       =       qiskit.QuantumCircuit(lowercase__				,       lowercase__						)

				# Adding a H gate on qubit 0 (now q0 in superposition)
				circuit.h(0						)

				for i in range(1				,       lowercase__						):
								# Adding CX (CNOT) gate
								circuit.cx(i - 1				,       lowercase__						)

				# Mapping the quantum measurement to the classical bits
				circuit.measure(list(range(lowercase__						)						)				,       list(range(lowercase__						)						)						)

				# Now measuring any one qubit would affect other qubits to collapse
				# their super position and have same state as the measured one.

				# Executing the circuit on the simulator
				_lowerCamelCase	:     Optional[Any]       =       qiskit.execute(lowercase__				,       lowercase__				,       shots=1000						)

				return job.result().get_counts(lowercase__						)


if __name__ == "__main__":
			print(F"Total count for various states are: {quantum_entanglement(3)}")