
'''simple docstring'''

import csv
import json
import logging
import os
import re
import subprocess
import tempfile
import unittest
import zipfile
from pathlib import Path
from typing import Optional
from unittest import mock

import torch

# We use TF to parse the logs
from accelerate import Accelerator
from accelerate.test_utils.testing import (
    MockingTestCase,
    TempDirTestCase,
    require_comet_ml,
    require_tensorboard,
    require_wandb,
    skip,
)
from accelerate.tracking import CometMLTracker, GeneralTracker
from accelerate.utils import ProjectConfiguration, is_comet_ml_available, is_tensorboard_available


if is_comet_ml_available():
   from comet_ml import OfflineExperiment

if is_tensorboard_available():
   import struct

   import tensorboard.compat.proto.event_pba as event_pba

lowerCAmelCase_			:			List[Any]			   =     logging.getLogger(__name__)







@require_tensorboard
class 	__lowerCAmelCase    (  unittest.TestCase				):



     def     snake_case_ (self	):
          _UpperCAmelCase					:  Dict         =    """test_project_with_config"""
          with tempfile.TemporaryDirectory() as dirpath:
               _UpperCAmelCase					:  Optional[int]         =    Accelerator(log_with="""tensorboard"""						,	project_dir=lowerCAmelCase__	)
               _UpperCAmelCase					:  Any         =    {"""num_iterations""": 1_2, """learning_rate""": 1e-2, """some_boolean""": False, """some_string""": """some_value"""}
               accelerator.init_trackers(lowerCAmelCase__						,	lowerCAmelCase__	)
               accelerator.end_training()
               for child in Path(F"{dirpath}/{project_name}"	).glob("""*/**"""	):
                    _UpperCAmelCase					:  Any         =    list(filter(lambda lowerCAmelCase__	: x.is_file()						,	child.iterdir()	)	)[0]
               self.assertNotEqual(str(lowerCAmelCase__	)						,	""""""	)



     def     snake_case_ (self	):
          _UpperCAmelCase					:  Any         =    """test_project_with_log"""
          with tempfile.TemporaryDirectory() as dirpath:
               _UpperCAmelCase					:  Tuple         =    Accelerator(log_with="""tensorboard"""						,	project_dir=lowerCAmelCase__	)
               accelerator.init_trackers(lowerCAmelCase__	)
               _UpperCAmelCase					:  str         =    {"""total_loss""": 0.1, """iteration""": 1, """my_text""": """some_value"""}
               accelerator.log(lowerCAmelCase__						,	step=0	)
               accelerator.end_training()
               # Logged values are stored in the outermost-tfevents file and can be read in as a TFRecord
               # Names are randomly generated each time
               _UpperCAmelCase					:  Tuple         =    list(filter(lambda lowerCAmelCase__	: x.is_file()						,	Path(F"{dirpath}/{project_name}"	).iterdir()	)	)[0]
               self.assertNotEqual(str(lowerCAmelCase__	)						,	""""""	)



     def     snake_case_ (self	):
          _UpperCAmelCase					:  List[str]         =    """test_project_with_log"""
          with tempfile.TemporaryDirectory() as dirpath:
               _UpperCAmelCase					:  List[str]         =    Accelerator(log_with="""tensorboard"""						,	project_dir=lowerCAmelCase__	)
               accelerator.init_trackers(lowerCAmelCase__	)
               _UpperCAmelCase					:  int         =    {"""tensor""": torch.tensor(1	)}
               accelerator.log(lowerCAmelCase__						,	step=0	)
               accelerator.end_training()
               # Logged values are stored in the outermost-tfevents file and can be read in as a TFRecord
               # Names are randomly generated each time
               _UpperCAmelCase					:  Any         =    list(filter(lambda lowerCAmelCase__	: x.is_file()						,	Path(F"{dirpath}/{project_name}"	).iterdir()	)	)[0]
               # Reading implementation based on https://github.com/pytorch/pytorch/issues/45327#issuecomment-703757685
               with open(lowerCAmelCase__						,	"""rb"""	) as f:
                    _UpperCAmelCase					:  Dict         =    f.read()
               _UpperCAmelCase					:  Dict         =    False
               while data:
                    _UpperCAmelCase					:  str         =    struct.unpack("""Q"""						,	data[:8]	)

                    _UpperCAmelCase					:  Tuple         =    data[1_2 : 1_2 + int(header[0]	)]  # 8+4
                    _UpperCAmelCase					:  Tuple         =    data[1_2 + int(header[0]	) + 4 :]
                    _UpperCAmelCase					:  Any         =    event_pba.Event()

                    event.ParseFromString(lowerCAmelCase__	)
                    if event.HasField("""summary"""	):
                         for value in event.summary.value:
                              if value.simple_value == 1.0 and value.tag == "tensor":
                                   _UpperCAmelCase					:  Optional[int]         =    True
               self.assertTrue(lowerCAmelCase__						,	"""Converted tensor was not found in the log file!"""	)



     def     snake_case_ (self	):
          with self.assertRaisesRegex(lowerCAmelCase__						,	"""Logging with `tensorboard` requires a `logging_dir`"""	):
               _UpperCAmelCase					:  List[Any]         =    Accelerator(log_with="""tensorboard"""	)
          with tempfile.TemporaryDirectory() as dirpath:
               _UpperCAmelCase					:  Dict         =    Accelerator(log_with="""tensorboard"""						,	project_dir=lowerCAmelCase__	)



     def     snake_case_ (self	):
          _UpperCAmelCase					:  List[str]         =    ProjectConfiguration(total_limit=3_0	)
          with tempfile.TemporaryDirectory() as dirpath:
               _UpperCAmelCase					:  Optional[int]         =    Accelerator(log_with="""tensorboard"""						,	project_dir=lowerCAmelCase__						,	project_config=lowerCAmelCase__	)







@require_wandb
@mock.patch.dict(os.environ       ,					{"""WANDB_MODE""": """offline"""}				)
class 	__lowerCAmelCase    (  __a       ,					__a				):



     def     snake_case_ (self	):
          super().setUp()
          # wandb let's us override where logs are stored to via the WANDB_DIR env var
          self.add_mocks(mock.patch.dict(os.environ						,	{"""WANDB_DIR""": self.tmpdir}	)	)



     @staticmethod
     def     snake_case_ (lowerCAmelCase__						,	lowerCAmelCase__						,	lowerCAmelCase__ = True	):
          _UpperCAmelCase					:  Optional[Any]         =    rF"{section} ([\S\s]*?)\n\n"
          if record:
               _UpperCAmelCase					:  Optional[int]         =    rF"Record: {pattern}"
          _UpperCAmelCase					:  int         =    re.findall(lowerCAmelCase__						,	lowerCAmelCase__	)[0]
          # A config
          if section == "config" or section == "history":
               _UpperCAmelCase					:  Any         =    re.findall(r"""\"([a-zA-Z0-9_.,]+)"""						,	lowerCAmelCase__	)
               return {key: val for key, val in zip(cleaned_record[0::2]						,	cleaned_record[1::2]	)}
          # Everything else
          else:
               return dict(re.findall(r"""(\w+): \"([^\s]+)\""""						,	lowerCAmelCase__	)	)



     @skip
     def     snake_case_ (self	):
          _UpperCAmelCase					:  Dict         =    """test_project_with_config"""
          _UpperCAmelCase					:  Any         =    Accelerator(log_with="""wandb"""	)
          _UpperCAmelCase					:  Optional[int]         =    {"""num_iterations""": 1_2, """learning_rate""": 1e-2, """some_boolean""": False, """some_string""": """some_value"""}
          _UpperCAmelCase					:  Optional[Any]         =    {"""wandb""": {"""tags""": ["""my_tag"""]}}
          accelerator.init_trackers(lowerCAmelCase__						,	lowerCAmelCase__						,	lowerCAmelCase__	)
          _UpperCAmelCase					:  Tuple         =    {"""total_loss""": 0.1, """iteration""": 1, """my_text""": """some_value"""}
          accelerator.log(lowerCAmelCase__						,	step=0	)
          accelerator.end_training()
          # The latest offline log is stored at wandb/latest-run/*.wandb
          for child in Path(F"{self.tmpdir}/wandb/latest-run"	).glob("""*"""	):
               if child.is_file() and child.suffix == ".wandb":
                    _UpperCAmelCase					:  Dict         =    subprocess.check_output(
                        ["""wandb""", """sync""", """--view""", """--verbose""", str(lowerCAmelCase__	)]						,	env=os.environ.copy()	).decode("""utf8"""						,	"""ignore"""	)
                    break

        # Check HPS through careful parsing and cleaning
          _UpperCAmelCase					:  Tuple         =    self.parse_log(lowerCAmelCase__						,	"""config"""	)
          self.assertEqual(logged_items["""num_iterations"""]						,	"""12"""	)
          self.assertEqual(logged_items["""learning_rate"""]						,	"""0.01"""	)
          self.assertEqual(logged_items["""some_boolean"""]						,	"""false"""	)
          self.assertEqual(logged_items["""some_string"""]						,	"""some_value"""	)
          self.assertEqual(logged_items["""some_string"""]						,	"""some_value"""	)

          # Run tags
          _UpperCAmelCase					:  Any         =    self.parse_log(lowerCAmelCase__						,	"""run"""						,	lowerCAmelCase__	)
          self.assertEqual(logged_items["""tags"""]						,	"""my_tag"""	)

          # Actual logging
          _UpperCAmelCase					:  List[str]         =    self.parse_log(lowerCAmelCase__						,	"""history"""	)
          self.assertEqual(logged_items["""total_loss"""]						,	"""0.1"""	)
          self.assertEqual(logged_items["""iteration"""]						,	"""1"""	)
          self.assertEqual(logged_items["""my_text"""]						,	"""some_value"""	)
          self.assertEqual(logged_items["""_step"""]						,	"""0"""	)







def __A  (    self ,       lowerCAmelCase_ ,       lowerCAmelCase_       ):
     _UpperCAmelCase					:  Optional[int]         =    run_name
     _UpperCAmelCase					:  Union[str, Any]         =    OfflineExperiment(project_name=lowerCAmelCase_ ,       offline_directory=lowerCAmelCase_       )
     logger.info(f"Initialized offline CometML project {self.run_name}"       )
     logger.info("""Make sure to log any initial configurations with `self.store_init_configuration` before training!"""       )







@require_comet_ml
@mock.patch.object(__a       ,					"""__init__"""       ,					__a				)
class 	__lowerCAmelCase    (  unittest.TestCase				):



     @staticmethod
     def     snake_case_ (lowerCAmelCase__						,	lowerCAmelCase__						,	lowerCAmelCase__ = False	):
          for log in log_list:
               _UpperCAmelCase					:  Tuple         =    json.loads(lowerCAmelCase__	)["""payload"""]
               if is_param and "param" in j.keys():
                    if j["param"]["paramName"] == key:
                         return j["param"]["paramValue"]
               if "log_other" in j.keys():
                    if j["log_other"]["key"] == key:
                         return j["log_other"]["val"]
               if "metric" in j.keys():
                    if j["metric"]["metricName"] == key:
                         return j["metric"]["metricValue"]



     def     snake_case_ (self	):
          with tempfile.TemporaryDirectory() as d:
               _UpperCAmelCase					:  Any         =    CometMLTracker("""test_project_with_config"""						,	lowerCAmelCase__	)
               _UpperCAmelCase					:  Tuple         =    Accelerator(log_with=lowerCAmelCase__	)
               _UpperCAmelCase					:  Dict         =    {"""num_iterations""": 1_2, """learning_rate""": 1e-2, """some_boolean""": False, """some_string""": """some_value"""}
               accelerator.init_trackers(lowerCAmelCase__						,	lowerCAmelCase__	)
               accelerator.end_training()
               _UpperCAmelCase					:  List[Any]         =    os.listdir(lowerCAmelCase__	)[0]  # Comet is nice, it's just a zip file here
               # We parse the raw logs
               _UpperCAmelCase					:  Dict         =    os.path.join(lowerCAmelCase__						,	lowerCAmelCase__	)
               _UpperCAmelCase					:  List[str]         =    zipfile.ZipFile(lowerCAmelCase__						,	"""r"""	)
               _UpperCAmelCase					:  Union[str, Any]         =    archive.open("""messages.json"""	).read().decode("""utf-8"""	)
          _UpperCAmelCase					:  Optional[int]         =    log.split("""\n"""	)[:-1]
          self.assertEqual(self.get_value_from_key(lowerCAmelCase__						,	"""num_iterations"""						,	lowerCAmelCase__	)						,	1_2	)
          self.assertEqual(self.get_value_from_key(lowerCAmelCase__						,	"""learning_rate"""						,	lowerCAmelCase__	)						,	0.0_1	)
          self.assertEqual(self.get_value_from_key(lowerCAmelCase__						,	"""some_boolean"""						,	lowerCAmelCase__	)						,	lowerCAmelCase__	)
          self.assertEqual(self.get_value_from_key(lowerCAmelCase__						,	"""some_string"""						,	lowerCAmelCase__	)						,	"""some_value"""	)



     def     snake_case_ (self	):
          with tempfile.TemporaryDirectory() as d:
               _UpperCAmelCase					:  List[str]         =    CometMLTracker("""test_project_with_config"""						,	lowerCAmelCase__	)
               _UpperCAmelCase					:  Optional[int]         =    Accelerator(log_with=lowerCAmelCase__	)
               accelerator.init_trackers(lowerCAmelCase__	)
               _UpperCAmelCase					:  str         =    {"""total_loss""": 0.1, """iteration""": 1, """my_text""": """some_value"""}
               accelerator.log(lowerCAmelCase__						,	step=0	)
               accelerator.end_training()
               _UpperCAmelCase					:  Any         =    os.listdir(lowerCAmelCase__	)[0]  # Comet is nice, it's just a zip file here
               # We parse the raw logs
               _UpperCAmelCase					:  int         =    os.path.join(lowerCAmelCase__						,	lowerCAmelCase__	)
               _UpperCAmelCase					:  List[str]         =    zipfile.ZipFile(lowerCAmelCase__						,	"""r"""	)
               _UpperCAmelCase					:  Union[str, Any]         =    archive.open("""messages.json"""	).read().decode("""utf-8"""	)
          _UpperCAmelCase					:  Optional[Any]         =    log.split("""\n"""	)[:-1]
          self.assertEqual(self.get_value_from_key(lowerCAmelCase__						,	"""curr_step"""						,	lowerCAmelCase__	)						,	0	)
          self.assertEqual(self.get_value_from_key(lowerCAmelCase__						,	"""total_loss"""	)						,	0.1	)
          self.assertEqual(self.get_value_from_key(lowerCAmelCase__						,	"""iteration"""	)						,	1	)
          self.assertEqual(self.get_value_from_key(lowerCAmelCase__						,	"""my_text"""	)						,	"""some_value"""	)







class 	__lowerCAmelCase    (  __a				):
     snake_case   :    Dict							      =      [
         """total_loss""",
         """iteration""",
         """my_text""",
         """learning_rate""",
         """num_iterations""",
         """some_boolean""",
         """some_string""",
     ]

     snake_case   :    Any							      =      """my_custom_tracker"""
     snake_case   :    Tuple							      =      False



     def __init__(self						,	lowerCAmelCase__	):
          _UpperCAmelCase					:  str         =    open(F"{dir}/log.csv"						,	"""w+"""	)
          _UpperCAmelCase					:  List[Any]         =    csv.DictWriter(self.f						,	fieldnames=self._col_names	)
          self.writer.writeheader()



     @property
     def     snake_case_ (self	):
          return self.writer



     def     snake_case_ (self						,	lowerCAmelCase__	):
          logger.info("""Call init"""	)
          self.writer.writerow(lowerCAmelCase__	)



     def     snake_case_ (self						,	lowerCAmelCase__						,	lowerCAmelCase__	):
          logger.info("""Call log"""	)
          self.writer.writerow(lowerCAmelCase__	)



     def     snake_case_ (self	):
          self.f.close()







class 	__lowerCAmelCase    (  unittest.TestCase				):



     def     snake_case_ (self	):
          with tempfile.TemporaryDirectory() as d:
               _UpperCAmelCase					:  str         =    MyCustomTracker(lowerCAmelCase__	)
               _UpperCAmelCase					:  Any         =    Accelerator(log_with=lowerCAmelCase__	)
               _UpperCAmelCase					:  List[str]         =    {"""num_iterations""": 1_2, """learning_rate""": 1e-2, """some_boolean""": False, """some_string""": """some_value"""}
               accelerator.init_trackers("""Some name"""						,	lowerCAmelCase__	)
               accelerator.end_training()
               with open(F"{d}/log.csv"						,	"""r"""	) as f:
                    _UpperCAmelCase					:  Any         =    csv.DictReader(lowerCAmelCase__	)
                    _UpperCAmelCase					:  Optional[Any]         =    next(lowerCAmelCase__	)
                    _UpperCAmelCase					:  Optional[Any]         =    {
                        """total_loss""": """""",
                        """iteration""": """""",
                        """my_text""": """""",
                        """learning_rate""": """0.01""",
                        """num_iterations""": """12""",
                        """some_boolean""": """False""",
                        """some_string""": """some_value""",
                    }
                    self.assertDictEqual(lowerCAmelCase__						,	lowerCAmelCase__	)



     def     snake_case_ (self	):
          with tempfile.TemporaryDirectory() as d:
               _UpperCAmelCase					:  Any         =    MyCustomTracker(lowerCAmelCase__	)
               _UpperCAmelCase					:  Any         =    Accelerator(log_with=lowerCAmelCase__	)
               accelerator.init_trackers("""Some name"""	)
               _UpperCAmelCase					:  List[str]         =    {"""total_loss""": 0.1, """iteration""": 1, """my_text""": """some_value"""}
               accelerator.log(lowerCAmelCase__						,	step=0	)
               accelerator.end_training()
               with open(F"{d}/log.csv"						,	"""r"""	) as f:
                    _UpperCAmelCase					:  Optional[Any]         =    csv.DictReader(lowerCAmelCase__	)
                    _UpperCAmelCase					:  Dict         =    next(lowerCAmelCase__	)
                    _UpperCAmelCase					:  str         =    {
                        """total_loss""": """0.1""",
                        """iteration""": """1""",
                        """my_text""": """some_value""",
                        """learning_rate""": """""",
                        """num_iterations""": """""",
                        """some_boolean""": """""",
                        """some_string""": """""",
                    }
                    self.assertDictEqual(lowerCAmelCase__						,	lowerCAmelCase__	)



import gc
import unittest

import numpy as np
import torch
from transformers import CLIPTextConfig, CLIPTextModel, CLIPTokenizer

from diffusers import (
    AutoencoderKL,
    DDIMParallelScheduler,
    DDPMParallelScheduler,
    StableDiffusionParadigmsPipeline,
    UNetaDConditionModel,
)
from diffusers.utils import slow, torch_device
from diffusers.utils.testing_utils import (
    enable_full_determinism,
    require_torch_gpu,
)

from ..pipeline_params import TEXT_TO_IMAGE_BATCH_PARAMS, TEXT_TO_IMAGE_IMAGE_PARAMS, TEXT_TO_IMAGE_PARAMS
from ..test_pipelines_common import PipelineLatentTesterMixin, PipelineTesterMixin


enable_full_determinism()




class       __lowerCAmelCase     (				a_,		a_,		unittest.TestCase  ):



 '''simple docstring'''
 _lowerCamelCase  :		Any 			=       StableDiffusionParadigmsPipeline
 _lowerCamelCase  :		Any 			=       TEXT_TO_IMAGE_PARAMS
 _lowerCamelCase  :		str 			=       TEXT_TO_IMAGE_BATCH_PARAMS
 _lowerCamelCase  :		int 			=       TEXT_TO_IMAGE_IMAGE_PARAMS
 _lowerCamelCase  :		List[Any] 			=       TEXT_TO_IMAGE_IMAGE_PARAMS

 def     __snake_case    (	self				:						int     )       ->							List[str]:
        torch.manual_seed(0     )
        _UpperCamelCase   :				Union[str, Any]       =     UNetaDConditionModel(
            block_out_channels=(3_2, 6_4)				,	layers_per_block=2				,	sample_size=3_2				,	in_channels=4				,	out_channels=4				,	down_block_types=('DownBlock2D', 'CrossAttnDownBlock2D')				,	up_block_types=('CrossAttnUpBlock2D', 'UpBlock2D')				,	cross_attention_dim=3_2				,	attention_head_dim=(2, 4)				,	use_linear_projection=lowerCamelCase				,	)
        _UpperCamelCase   :				str       =     DDIMParallelScheduler(
            beta_start=0.0_0_0_8_5				,	beta_end=0.0_1_2				,	beta_schedule='scaled_linear'				,	clip_sample=lowerCamelCase				,	set_alpha_to_one=lowerCamelCase				,	)
        torch.manual_seed(0     )
        _UpperCamelCase   :				str       =     AutoencoderKL(
            block_out_channels=[3_2, 6_4]				,	in_channels=3				,	out_channels=3				,	down_block_types=['DownEncoderBlock2D', 'DownEncoderBlock2D']				,	up_block_types=['UpDecoderBlock2D', 'UpDecoderBlock2D']				,	latent_channels=4				,	sample_size=1_2_8				,	)
        torch.manual_seed(0     )
        _UpperCamelCase   :				int       =     CLIPTextConfig(
            bos_token_id=0				,	eos_token_id=2				,	hidden_size=3_2				,	intermediate_size=3_7				,	layer_norm_eps=1e-05				,	num_attention_heads=4				,	num_hidden_layers=5				,	pad_token_id=1				,	vocab_size=1_0_0_0				,	hidden_act='gelu'				,	projection_dim=5_1_2				,	)
        _UpperCamelCase   :				Any       =     CLIPTextModel(lowerCamelCase     )
        _UpperCamelCase   :				Optional[Any]       =     CLIPTokenizer.from_pretrained('hf-internal-testing/tiny-random-clip'     )

        _UpperCamelCase   :				Tuple       =     {
            'unet': unet,
            'scheduler': scheduler,
            'vae': vae,
            'text_encoder': text_encoder,
            'tokenizer': tokenizer,
            'safety_checker': None,
            'feature_extractor': None,
        }
        return components

 def     __snake_case    (	self				:						List[Any]				,	lowerCamelCase				:						Any				,	lowerCamelCase				:						str=0     )       ->							Dict:
        if str(lowerCamelCase     ).startswith('mps'     ):
               _UpperCamelCase   :				List[Any]       =     torch.manual_seed(lowerCamelCase     )
        else:
               _UpperCamelCase   :				Union[str, Any]       =     torch.Generator(device=lowerCamelCase     ).manual_seed(lowerCamelCase     )
        _UpperCamelCase   :				Optional[Any]       =     {
            'prompt': 'a photograph of an astronaut riding a horse',
            'generator': generator,
            'num_inference_steps': 1_0,
            'guidance_scale': 6.0,
            'output_type': 'numpy',
            'parallel': 3,
            'debug': True,
        }
        return inputs

 def     __snake_case    (	self				:						List[Any]     )       ->							str:
        _UpperCamelCase   :				List[str]       =     'cpu'  # ensure determinism for the device-dependent torch.Generator
        _UpperCamelCase   :				List[Any]       =     self.get_dummy_components()
        _UpperCamelCase   :				Tuple       =     StableDiffusionParadigmsPipeline(**lowerCamelCase     )
        _UpperCamelCase   :				str       =     sd_pipe.to(lowerCamelCase     )
        sd_pipe.set_progress_bar_config(disable=lowerCamelCase     )

        _UpperCamelCase   :				Any       =     self.get_dummy_inputs(lowerCamelCase     )
        _UpperCamelCase   :				str       =     sd_pipe(**lowerCamelCase     ).images
        _UpperCamelCase   :				str       =     image[0, -3:, -3:, -1]
        assert image.shape == (1, 6_4, 6_4, 3)

        _UpperCamelCase   :				Any       =     np.array([0.4_7_7_3, 0.5_4_1_7, 0.4_7_2_3, 0.4_9_2_5, 0.5_6_3_1, 0.4_7_5_2, 0.5_2_4_0, 0.4_9_3_5, 0.5_0_2_3]     )

        assert np.abs(image_slice.flatten() - expected_slice     ).max() < 1e-2

 def     __snake_case    (	self				:						List[Any]     )       ->							Union[str, Any]:
        _UpperCamelCase   :				Dict       =     'cpu'  # ensure determinism for the device-dependent torch.Generator
        _UpperCamelCase   :				str       =     self.get_dummy_components()
        torch.manual_seed(0     )
        _UpperCamelCase   :				Union[str, Any]       =     DDPMParallelScheduler()
        torch.manual_seed(0     )
        _UpperCamelCase   :				List[Any]       =     StableDiffusionParadigmsPipeline(**lowerCamelCase     )
        _UpperCamelCase   :				str       =     sd_pipe.to(lowerCamelCase     )
        sd_pipe.set_progress_bar_config(disable=lowerCamelCase     )

        _UpperCamelCase   :				int       =     self.get_dummy_inputs(lowerCamelCase     )
        _UpperCamelCase   :				Optional[int]       =     sd_pipe(**lowerCamelCase     ).images
        _UpperCamelCase   :				Dict       =     image[0, -3:, -3:, -1]
        assert image.shape == (1, 6_4, 6_4, 3)

        _UpperCamelCase   :				List[Any]       =     np.array([0.3_5_7_3, 0.4_4_2_0, 0.4_9_6_0, 0.4_7_9_9, 0.3_7_9_6, 0.3_8_7_9, 0.4_8_1_9, 0.4_3_6_5, 0.4_4_6_8]     )

        assert np.abs(image_slice.flatten() - expected_slice     ).max() < 1e-2

 def     __snake_case    (	self				:						Any     )       ->							List[Any]:
        super().test_inference_batch_consistent(batch_sizes=[1, 2]     )

 def     __snake_case    (	self				:						Union[str, Any]     )       ->							Tuple:
        super().test_inference_batch_single_identical(batch_size=2				,	expected_max_diff=3e-3     )

 def     __snake_case    (	self				:						List[str]     )       ->							List[str]:
        _UpperCamelCase   :				List[str]       =     'cpu'  # ensure determinism for the device-dependent torch.Generator
        _UpperCamelCase   :				Tuple       =     self.get_dummy_components()
        _UpperCamelCase   :				Any       =     StableDiffusionParadigmsPipeline(**lowerCamelCase     )
        _UpperCamelCase   :				List[Any]       =     sd_pipe.to(lowerCamelCase     )
        sd_pipe.set_progress_bar_config(disable=lowerCamelCase     )

        _UpperCamelCase   :				Tuple       =     self.get_dummy_inputs(lowerCamelCase     )
        _UpperCamelCase   :				int       =     'french fries'
        _UpperCamelCase   :				Optional[Any]       =     sd_pipe(**lowerCamelCase				,	negative_prompt=lowerCamelCase     )
        _UpperCamelCase   :				int       =     output.images
        _UpperCamelCase   :				str       =     image[0, -3:, -3:, -1]

        assert image.shape == (1, 6_4, 6_4, 3)

        _UpperCamelCase   :				Dict       =     np.array([0.4_7_7_1, 0.5_4_2_0, 0.4_6_8_3, 0.4_9_1_8, 0.5_6_3_6, 0.4_7_2_5, 0.5_2_3_0, 0.4_9_2_3, 0.5_0_1_5]     )

        assert np.abs(image_slice.flatten() - expected_slice     ).max() < 1e-2







@slow
@require_torch_gpu
class       __lowerCAmelCase     (				unittest.TestCase  ):



 '''simple docstring'''

 def     __snake_case    (	self				:						Optional[Any]     )       ->							Optional[int]:
        super().tearDown()
        gc.collect()
        torch.cuda.empty_cache()

 def     __snake_case    (	self				:						Dict				,	lowerCamelCase				:						Any=0     )       ->							Dict:
        _UpperCamelCase   :				List[str]       =     torch.Generator(device=lowerCamelCase     ).manual_seed(lowerCamelCase     )
        _UpperCamelCase   :				Tuple       =     {
            'prompt': 'a photograph of an astronaut riding a horse',
            'generator': generator,
            'num_inference_steps': 1_0,
            'guidance_scale': 7.5,
            'output_type': 'numpy',
            'parallel': 3,
            'debug': True,
        }
        return inputs

 def     __snake_case    (	self				:						List[Any]     )       ->							str:
        _UpperCamelCase   :				List[str]       =     'stabilityai/stable-diffusion-2-base'
        _UpperCamelCase   :				int       =     DDIMParallelScheduler.from_pretrained(lowerCamelCase				,	subfolder='scheduler'     )
        _UpperCamelCase   :				List[Any]       =     StableDiffusionParadigmsPipeline.from_pretrained(lowerCamelCase				,	scheduler=lowerCamelCase				,	safety_checker=lowerCamelCase     )
        pipe.to(lowerCamelCase     )
        pipe.set_progress_bar_config(disable=lowerCamelCase     )
        pipe.enable_attention_slicing()

        _UpperCamelCase   :				Optional[Any]       =     self.get_inputs()
        _UpperCamelCase   :				Optional[int]       =     pipe(**lowerCamelCase     ).images
        _UpperCamelCase   :				Optional[int]       =     image[0, -3:, -3:, -1].flatten()

        assert image.shape == (1, 5_1_2, 5_1_2, 3)

        _UpperCamelCase   :				int       =     np.array([0.9_6_2_2, 0.9_6_0_2, 0.9_7_4_8, 0.9_5_9_1, 0.9_6_3_0, 0.9_6_9_1, 0.9_6_6_1, 0.9_6_3_1, 0.9_7_4_1]     )

        assert np.abs(expected_slice - image_slice     ).max() < 1e-2



