

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_5_0, """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_0_0, """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_0_0, """eval_accuracy""": 0.6, """eval_loss""": 0.7},
        },
    ]       )
class        a__       (   unittest.TestCase       ):







  """simple docstring"""




  def  _lowercase      (			self    : Union[str, Any]		)					->Optional[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=_lowerCAmelCase				,				)
        assert hasattr(self				,				"""env"""		)


  def  _lowercase      (			self    : str				,				UpperCAmelCase__    : Union[str, Any]		)					->Any:
        """simple docstring"""







        SCREAMING_SNAKE_CASE :      List[Any]       =   f"{self.env.base_job_name}-{instance_count}-{'ddp' if 'ddp' in self.script else 'smd'}"
        # distributed data settings
        SCREAMING_SNAKE_CASE :      Dict       =   {"""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=_lowerCAmelCase				,				instance_count=_lowerCAmelCase				,				instance_type=self.instance_type				,				debugger_hook_config=_lowerCAmelCase				,				hyperparameters={**self.env.distributed_hyperparameters, """model_name_or_path""": self.model_name_or_path}				,				metric_definitions=self.env.metric_definitions				,				distribution=_lowerCAmelCase				,				py_version="""py36"""				,				)


  def  _lowercase      (			self    : List[Any]				,				UpperCAmelCase__    : Any		)					->Dict:
        """simple docstring"""







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




  @parameterized.expand([(2,)]		)
  def  _lowercase      (			self    : List[str]				,				UpperCAmelCase__    : Optional[int]		)					->Union[str, Any]:
        """simple docstring"""







        SCREAMING_SNAKE_CASE :      Dict       =   self.create_estimator(_lowerCAmelCase		)

        # run training
        estimator.fit()

        # result dataframe
        SCREAMING_SNAKE_CASE :      List[str]       =   TrainingJobAnalytics(estimator.latest_training_job.name		).dataframe()

        # extract kpis
        SCREAMING_SNAKE_CASE :      str       =   list(result_metrics_df[result_metrics_df.metric_name == """eval_accuracy"""]["""value"""]		)
        SCREAMING_SNAKE_CASE :      str       =   list(result_metrics_df[result_metrics_df.metric_name == """eval_loss"""]["""value"""]		)
        # get train time from SageMaker job, this includes starting, preprocessing, stopping
        SCREAMING_SNAKE_CASE :      Dict       =   (
            Session().describe_training_job(estimator.latest_training_job.name		).get("""TrainingTimeInSeconds"""				,				9_9_9_9_9_9		)
        )

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







import inspect
from typing import Callable, List, Optional, Union

import torch
from transformers import CLIPImageProcessor, CLIPTextModel, CLIPTokenizer

from diffusers import DiffusionPipeline
from diffusers.models import AutoencoderKL, UNetaDConditionModel
from diffusers.pipelines.stable_diffusion import StableDiffusionPipelineOutput
from diffusers.pipelines.stable_diffusion.safety_checker import StableDiffusionSafetyChecker
from diffusers.schedulers import DDIMScheduler, LMSDiscreteScheduler, PNDMScheduler
from diffusers.utils import logging


UpperCAmelCase__ :     Any				    =    logging.get_logger(__name__)  # pylint: disable=invalid-name







class        a__       (   UpperCAmelCase       ):







  """simple docstring"""




  def __init__(			self    : Dict				,				UpperCAmelCase__    : AutoencoderKL				,				UpperCAmelCase__    : CLIPTextModel				,				UpperCAmelCase__    : CLIPTokenizer				,				UpperCAmelCase__    : UNetaDConditionModel				,				UpperCAmelCase__    : Union[DDIMScheduler, PNDMScheduler, LMSDiscreteScheduler]				,				UpperCAmelCase__    : StableDiffusionSafetyChecker				,				UpperCAmelCase__    : CLIPImageProcessor				,				)					->Optional[Any]:
        """simple docstring"""







        super().__init__()
        self.register_modules(
            vae=UpperCAmelCase__				,				text_encoder=UpperCAmelCase__				,				tokenizer=UpperCAmelCase__				,				unet=UpperCAmelCase__				,				scheduler=UpperCAmelCase__				,				safety_checker=UpperCAmelCase__				,				feature_extractor=UpperCAmelCase__				,				)


  def  _lowercase      (			self    : int				,				UpperCAmelCase__    : Optional[Union[str, int]] = "auto"		)					->Any:
        """simple docstring"""







        if slice_size == "auto":
              # half the attention head size is usually a good trade-off between
              # speed and memory
              SCREAMING_SNAKE_CASE :      int       =   self.unet.config.attention_head_dim // 2
        self.unet.set_attention_slice(UpperCAmelCase__		)


  def  _lowercase      (			self    : Dict		)					->Tuple:
        """simple docstring"""







        self.enable_attention_slicing(UpperCAmelCase__		)




  @torch.no_grad()
  def __call__(			self    : Optional[Any]				,				UpperCAmelCase__    : Union[str, List[str]]				,				UpperCAmelCase__    : int = 5_1_2				,				UpperCAmelCase__    : int = 5_1_2				,				UpperCAmelCase__    : int = 5_0				,				UpperCAmelCase__    : float = 7.5				,				UpperCAmelCase__    : Optional[Union[str, List[str]]] = None				,				UpperCAmelCase__    : Optional[int] = 1				,				UpperCAmelCase__    : float = 0.0				,				UpperCAmelCase__    : Optional[torch.Generator] = None				,				UpperCAmelCase__    : Optional[torch.FloatTensor] = None				,				UpperCAmelCase__    : Optional[str] = "pil"				,				UpperCAmelCase__    : bool = True				,				UpperCAmelCase__    : Optional[Callable[[int, int, torch.FloatTensor], None]] = None				,				UpperCAmelCase__    : int = 1				,				UpperCAmelCase__    : Optional[torch.FloatTensor] = None				,				**UpperCAmelCase__    : Tuple				,				)					->List[Any]:
        """simple docstring"""







        if isinstance(UpperCAmelCase__				,				UpperCAmelCase__		):
              SCREAMING_SNAKE_CASE :      Any       =   1
        elif isinstance(UpperCAmelCase__				,				UpperCAmelCase__		):
              SCREAMING_SNAKE_CASE :      Any       =   len(UpperCAmelCase__		)
        else:
              raise ValueError(f"`prompt` has to be of type `str` or `list` but is {type(UpperCAmelCase__		)}"		)

        if height % 8 != 0 or width % 8 != 0:
              raise ValueError(f"`height` and `width` have to be divisible by 8 but are {height} and {width}."		)

        if (callback_steps is None) or (
            callback_steps is not None and (not isinstance(UpperCAmelCase__				,				UpperCAmelCase__		) or callback_steps <= 0)
        ):
              raise ValueError(
                  f"`callback_steps` has to be a positive integer but is {callback_steps} of type"
                  f" {type(UpperCAmelCase__		)}."		)

        # get prompt text embeddings
        SCREAMING_SNAKE_CASE :      Optional[int]       =   self.tokenizer(
            UpperCAmelCase__				,				padding="""max_length"""				,				max_length=self.tokenizer.model_max_length				,				return_tensors="""pt"""				,				)
        SCREAMING_SNAKE_CASE :      str       =   text_inputs.input_ids

        if text_input_ids.shape[-1] > self.tokenizer.model_max_length:
              SCREAMING_SNAKE_CASE :      Union[str, Any]       =   self.tokenizer.batch_decode(text_input_ids[:, self.tokenizer.model_max_length :]		)
              logger.warning(
                  """The following part of your input was truncated because CLIP can only handle sequences up to"""
                  f" {self.tokenizer.model_max_length} tokens: {removed_text}"		)
              SCREAMING_SNAKE_CASE :      Tuple       =   text_input_ids[:, : self.tokenizer.model_max_length]

        if text_embeddings is None:
              SCREAMING_SNAKE_CASE :      Any       =   self.text_encoder(text_input_ids.to(self.device		)		)[0]

        # duplicate text embeddings for each generation per prompt, using mps friendly method
        SCREAMING_SNAKE_CASE     ,	SCREAMING_SNAKE_CASE     ,	SCREAMING_SNAKE_CASE :      str       =   text_embeddings.shape
        SCREAMING_SNAKE_CASE :      Tuple       =   text_embeddings.repeat(1				,				UpperCAmelCase__				,				1		)
        SCREAMING_SNAKE_CASE :      int       =   text_embeddings.view(bs_embed * num_images_per_prompt				,				UpperCAmelCase__				,				-1		)

        # here `guidance_scale` is defined analog to the guidance weight `w` of equation (2)
        # of the Imagen paper: https://arxiv.org/pdf/2205.11487.pdf . `guidance_scale = 1`
        # corresponds to doing no classifier free guidance.
        SCREAMING_SNAKE_CASE :      List[str]       =   guidance_scale > 1.0
        # get unconditional embeddings for classifier free guidance
        if do_classifier_free_guidance:
              SCREAMING_SNAKE_CASE :      List[str]
              if negative_prompt is None:
                    SCREAMING_SNAKE_CASE :      Union[str, Any]       =   [""""""]
              elif type(UpperCAmelCase__		) is not type(UpperCAmelCase__		):
                    raise TypeError(
                        f"`negative_prompt` should be the same type to `prompt`, but got {type(UpperCAmelCase__		)} !="
                        f" {type(UpperCAmelCase__		)}."		)
              elif isinstance(UpperCAmelCase__				,				UpperCAmelCase__		):
                    SCREAMING_SNAKE_CASE :      Union[str, Any]       =   [negative_prompt]
              elif batch_size != len(UpperCAmelCase__		):
                    raise ValueError(
                        f"`negative_prompt`: {negative_prompt} has batch size {len(UpperCAmelCase__		)}, but `prompt`:"
                        f" {prompt} has batch size {batch_size}. Please make sure that passed `negative_prompt` matches"
                        """ the batch size of `prompt`."""		)
              else:
                    SCREAMING_SNAKE_CASE :      Optional[int]       =   negative_prompt

              SCREAMING_SNAKE_CASE :      Dict       =   text_input_ids.shape[-1]
              SCREAMING_SNAKE_CASE :      int       =   self.tokenizer(
                  UpperCAmelCase__				,				padding="""max_length"""				,				max_length=UpperCAmelCase__				,				truncation=UpperCAmelCase__				,				return_tensors="""pt"""				,				)
              SCREAMING_SNAKE_CASE :      Union[str, Any]       =   self.text_encoder(uncond_input.input_ids.to(self.device		)		)[0]

              # duplicate unconditional embeddings for each generation per prompt, using mps friendly method
              SCREAMING_SNAKE_CASE :      Optional[int]       =   uncond_embeddings.shape[1]
              SCREAMING_SNAKE_CASE :      List[str]       =   uncond_embeddings.repeat(UpperCAmelCase__				,				UpperCAmelCase__				,				1		)
              SCREAMING_SNAKE_CASE :      int       =   uncond_embeddings.view(batch_size * num_images_per_prompt				,				UpperCAmelCase__				,				-1		)

              # For classifier free guidance, we need to do two forward passes.
              # Here we concatenate the unconditional and text embeddings into a single batch
              # to avoid doing two forward passes
              SCREAMING_SNAKE_CASE :      int       =   torch.cat([uncond_embeddings, text_embeddings]		)

        # get the initial random noise unless the user supplied it

        # Unlike in other pipelines, latents need to be generated in the target device
        # for 1-to-1 results reproducibility with the CompVis implementation.
        # However this currently doesn't work in `mps`.
        SCREAMING_SNAKE_CASE :      Dict       =   (batch_size * num_images_per_prompt, self.unet.config.in_channels, height // 8, width // 8)
        SCREAMING_SNAKE_CASE :      Dict       =   (batch_size * num_images_per_prompt, self.unet.config.in_channels, 6_4, 6_4)
        SCREAMING_SNAKE_CASE :      List[str]       =   text_embeddings.dtype
        if latents is None:
              if self.device.type == "mps":
                    # randn does not exist on mps
                    SCREAMING_SNAKE_CASE :      Union[str, Any]       =   torch.randn(
                        UpperCAmelCase__				,				generator=UpperCAmelCase__				,				device="""cpu"""				,				dtype=UpperCAmelCase__		).to(self.device		)
                    SCREAMING_SNAKE_CASE :      Dict       =   torch.randn(UpperCAmelCase__				,				generator=UpperCAmelCase__				,				device="""cpu"""				,				dtype=UpperCAmelCase__		).to(
                        self.device		)
              else:
                    SCREAMING_SNAKE_CASE :      List[Any]       =   torch.randn(
                        UpperCAmelCase__				,				generator=UpperCAmelCase__				,				device=self.device				,				dtype=UpperCAmelCase__		)
                    SCREAMING_SNAKE_CASE :      Any       =   torch.randn(UpperCAmelCase__				,				generator=UpperCAmelCase__				,				device=self.device				,				dtype=UpperCAmelCase__		)
        else:
              if latents_reference.shape != latents_shape:
                    raise ValueError(f"Unexpected latents shape, got {latents.shape}, expected {latents_shape}"		)
              SCREAMING_SNAKE_CASE :      List[str]       =   latents_reference.to(self.device		)
              SCREAMING_SNAKE_CASE :      Any       =   latents.to(self.device		)

        # This is the key part of the pipeline where we
        # try to ensure that the generated images w/ the same seed
        # but different sizes actually result in similar images
        SCREAMING_SNAKE_CASE :      Optional[Any]       =   (latents_shape[3] - latents_shape_reference[3]) // 2
        SCREAMING_SNAKE_CASE :      Tuple       =   (latents_shape[2] - latents_shape_reference[2]) // 2
        SCREAMING_SNAKE_CASE :      str       =   latents_shape_reference[3] if dx >= 0 else latents_shape_reference[3] + 2 * dx
        SCREAMING_SNAKE_CASE :      List[Any]       =   latents_shape_reference[2] if dy >= 0 else latents_shape_reference[2] + 2 * dy
        SCREAMING_SNAKE_CASE :      Optional[Any]       =   0 if dx < 0 else dx
        SCREAMING_SNAKE_CASE :      Any       =   0 if dy < 0 else dy
        SCREAMING_SNAKE_CASE :      int       =   max(-dx				,				0		)
        SCREAMING_SNAKE_CASE :      Optional[Any]       =   max(-dy				,				0		)
        # import pdb
        # pdb.set_trace()
        SCREAMING_SNAKE_CASE :      Tuple       =   latents_reference[:, :, dy : dy + h, dx : dx + w]

        # set timesteps
        self.scheduler.set_timesteps(UpperCAmelCase__		)

        # Some schedulers like PNDM have timesteps as arrays
        # It's more optimized to move all timesteps to correct device beforehand
        SCREAMING_SNAKE_CASE :      str       =   self.scheduler.timesteps.to(self.device		)

        # scale the initial noise by the standard deviation required by the scheduler
        SCREAMING_SNAKE_CASE :      Optional[int]       =   latents * self.scheduler.init_noise_sigma

        # prepare extra kwargs for the scheduler step, since not all schedulers have the same signature
        # eta (η) is only used with the DDIMScheduler, it will be ignored for other schedulers.
        # eta corresponds to η in DDIM paper: https://arxiv.org/abs/2010.02502
        # and should be between [0, 1]
        SCREAMING_SNAKE_CASE :      Any       =   """eta""" in set(inspect.signature(self.scheduler.step		).parameters.keys()		)
        SCREAMING_SNAKE_CASE :      str       =   {}
        if accepts_eta:
              SCREAMING_SNAKE_CASE :      Any       =   eta

        for i, t in enumerate(self.progress_bar(UpperCAmelCase__		)		):
              # expand the latents if we are doing classifier free guidance
              SCREAMING_SNAKE_CASE :      List[Any]       =   torch.cat([latents] * 2		) if do_classifier_free_guidance else latents
              SCREAMING_SNAKE_CASE :      Any       =   self.scheduler.scale_model_input(UpperCAmelCase__				,				UpperCAmelCase__		)

              # predict the noise residual
              SCREAMING_SNAKE_CASE :      Optional[int]       =   self.unet(UpperCAmelCase__				,				UpperCAmelCase__				,				encoder_hidden_states=UpperCAmelCase__		).sample

              # perform guidance
              if do_classifier_free_guidance:
                    SCREAMING_SNAKE_CASE     ,	SCREAMING_SNAKE_CASE :      Union[str, Any]       =   noise_pred.chunk(2		)
                    SCREAMING_SNAKE_CASE :      Optional[Any]       =   noise_pred_uncond + guidance_scale * (noise_pred_text - noise_pred_uncond)

              # compute the previous noisy sample x_t -> x_t-1
              SCREAMING_SNAKE_CASE :      Dict       =   self.scheduler.step(UpperCAmelCase__				,				UpperCAmelCase__				,				UpperCAmelCase__				,				**UpperCAmelCase__		).prev_sample

              # call the callback, if provided
              if callback is not None and i % callback_steps == 0:
                    callback(UpperCAmelCase__				,				UpperCAmelCase__				,				UpperCAmelCase__		)

        SCREAMING_SNAKE_CASE :      Union[str, Any]       =   1 / 0.1_82_15 * latents
        SCREAMING_SNAKE_CASE :      Tuple       =   self.vae.decode(UpperCAmelCase__		).sample

        SCREAMING_SNAKE_CASE :      Tuple       =   (image / 2 + 0.5).clamp(0				,				1		)

        # we always cast to float32 as this does not cause significant overhead and is compatible with bfloat16
        SCREAMING_SNAKE_CASE :      Dict       =   image.cpu().permute(0				,				2				,				3				,				1		).float().numpy()

        if self.safety_checker is not None:
              SCREAMING_SNAKE_CASE :      Tuple       =   self.feature_extractor(self.numpy_to_pil(UpperCAmelCase__		)				,				return_tensors="""pt"""		).to(
                  self.device		)
              SCREAMING_SNAKE_CASE     ,	SCREAMING_SNAKE_CASE :      Any       =   self.safety_checker(
                  images=UpperCAmelCase__				,				clip_input=safety_checker_input.pixel_values.to(text_embeddings.dtype		)		)
        else:
              SCREAMING_SNAKE_CASE :      Optional[Any]       =   None

        if output_type == "pil":
              SCREAMING_SNAKE_CASE :      Any       =   self.numpy_to_pil(UpperCAmelCase__		)

        if not return_dict:
              return (image, has_nsfw_concept)

        return StableDiffusionPipelineOutput(images=UpperCAmelCase__				,				nsfw_content_detected=UpperCAmelCase__		)




