


import inspect
import logging
import os
import random
import shutil
import tempfile
import unittest

import pytest
import torch
from torch import nn
from torch.utils.data import DataLoader, TensorDataset

from accelerate import Accelerator
from accelerate.test_utils import execute_subprocess_async, require_cuda
from accelerate.utils import ProjectConfiguration, set_seed


lowercase_	   =     logging.getLogger(__name__)
def 				UpperCamelCase__					(  SCREAMING_SNAKE_CASE__=2      ,	SCREAMING_SNAKE_CASE__=3      ,	SCREAMING_SNAKE_CASE__=16      ,	SCREAMING_SNAKE_CASE__ = 10      ,	SCREAMING_SNAKE_CASE__ = 2						):

 def get_dataset(SCREAMING_SNAKE_CASE__						):
  __lowerCamelCase		:							str		 =						torch.randn(batch_size * n_batches      ,	1						)
  return TensorDataset(SCREAMING_SNAKE_CASE__      ,	a * x + b + 0.1 * torch.randn(batch_size * n_batches      ,	1						)						)

 __lowerCamelCase		:							List[Any]		 =						get_dataset(SCREAMING_SNAKE_CASE__						)
 __lowerCamelCase		:							Any		 =						get_dataset(SCREAMING_SNAKE_CASE__						)
 __lowerCamelCase		:							int		 =						DataLoader(SCREAMING_SNAKE_CASE__      ,	shuffle=SCREAMING_SNAKE_CASE__      ,	batch_size=SCREAMING_SNAKE_CASE__      ,	num_workers=4						)
 __lowerCamelCase		:							Tuple		 =						DataLoader(SCREAMING_SNAKE_CASE__      ,	shuffle=SCREAMING_SNAKE_CASE__      ,	batch_size=SCREAMING_SNAKE_CASE__      ,	num_workers=4						)
 return (train_dataloader, valid_dataloader)
def 				UpperCamelCase__					(  SCREAMING_SNAKE_CASE__      ,	SCREAMING_SNAKE_CASE__      ,	SCREAMING_SNAKE_CASE__      ,	SCREAMING_SNAKE_CASE__      ,	SCREAMING_SNAKE_CASE__      ,	SCREAMING_SNAKE_CASE__=None						):
 __lowerCamelCase		:							str		 =						[]
 for epoch in range(SCREAMING_SNAKE_CASE__						):
  # Train quickly
  model.train()
  for batch in dataloader:
   __lowerCamelCase ,					__lowerCamelCase		:							Dict		 =						batch
   __lowerCamelCase		:							List[str]		 =						model(SCREAMING_SNAKE_CASE__						)
   __lowerCamelCase		:							Any		 =						torch.nn.functional.mse_loss(SCREAMING_SNAKE_CASE__      ,	SCREAMING_SNAKE_CASE__						)
   accelerator.backward(SCREAMING_SNAKE_CASE__						)
   optimizer.step()
   optimizer.zero_grad()
  rands.append(random.random()						)  # Introduce some randomness
  if scheduler is not None:
   scheduler.step()
 return rands


class 	A_ (	nn.Module   ):




      '''simple docstring'''


      def __init__(    self:	Union[str, Any]    ):
       super().__init__()
       __lowerCamelCase		:							Dict		 =						nn.Parameter(torch.randn(1    )    )
       __lowerCamelCase		:							Tuple		 =						nn.Parameter(torch.randn(1    )    )




      def 					_snake_case     (    self:	Tuple						,				a:	Union[str, Any]    ):
       return x * self.a + self.b



class 	A_ (	unittest.TestCase   ):




      '''simple docstring'''


      def 					_snake_case     (    self:	Optional[int]    ):
       with tempfile.TemporaryDirectory() as tmpdir:
        set_seed(42    )
        __lowerCamelCase		:							str		 =						DummyModel()
        __lowerCamelCase		:							Dict		 =						torch.optim.Adam(params=model.parameters()						,				lr=1e-3    )
        __lowerCamelCase ,					__lowerCamelCase		:							List[str]		 =						dummy_dataloaders()
        __lowerCamelCase		:							Dict		 =						ProjectConfiguration(total_limit=1						,				project_dir=a						,				automatic_checkpoint_naming=a    )
        # Train baseline
        __lowerCamelCase		:							Optional[Any]		 =						Accelerator(project_config=a    )
        __lowerCamelCase ,					__lowerCamelCase ,					__lowerCamelCase ,					__lowerCamelCase		:							str		 =						accelerator.prepare(
            a						,				a						,				a						,				a    )
        # Save initial
        accelerator.save_state()

        # Save second state
        accelerator.save_state()
        self.assertEqual(len(os.listdir(accelerator.project_dir    )    )						,				1    )




      def 					_snake_case     (    self:	List[str]    ):
       with tempfile.TemporaryDirectory() as tmpdir:
        set_seed(42    )
        __lowerCamelCase		:							Optional[int]		 =						DummyModel()
        __lowerCamelCase		:							Optional[int]		 =						torch.optim.Adam(params=model.parameters()						,				lr=1e-3    )
        __lowerCamelCase ,					__lowerCamelCase		:							List[str]		 =						dummy_dataloaders()
        # Train baseline
        __lowerCamelCase		:							int		 =						Accelerator()
        __lowerCamelCase ,					__lowerCamelCase ,					__lowerCamelCase ,					__lowerCamelCase		:							Any		 =						accelerator.prepare(
            a						,				a						,				a						,				a    )
        # Save initial
        __lowerCamelCase		:							Tuple		 =						os.path.join(a						,				'initial'    )
        accelerator.save_state(a    )
        ((__lowerCamelCase) ,					(__lowerCamelCase))		:							List[str]		 =						model.a.item(), model.b.item()
        __lowerCamelCase		:							Any		 =						optimizer.state_dict()
        __lowerCamelCase		:							str		 =						train(3						,				a						,				a						,				a						,				a    )
        ((__lowerCamelCase) ,					(__lowerCamelCase))		:							Optional[int]		 =						model.a.item(), model.b.item()
        __lowerCamelCase		:							str		 =						optimizer.state_dict()

        # Train partially
        set_seed(42    )
        __lowerCamelCase		:							Optional[int]		 =						DummyModel()
        __lowerCamelCase		:							str		 =						torch.optim.Adam(params=model.parameters()						,				lr=1e-3    )
        __lowerCamelCase ,					__lowerCamelCase		:							Dict		 =						dummy_dataloaders()
        __lowerCamelCase		:							Dict		 =						Accelerator()
        __lowerCamelCase ,					__lowerCamelCase ,					__lowerCamelCase ,					__lowerCamelCase		:							Any		 =						accelerator.prepare(
            a						,				a						,				a						,				a    )
        accelerator.load_state(a    )
        ((__lowerCamelCase) ,					(__lowerCamelCase))		:							Any		 =						model.a.item(), model.b.item()
        __lowerCamelCase		:							Union[str, Any]		 =						optimizer.state_dict()
        self.assertEqual(a						,				a    )
        self.assertEqual(a						,				a    )
        self.assertEqual(a						,				a    )

        __lowerCamelCase		:							int		 =						train(2						,				a						,				a						,				a						,				a    )
        # Save everything
        __lowerCamelCase		:							List[Any]		 =						os.path.join(a						,				'checkpoint'    )
        accelerator.save_state(a    )

        # Load everything back in and make sure all states work
        accelerator.load_state(a    )
        test_rands += train(1						,				a						,				a						,				a						,				a    )
        ((__lowerCamelCase) ,					(__lowerCamelCase))		:							Dict		 =						model.a.item(), model.b.item()
        __lowerCamelCase		:							List[str]		 =						optimizer.state_dict()
        self.assertEqual(a						,				a    )
        self.assertEqual(a						,				a    )
        self.assertEqual(a						,				a    )
        self.assertEqual(a						,				a    )




      def 					_snake_case     (    self:	Optional[Any]    ):
       with tempfile.TemporaryDirectory() as tmpdir:
        set_seed(42    )
        __lowerCamelCase		:							str		 =						DummyModel()
        __lowerCamelCase		:							Optional[Any]		 =						torch.optim.Adam(params=model.parameters()						,				lr=1e-3    )
        __lowerCamelCase ,					__lowerCamelCase		:							Dict		 =						dummy_dataloaders()
        __lowerCamelCase		:							Optional[Any]		 =						ProjectConfiguration(automatic_checkpoint_naming=a    )

        # Train baseline
        __lowerCamelCase		:							Any		 =						Accelerator(project_dir=a						,				project_config=a    )
        __lowerCamelCase ,					__lowerCamelCase ,					__lowerCamelCase ,					__lowerCamelCase		:							Union[str, Any]		 =						accelerator.prepare(
            a						,				a						,				a						,				a    )
        # Save initial
        accelerator.save_state()
        ((__lowerCamelCase) ,					(__lowerCamelCase))		:							Any		 =						model.a.item(), model.b.item()
        __lowerCamelCase		:							List[Any]		 =						optimizer.state_dict()
        __lowerCamelCase		:							Tuple		 =						train(3						,				a						,				a						,				a						,				a    )
        ((__lowerCamelCase) ,					(__lowerCamelCase))		:							str		 =						model.a.item(), model.b.item()
        __lowerCamelCase		:							int		 =						optimizer.state_dict()

        # Train partially
        set_seed(42    )
        __lowerCamelCase		:							Optional[Any]		 =						DummyModel()
        __lowerCamelCase		:							int		 =						torch.optim.Adam(params=model.parameters()						,				lr=1e-3    )
        __lowerCamelCase ,					__lowerCamelCase		:							List[str]		 =						dummy_dataloaders()
        __lowerCamelCase		:							Tuple		 =						ProjectConfiguration(iteration=1						,				automatic_checkpoint_naming=a    )
        __lowerCamelCase		:							str		 =						Accelerator(project_dir=a						,				project_config=a    )
        __lowerCamelCase ,					__lowerCamelCase ,					__lowerCamelCase ,					__lowerCamelCase		:							Dict		 =						accelerator.prepare(
            a						,				a						,				a						,				a    )
        accelerator.load_state(os.path.join(a						,				'checkpoints'						,				'checkpoint_0'    )    )
        ((__lowerCamelCase) ,					(__lowerCamelCase))		:							List[Any]		 =						model.a.item(), model.b.item()
        __lowerCamelCase		:							Tuple		 =						optimizer.state_dict()
        self.assertEqual(a						,				a    )
        self.assertEqual(a						,				a    )
        self.assertEqual(a						,				a    )

        __lowerCamelCase		:							List[Any]		 =						train(2						,				a						,				a						,				a						,				a    )
        # Save everything
        accelerator.save_state()

        # Load everything back in and make sure all states work
        accelerator.load_state(os.path.join(a						,				'checkpoints'						,				'checkpoint_1'    )    )
        test_rands += train(1						,				a						,				a						,				a						,				a    )
        ((__lowerCamelCase) ,					(__lowerCamelCase))		:							Any		 =						model.a.item(), model.b.item()
        __lowerCamelCase		:							Any		 =						optimizer.state_dict()
        self.assertEqual(a						,				a    )
        self.assertEqual(a						,				a    )
        self.assertEqual(a						,				a    )
        self.assertEqual(a						,				a    )




      def 					_snake_case     (    self:	Optional[int]    ):
       __lowerCamelCase		:							int		 =						torch.tensor([1, 2, 3]    )
       __lowerCamelCase		:							Optional[int]		 =						torch.tensor([2, 3, 4]    )
       __lowerCamelCase		:							Any		 =						DummyModel()
       __lowerCamelCase		:							Tuple		 =						torch.optim.Adam(net.parameters()    )
       __lowerCamelCase		:							List[Any]		 =						Accelerator()
       with self.assertRaises(a    ) as ve:
        accelerator.register_for_checkpointing(a						,				a						,				a						,				a    )
       __lowerCamelCase		:							List[str]		 =						str(ve.exception    )
       self.assertTrue('Item at index 0' in message    )
       self.assertTrue('Item at index 1' in message    )
       self.assertFalse('Item at index 2' in message    )
       self.assertFalse('Item at index 3' in message    )




      def 					_snake_case     (    self:	Tuple    ):
       with tempfile.TemporaryDirectory() as tmpdir:
        set_seed(42    )
        __lowerCamelCase		:							Tuple		 =						DummyModel()
        __lowerCamelCase		:							List[Any]		 =						torch.optim.Adam(params=model.parameters()						,				lr=1e-3    )
        __lowerCamelCase		:							str		 =						torch.optim.lr_scheduler.StepLR(a						,				step_size=1						,				gamma=0.9_9    )
        __lowerCamelCase ,					__lowerCamelCase		:							Dict		 =						dummy_dataloaders()
        __lowerCamelCase		:							Optional[Any]		 =						ProjectConfiguration(automatic_checkpoint_naming=a    )
        # Train baseline
        __lowerCamelCase		:							Dict		 =						Accelerator(project_dir=a						,				project_config=a    )
        __lowerCamelCase ,					__lowerCamelCase ,					__lowerCamelCase ,					__lowerCamelCase ,					__lowerCamelCase		:							Optional[int]		 =						accelerator.prepare(
            a						,				a						,				a						,				a						,				a    )
        # Save initial
        accelerator.save_state()
        __lowerCamelCase		:							List[str]		 =						scheduler.state_dict()
        train(3						,				a						,				a						,				a						,				a						,				a    )
        self.assertNotEqual(a						,				scheduler.state_dict()    )

        # Load everything back in and make sure all states work
        accelerator.load_state(os.path.join(a						,				'checkpoints'						,				'checkpoint_0'    )    )
        self.assertEqual(a						,				scheduler.state_dict()    )




      def 					_snake_case     (    self:	List[str]    ):
       with tempfile.TemporaryDirectory() as tmpdir:
        set_seed(42    )
        __lowerCamelCase		:							Union[str, Any]		 =						DummyModel()
        __lowerCamelCase		:							List[str]		 =						ProjectConfiguration(automatic_checkpoint_naming=a						,				total_limit=2    )
        # Train baseline
        __lowerCamelCase		:							Tuple		 =						Accelerator(project_dir=a						,				project_config=a    )
        __lowerCamelCase		:							str		 =						accelerator.prepare(a    )
        # Save 3 states:
        for _ in range(11    ):
         accelerator.save_state()
        self.assertTrue(not os.path.exists(os.path.join(a						,				'checkpoints'						,				'checkpoint_0'    )    )    )
        self.assertTrue(os.path.exists(os.path.join(a						,				'checkpoints'						,				'checkpoint_9'    )    )    )
        self.assertTrue(os.path.exists(os.path.join(a						,				'checkpoints'						,				'checkpoint_10'    )    )    )




      @require_cuda
      def 					_snake_case     (    self:	Optional[int]    ):
       __lowerCamelCase		:							List[Any]		 =						['torchrun', F'--nproc_per_node={torch.cuda.device_count()}', inspect.getfile(self.__class__    )]
       execute_subprocess_async(a						,				env=os.environ.copy()    )


if __name__ == "__main__":
       lowercase_	   =     '/tmp/accelerate/state_checkpointing'
       lowercase_	   =     DummyModel()
       lowercase_	   =     torch.optim.Adam(params=model.parameters(), lr=1E-3)
       lowercase_	   =     torch.optim.lr_scheduler.StepLR(optimizer, step_size=1, gamma=0.99)
       lowercase_   ,lowercase_	   =     dummy_dataloaders()
       lowercase_	   =     ProjectConfiguration(automatic_checkpoint_naming=True)
       # Train baseline
       lowercase_	   =     Accelerator(project_dir=savedir, project_config=project_config, mixed_precision='no')
       if accelerator.process_index == 0:
              if os.path.exists(savedir):
                     shutil.rmtree(savedir)
              os.makedirs(savedir)
       lowercase_   ,lowercase_   ,lowercase_   ,lowercase_   ,lowercase_	   =     accelerator.prepare(
           model, optimizer, train_dataloader, valid_dataloader, scheduler
       )
       lowercase_   ,lowercase_	   =     accelerator.prepare(model, optimizer)
       train(3, model, train_dataloader, optimizer, accelerator, scheduler)
       # Check that the intial optimizer is loaded on the GPU
       for group in optimizer.param_groups:
              lowercase_	   =     group['params'][0].device
              break
       assert param_device.type == accelerator.device.type
       lowercase_	   =     model.cpu()
       accelerator.wait_for_everyone()
       accelerator.save_state()
       accelerator.wait_for_everyone()

       # Check CPU state
       accelerator.load_state(os.path.join(savedir, 'checkpoints', 'checkpoint_0'), map_location='cpu')
       for group in optimizer.param_groups:
              lowercase_	   =     group['params'][0].device
              break
       assert (
           param_device.type == torch.device('cpu').type
       ), F"Loaded optimizer states did not match, expected to be loaded on the CPU but got {param_device}"

       # Check device state
       model.to(accelerator.device)
       accelerator.load_state(os.path.join(savedir, 'checkpoints', 'checkpoint_0'), map_location='on_device')
       for group in optimizer.param_groups:
              lowercase_	   =     group['params'][0].device
              break
       assert (
           param_device.type == accelerator.device.type
       ), F"Loaded optimizer states did not match, expected to be loaded on {accelerator.device} but got {param_device}"

       # Check error
       with pytest.raises(TypeError, match='Unsupported optimizer map location passed'):
              accelerator.load_state(os.path.join(savedir, 'checkpoints', 'checkpoint_0'), map_location='invalid')
       accelerator.wait_for_everyone()
       if accelerator.process_index == 0:
              shutil.rmtree(savedir)
       accelerator.wait_for_everyone()





import inspect
from typing import Optional, Union

import numpy as np
import PIL
import torch
from torch.nn import functional as F
from torchvision import transforms
from transformers import CLIPFeatureExtractor, CLIPModel, CLIPTextModel, CLIPTokenizer

from diffusers import (
    AutoencoderKL,
    DDIMScheduler,
    DiffusionPipeline,
    DPMSolverMultistepScheduler,
    LMSDiscreteScheduler,
    PNDMScheduler,
    UNetaDConditionModel,
)
from diffusers.pipelines.stable_diffusion.pipeline_stable_diffusion import StableDiffusionPipelineOutput
from diffusers.utils import (
    PIL_INTERPOLATION,
    randn_tensor,
)
def 				UpperCamelCase__					(  SCREAMING_SNAKE_CASE__      ,	SCREAMING_SNAKE_CASE__      ,	SCREAMING_SNAKE_CASE__						):
 if isinstance(SCREAMING_SNAKE_CASE__      ,	torch.Tensor						):
  return image
 elif isinstance(SCREAMING_SNAKE_CASE__      ,	PIL.Image.Image						):
  __lowerCamelCase		:							Union[str, Any]		 =						[image]

 if isinstance(image[0]      ,	PIL.Image.Image						):
  __lowerCamelCase		:							Any		 =						[np.array(i.resize((w, h)      ,	resample=PIL_INTERPOLATION['lanczos']						)						)[None, :] for i in image]
  __lowerCamelCase		:							Optional[int]		 =						np.concatenate(SCREAMING_SNAKE_CASE__      ,	axis=0						)
  __lowerCamelCase		:							str		 =						np.array(SCREAMING_SNAKE_CASE__						).astype(np.floataa						) / 255.0
  __lowerCamelCase		:							List[str]		 =						image.transpose(0      ,	3      ,	1      ,	2						)
  __lowerCamelCase		:							Union[str, Any]		 =						2.0 * image - 1.0
  __lowerCamelCase		:							Tuple		 =						torch.from_numpy(SCREAMING_SNAKE_CASE__						)
 elif isinstance(image[0]      ,	torch.Tensor						):
  __lowerCamelCase		:							str		 =						torch.cat(SCREAMING_SNAKE_CASE__      ,	dim=0						)
 return image
def 				UpperCamelCase__					(  SCREAMING_SNAKE_CASE__      ,	SCREAMING_SNAKE_CASE__      ,	SCREAMING_SNAKE_CASE__      ,	SCREAMING_SNAKE_CASE__=0.9_995						):
 if not isinstance(SCREAMING_SNAKE_CASE__      ,	np.ndarray						):
  __lowerCamelCase		:							List[str]		 =						True
  __lowerCamelCase		:							str		 =						va.device
  __lowerCamelCase		:							int		 =						va.cpu().numpy()
  __lowerCamelCase		:							List[str]		 =						va.cpu().numpy()

 __lowerCamelCase		:							str		 =						np.sum(va * va / (np.linalg.norm(SCREAMING_SNAKE_CASE__						) * np.linalg.norm(SCREAMING_SNAKE_CASE__						))						)
 if np.abs(SCREAMING_SNAKE_CASE__						) > DOT_THRESHOLD:
  __lowerCamelCase		:							Union[str, Any]		 =						(1 - t) * va + t * va
 else:
  __lowerCamelCase		:							List[Any]		 =						np.arccos(SCREAMING_SNAKE_CASE__						)
  __lowerCamelCase		:							Dict		 =						np.sin(SCREAMING_SNAKE_CASE__						)
  __lowerCamelCase		:							str		 =						theta_a * t
  __lowerCamelCase		:							List[Any]		 =						np.sin(SCREAMING_SNAKE_CASE__						)
  __lowerCamelCase		:							str		 =						np.sin(theta_a - theta_t						) / sin_theta_a
  __lowerCamelCase		:							List[Any]		 =						sin_theta_t / sin_theta_a
  __lowerCamelCase		:							Union[str, Any]		 =						sa * va + sa * va

 if inputs_are_torch:
  __lowerCamelCase		:							str		 =						torch.from_numpy(SCREAMING_SNAKE_CASE__						).to(SCREAMING_SNAKE_CASE__						)

 return va
def 				UpperCamelCase__					(  SCREAMING_SNAKE_CASE__      ,	SCREAMING_SNAKE_CASE__						):
 __lowerCamelCase		:							List[Any]		 =						F.normalize(SCREAMING_SNAKE_CASE__      ,	dim=-1						)
 __lowerCamelCase		:							Union[str, Any]		 =						F.normalize(SCREAMING_SNAKE_CASE__      ,	dim=-1						)
 return (x - y).norm(dim=-1						).div(2						).arcsin().pow(2						).mul(2						)
def 				UpperCamelCase__					(  SCREAMING_SNAKE_CASE__      ,	SCREAMING_SNAKE_CASE__						):
 for param in model.parameters():
  __lowerCamelCase		:							Any		 =						value



class 	A_ (	__UpperCamelCase   ):




      '''simple docstring'''


      def __init__(    self:	Any						,				a:	AutoencoderKL						,				a:	CLIPTextModel						,				a:	CLIPModel						,				a:	CLIPTokenizer						,				a:	UNetaDConditionModel						,				a:	Union[PNDMScheduler, LMSDiscreteScheduler, DDIMScheduler, DPMSolverMultistepScheduler]						,				a:	CLIPFeatureExtractor						,				a:	Union[str, Any]=None						,				a:	Union[str, Any]=None						,				a:	Union[str, Any]=None						,				):
       super().__init__()
       self.register_modules(
           vae=a						,				text_encoder=a						,				clip_model=a						,				tokenizer=a						,				unet=a						,				scheduler=a						,				feature_extractor=a						,				coca_model=a						,				coca_tokenizer=a						,				coca_transform=a						,				)
       __lowerCamelCase		:							Tuple		 =						(
           feature_extractor.size
           if isinstance(feature_extractor.size						,				a    )
           else feature_extractor.size['shortest_edge']
       )
       __lowerCamelCase		:							List[Any]		 =						transforms.Normalize(mean=feature_extractor.image_mean						,				std=feature_extractor.image_std    )
       set_requires_grad(self.text_encoder						,				a    )
       set_requires_grad(self.clip_model						,				a    )




      def 					_snake_case     (    self:	Optional[Any]						,				a:	Optional[Union[str, int]] = "auto"    ):
       if slice_size == "auto":
        # half the attention head size is usually a good trade-off between
        # speed and memory
        __lowerCamelCase		:							Any		 =						self.unet.config.attention_head_dim // 2
       self.unet.set_attention_slice(a    )




      def 					_snake_case     (    self:	Dict    ):
       self.enable_attention_slicing(a    )




      def 					_snake_case     (    self:	Optional[Any]    ):
       set_requires_grad(self.vae						,				a    )




      def 					_snake_case     (    self:	List[Any]    ):
       set_requires_grad(self.vae						,				a    )




      def 					_snake_case     (    self:	int    ):
       set_requires_grad(self.unet						,				a    )




      def 					_snake_case     (    self:	int    ):
       set_requires_grad(self.unet						,				a    )




      def 					_snake_case     (    self:	Optional[Any]						,				a:	Union[str, Any]						,				a:	List[str]						,				a:	List[Any]    ):
       # get the original timestep using init_timestep
       __lowerCamelCase		:							List[Any]		 =						min(int(num_inference_steps * strength    )						,				a    )

       __lowerCamelCase		:							str		 =						max(num_inference_steps - init_timestep						,				0    )
       __lowerCamelCase		:							List[Any]		 =						self.scheduler.timesteps[t_start:]

       return timesteps, num_inference_steps - t_start




      def 					_snake_case     (    self:	Union[str, Any]						,				a:	Optional[Any]						,				a:	Any						,				a:	Optional[int]						,				a:	Optional[Any]						,				a:	Union[str, Any]						,				a:	List[str]=None    ):
       if not isinstance(a						,				torch.Tensor    ):
        raise ValueError(F'`image` has to be of type `torch.Tensor` but is {type(a    )}'    )

       __lowerCamelCase		:							Union[str, Any]		 =						image.to(device=a						,				dtype=a    )

       if isinstance(a						,				a    ):
        __lowerCamelCase		:							str		 =						[
            self.vae.encode(image[i : i + 1]    ).latent_dist.sample(generator[i]    ) for i in range(a    )
        ]
        __lowerCamelCase		:							Tuple		 =						torch.cat(a						,				dim=0    )
       else:
        __lowerCamelCase		:							List[Any]		 =						self.vae.encode(a    ).latent_dist.sample(a    )

       # Hardcode 0.18215 because stable-diffusion-2-base has not self.vae.config.scaling_factor
       __lowerCamelCase		:							List[str]		 =						0.1_8_2_1_5 * init_latents
       __lowerCamelCase		:							Union[str, Any]		 =						init_latents.repeat_interleave(a						,				dim=0    )

       __lowerCamelCase		:							Optional[int]		 =						randn_tensor(init_latents.shape						,				generator=a						,				device=a						,				dtype=a    )

       # get latents
       __lowerCamelCase		:							Union[str, Any]		 =						self.scheduler.add_noise(a						,				a						,				a    )
       __lowerCamelCase		:							int		 =						init_latents

       return latents




      def 					_snake_case     (    self:	Optional[int]						,				a:	Any    ):
       __lowerCamelCase		:							List[Any]		 =						self.coca_transform(a    ).unsqueeze(0    )
       with torch.no_grad(), torch.cuda.amp.autocast():
        __lowerCamelCase		:							Any		 =						self.coca_model.generate(transformed_image.to(device=self.device						,				dtype=self.coca_model.dtype    )    )
       __lowerCamelCase		:							str		 =						self.coca_tokenizer.decode(generated[0].cpu().numpy()    )
       return generated.split('<end_of_text>'    )[0].replace('<start_of_text>'						,				''    ).rstrip(' .,'    )




      def 					_snake_case     (    self:	Any						,				a:	Tuple						,				a:	Tuple    ):
       __lowerCamelCase		:							Dict		 =						self.feature_extractor.preprocess(a    )
       __lowerCamelCase		:							Dict		 =						torch.from_numpy(clip_image_input['pixel_values'][0]    ).unsqueeze(0    ).to(self.device    ).half()
       __lowerCamelCase		:							List[str]		 =						self.clip_model.get_image_features(a    )
       __lowerCamelCase		:							Optional[Any]		 =						image_embeddings_clip / image_embeddings_clip.norm(p=2						,				dim=-1						,				keepdim=a    )
       __lowerCamelCase		:							Tuple		 =						image_embeddings_clip.repeat_interleave(a						,				dim=0    )
       return image_embeddings_clip




      @torch.enable_grad()
      def 					_snake_case     (    self:	str						,				a:	str						,				a:	int						,				a:	List[Any]						,				a:	str						,				a:	List[Any]						,				a:	Dict						,				a:	int						,				):
       __lowerCamelCase		:							Optional[Any]		 =						latents.detach().requires_grad_()

       __lowerCamelCase		:							str		 =						self.scheduler.scale_model_input(a						,				a    )

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

       if isinstance(self.scheduler						,				(PNDMScheduler, DDIMScheduler, DPMSolverMultistepScheduler)    ):
        __lowerCamelCase		:							str		 =						self.scheduler.alphas_cumprod[timestep]
        __lowerCamelCase		:							Dict		 =						1 - alpha_prod_t
        # compute predicted original sample from predicted noise also called
        # "predicted x_0" of formula (12) from https://arxiv.org/pdf/2010.02502.pdf
        __lowerCamelCase		:							Optional[int]		 =						(latents - beta_prod_t ** 0.5 * noise_pred) / alpha_prod_t ** 0.5

        __lowerCamelCase		:							Optional[int]		 =						torch.sqrt(a    )
        __lowerCamelCase		:							int		 =						pred_original_sample * (fac) + latents * (1 - fac)
       elif isinstance(self.scheduler						,				a    ):
        __lowerCamelCase		:							str		 =						self.scheduler.sigmas[index]
        __lowerCamelCase		:							List[Any]		 =						latents - sigma * noise_pred
       else:
        raise ValueError(F'scheduler type {type(self.scheduler    )} not supported'    )

       # Hardcode 0.18215 because stable-diffusion-2-base has not self.vae.config.scaling_factor
       __lowerCamelCase		:							Optional[int]		 =						1 / 0.1_8_2_1_5 * sample
       __lowerCamelCase		:							Optional[Any]		 =						self.vae.decode(a    ).sample
       __lowerCamelCase		:							Tuple		 =						(image / 2 + 0.5).clamp(0						,				1    )

       __lowerCamelCase		:							Any		 =						transforms.Resize(self.feature_extractor_size    )(a    )
       __lowerCamelCase		:							Union[str, Any]		 =						self.normalize(a    ).to(latents.dtype    )

       __lowerCamelCase		:							Tuple		 =						self.clip_model.get_image_features(a    )
       __lowerCamelCase		:							List[str]		 =						image_embeddings_clip / image_embeddings_clip.norm(p=2						,				dim=-1						,				keepdim=a    )

       __lowerCamelCase		:							List[str]		 =						spherical_dist_loss(a						,				a    ).mean() * clip_guidance_scale

       __lowerCamelCase		:							Tuple		 =						-torch.autograd.grad(a						,				a    )[0]

       if isinstance(self.scheduler						,				a    ):
        __lowerCamelCase		:							Optional[int]		 =						latents.detach() + grads * (sigma**2)
        __lowerCamelCase		:							List[Any]		 =						noise_pred_original
       else:
        __lowerCamelCase		:							str		 =						noise_pred_original - torch.sqrt(a    ) * grads
       return noise_pred, latents




      @torch.no_grad()
      def __call__(    self:	Any						,				a:	Union[torch.FloatTensor, PIL.Image.Image]						,				a:	Union[torch.FloatTensor, PIL.Image.Image]						,				a:	Optional[str] = None						,				a:	Optional[str] = None						,				a:	Optional[int] = 512						,				a:	Optional[int] = 512						,				a:	float = 0.6						,				a:	Optional[int] = 50						,				a:	Optional[float] = 7.5						,				a:	Optional[int] = 1						,				a:	float = 0.0						,				a:	Optional[float] = 100						,				a:	Optional[torch.Generator] = None						,				a:	Optional[str] = "pil"						,				a:	bool = True						,				a:	float = 0.8						,				a:	float = 0.1						,				a:	float = 0.1						,				):
       if isinstance(a						,				a    ) and len(a    ) != batch_size:
        raise ValueError(F'You have passed {batch_size} batch_size, but only {len(a    )} generators.'    )

       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 isinstance(a						,				torch.Generator    ) and batch_size > 1:
        __lowerCamelCase		:							List[Any]		 =						[generator] + [None] * (batch_size - 1)

       __lowerCamelCase		:							Dict		 =						[
           ('model', self.coca_model is None),
           ('tokenizer', self.coca_tokenizer is None),
           ('transform', self.coca_transform is None),
       ]
       __lowerCamelCase		:							Any		 =						[x[0] for x in coca_is_none if x[1]]
       __lowerCamelCase		:							str		 =						', '.join(a    )
       # generate prompts with coca model if prompt is None
       if content_prompt is None:
        if len(a    ):
         raise ValueError(
             F'Content prompt is None and CoCa [{coca_is_none_str}] is None.'
             F'Set prompt or pass Coca [{coca_is_none_str}] to DiffusionPipeline.'    )
        __lowerCamelCase		:							Any		 =						self.get_image_description(a    )
       if style_prompt is None:
        if len(a    ):
         raise ValueError(
             F'Style prompt is None and CoCa [{coca_is_none_str}] is None.'
             F' Set prompt or pass Coca [{coca_is_none_str}] to DiffusionPipeline.'    )
        __lowerCamelCase		:							Tuple		 =						self.get_image_description(a    )

       # get prompt text embeddings for content and style
       __lowerCamelCase		:							int		 =						self.tokenizer(
           a						,				padding='max_length'						,				max_length=self.tokenizer.model_max_length						,				truncation=a						,				return_tensors='pt'						,				)
       __lowerCamelCase		:							Dict		 =						self.text_encoder(content_text_input.input_ids.to(self.device    )    )[0]

       __lowerCamelCase		:							Union[str, Any]		 =						self.tokenizer(
           a						,				padding='max_length'						,				max_length=self.tokenizer.model_max_length						,				truncation=a						,				return_tensors='pt'						,				)
       __lowerCamelCase		:							Any		 =						self.text_encoder(style_text_input.input_ids.to(self.device    )    )[0]

       __lowerCamelCase		:							List[Any]		 =						slerp(a						,				a						,				a    )

       # duplicate text embeddings for each generation per prompt
       __lowerCamelCase		:							Any		 =						text_embeddings.repeat_interleave(a						,				dim=0    )

       # set timesteps
       __lowerCamelCase		:							List[Any]		 =						'offset' in set(inspect.signature(self.scheduler.set_timesteps    ).parameters.keys()    )
       __lowerCamelCase		:							Union[str, Any]		 =						{}
       if accepts_offset:
        __lowerCamelCase		:							Dict		 =						1

       self.scheduler.set_timesteps(a						,				**a    )
       # Some schedulers like PNDM have timesteps as arrays
       # It's more optimized to move all timesteps to correct device beforehand
       self.scheduler.timesteps.to(self.device    )

       __lowerCamelCase ,					__lowerCamelCase		:							Dict		 =						self.get_timesteps(a						,				a						,				self.device    )
       __lowerCamelCase		:							Tuple		 =						timesteps[:1].repeat(a    )

       # Preprocess image
       __lowerCamelCase		:							Any		 =						preprocess(a						,				a						,				a    )
       __lowerCamelCase		:							str		 =						self.prepare_latents(
           a						,				a						,				a						,				text_embeddings.dtype						,				self.device						,				a    )

       __lowerCamelCase		:							Dict		 =						preprocess(a						,				a						,				a    )
       __lowerCamelCase		:							Optional[int]		 =						self.prepare_latents(
           a						,				a						,				a						,				text_embeddings.dtype						,				self.device						,				a    )

       __lowerCamelCase		:							int		 =						slerp(a						,				a						,				a    )

       if clip_guidance_scale > 0:
        __lowerCamelCase		:							List[str]		 =						self.get_clip_image_embeddings(a						,				a    )
        __lowerCamelCase		:							Union[str, Any]		 =						self.get_clip_image_embeddings(a						,				a    )
        __lowerCamelCase		:							Union[str, Any]		 =						slerp(
            a						,				a						,				a    )

       # 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.
       __lowerCamelCase		:							Tuple		 =						guidance_scale > 1.0
       # get unconditional embeddings for classifier free guidance
       if do_classifier_free_guidance:
        __lowerCamelCase		:							Optional[int]		 =						content_text_input.input_ids.shape[-1]
        __lowerCamelCase		:							int		 =						self.tokenizer(['']						,				padding='max_length'						,				max_length=a						,				return_tensors='pt'    )
        __lowerCamelCase		:							List[Any]		 =						self.text_encoder(uncond_input.input_ids.to(self.device    )    )[0]
        # duplicate unconditional embeddings for each generation per prompt
        __lowerCamelCase		:							List[Any]		 =						uncond_embeddings.repeat_interleave(a						,				dim=0    )

        # 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
        __lowerCamelCase		:							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`.
       __lowerCamelCase		:							str		 =						(batch_size, self.unet.config.in_channels, height // 8, width // 8)
       __lowerCamelCase		:							List[str]		 =						text_embeddings.dtype
       if latents is None:
        if self.device.type == "mps":
         # randn does not work reproducibly on mps
         __lowerCamelCase		:							Tuple		 =						torch.randn(a						,				generator=a						,				device='cpu'						,				dtype=a    ).to(
             self.device    )
        else:
         __lowerCamelCase		:							List[Any]		 =						torch.randn(a						,				generator=a						,				device=self.device						,				dtype=a    )
       else:
        if latents.shape != latents_shape:
         raise ValueError(F'Unexpected latents shape, got {latents.shape}, expected {latents_shape}'    )
        __lowerCamelCase		:							List[str]		 =						latents.to(self.device    )

       # scale the initial noise by the standard deviation required by the scheduler
       __lowerCamelCase		:							str		 =						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]
       __lowerCamelCase		:							int		 =						'eta' in set(inspect.signature(self.scheduler.step    ).parameters.keys()    )
       __lowerCamelCase		:							Dict		 =						{}
       if accepts_eta:
        __lowerCamelCase		:							List[str]		 =						eta

       # check if the scheduler accepts generator
       __lowerCamelCase		:							Optional[int]		 =						'generator' in set(inspect.signature(self.scheduler.step    ).parameters.keys()    )
       if accepts_generator:
        __lowerCamelCase		:							Optional[Any]		 =						generator

       with self.progress_bar(total=a    ):
        for i, t in enumerate(a    ):
         # expand the latents if we are doing classifier free guidance
         __lowerCamelCase		:							Optional[int]		 =						torch.cat([latents] * 2    ) if do_classifier_free_guidance else latents
         __lowerCamelCase		:							Union[str, Any]		 =						self.scheduler.scale_model_input(a						,				a    )

         # predict the noise residual
         __lowerCamelCase		:							Tuple		 =						self.unet(a						,				a						,				encoder_hidden_states=a    ).sample

         # perform classifier free guidance
         if do_classifier_free_guidance:
          __lowerCamelCase ,					__lowerCamelCase		:							str		 =						noise_pred.chunk(2    )
          __lowerCamelCase		:							List[str]		 =						noise_pred_uncond + guidance_scale * (noise_pred_text - noise_pred_uncond)

         # perform clip guidance
         if clip_guidance_scale > 0:
          __lowerCamelCase		:							str		 =						(
              text_embeddings.chunk(2    )[1] if do_classifier_free_guidance else text_embeddings
          )
          __lowerCamelCase ,					__lowerCamelCase		:							int		 =						self.cond_fn(
              a						,				a						,				a						,				a						,				a						,				a						,				a						,				)

         # compute the previous noisy sample x_t -> x_t-1
         __lowerCamelCase		:							Tuple		 =						self.scheduler.step(a						,				a						,				a						,				**a    ).prev_sample

        # Hardcode 0.18215 because stable-diffusion-2-base has not self.vae.config.scaling_factor
       __lowerCamelCase		:							List[Any]		 =						1 / 0.1_8_2_1_5 * latents
       __lowerCamelCase		:							Union[str, Any]		 =						self.vae.decode(a    ).sample

       __lowerCamelCase		:							Optional[int]		 =						(image / 2 + 0.5).clamp(0						,				1    )
       __lowerCamelCase		:							Any		 =						image.cpu().permute(0						,				2						,				3						,				1    ).numpy()

       if output_type == "pil":
        __lowerCamelCase		:							Union[str, Any]		 =						self.numpy_to_pil(a    )

       if not return_dict:
        return (image, None)

       return StableDiffusionPipelineOutput(images=a						,				nsfw_content_detected=a    )

