




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 		UpperCamelCase__	(unittest.TestCase     ):

    '''simple docstring'''



    def     _lowercase     (    self     )     ->      List[str]:
           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=UpperCamelCase__     ,  )
           assert hasattr(self     ,  "env"     )







    def     _lowercase     (    self     ,  UpperCamelCase__     )     ->      Dict:
           # configuration for running training on smdistributed Model Parallel
           lowerCamelCase   :      Any			 =					{
               "enabled": True,
               "processes_per_host": 8,
           }
           lowerCamelCase   :      Union[str, Any]			 =					{
               "enabled": True,
               "parameters": {
                   "microbatches": 4,
                   "placement_strategy": "spread",
                   "pipeline": "interleaved",
                   "optimize": "speed",
                   "partitions": 4,
                   "ddp": True,
               },
           }

           lowerCamelCase   :      List[Any]			 =					{"smdistributed": {"modelparallel": smp_options}, "mpi": mpi_options}

           lowerCamelCase   :      Tuple			 =					"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=UpperCamelCase__     ,  instance_type=self.instance_type     ,  debugger_hook_config=UpperCamelCase__     ,  hyperparameters={
                   **self.env.hyperparameters,
                   "model_name_or_path": self.model_name_or_path,
                   "max_steps": 500,
               }     ,  metric_definitions=self.env.metric_definitions     ,  distribution=UpperCamelCase__     ,  py_version="py36"     ,  )







    def     _lowercase     (    self     ,  UpperCamelCase__     )     ->      Optional[Any]:
           TrainingJobAnalytics(UpperCamelCase__     ).export_csv(F'''{self.env.test_path}/{job_name}_metrics.csv'''     )







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

           # run training
           estimator.fit()

           # result dataframe
           lowerCamelCase   :      Any			 =					TrainingJobAnalytics(estimator.latest_training_job.name     ).dataframe()

           # extract kpis
           lowerCamelCase   :      List[Any]			 =					list(result_metrics_df[result_metrics_df.metric_name == "eval_accuracy"]["value"]     )
           lowerCamelCase   :      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   :      Any			 =					(
               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}     ,  UpperCamelCase__     )










import argparse

import intel_extension_for_pytorch as ipex
import torch

from diffusers import DPMSolverMultistepScheduler, StableDiffusionPipeline


_snake_case             =     argparse.ArgumentParser('''Stable Diffusion script with intel optimization''', add_help=False)
parser.add_argument('''--dpm''', action='''store_true''', help='''Enable DPMSolver or not''')
parser.add_argument('''--steps''', default=None, type=int, help='''Num inference steps''')
_snake_case             =     parser.parse_args()


_snake_case             =     '''cpu'''
_snake_case             =     '''a lovely <dicoo> in red dress and hat, in the snowly and brightly night, with many brighly buildings'''

_snake_case             =     '''path-to-your-trained-model'''
_snake_case             =     StableDiffusionPipeline.from_pretrained(model_id)
if args.dpm:
   _snake_case             =     DPMSolverMultistepScheduler.from_config(pipe.scheduler.config)
_snake_case             =     pipe.to(device)

# to channels last
_snake_case             =     pipe.unet.to(memory_format=torch.channels_last)
_snake_case             =     pipe.vae.to(memory_format=torch.channels_last)
_snake_case             =     pipe.text_encoder.to(memory_format=torch.channels_last)
if pipe.requires_safety_checker:
   _snake_case             =     pipe.safety_checker.to(memory_format=torch.channels_last)

# optimize with ipex
_snake_case             =     torch.randn(2, 4, 64, 64)
_snake_case             =     torch.rand(1) * 9_99
_snake_case             =     torch.randn(2, 77, 7_68)
_snake_case             =     (sample, timestep, encoder_hidden_status)
try:
   _snake_case             =     ipex.optimize(pipe.unet.eval(), dtype=torch.bfloataa, inplace=True, sample_input=input_example)
except Exception:
   _snake_case             =     ipex.optimize(pipe.unet.eval(), dtype=torch.bfloataa, inplace=True)
_snake_case             =     ipex.optimize(pipe.vae.eval(), dtype=torch.bfloataa, inplace=True)
_snake_case             =     ipex.optimize(pipe.text_encoder.eval(), dtype=torch.bfloataa, inplace=True)
if pipe.requires_safety_checker:
   _snake_case             =     ipex.optimize(pipe.safety_checker.eval(), dtype=torch.bfloataa, inplace=True)

# compute
_snake_case             =     6_66
_snake_case             =     torch.Generator(device).manual_seed(seed)
_snake_case             =     {'''generator''': generator}
if args.steps is not None:
   _snake_case             =     args.steps

with torch.cpu.amp.autocast(enabled=True, dtype=torch.bfloataa):
   _snake_case             =     pipe(prompt, **generate_kwargs).images[0]

# save image
image.save('''generated.png''')



