


'''simple docstring'''






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

import numpy as np
import PIL
import torch
import torch.utils.checkpoint

from ...models import UNetaDModel, VQModel
from ...schedulers import (
    DDIMScheduler,
    DPMSolverMultistepScheduler,
    EulerAncestralDiscreteScheduler,
    EulerDiscreteScheduler,
    LMSDiscreteScheduler,
    PNDMScheduler,
)
from ...utils import PIL_INTERPOLATION, randn_tensor
from ..pipeline_utils import DiffusionPipeline, ImagePipelineOutput






def 		_A			(lowerCAmelCase__					:int						)				->					str:




     '''simple docstring'''



     _a ,				_a      =							image.size
     _a ,				_a      =							(x - x % 32 for x in (w, h))  # resize to integer multiple of 32
     _a      =							image.resize((w, h)			,  resample=PIL_INTERPOLATION['lanczos']						)
     _a      =							np.array(lowerCAmelCase__						).astype(np.floataa						) / 2_55.0
     _a      =							image[None].transpose(0			,  3			,  1			,  2						)
     _a      =							torch.from_numpy(lowerCAmelCase__						)
     return 2.0 * image - 1.0






class 	a       (       _SCREAMING_SNAKE_CASE    ):
 def __init__(	self ,			__magic_name__ ,			__magic_name__ ,			__magic_name__ ,			)   ->							List[Any]:
      super().__init__()
      self.register_modules(vqvae=__magic_name__ ,			unet=__magic_name__ ,			scheduler=__magic_name__  )




 @torch.no_grad()
 def __call__(	self ,			__magic_name__ = None ,			__magic_name__ = 1 ,			__magic_name__ = 1_00 ,			__magic_name__ = 0.0 ,			__magic_name__ = None ,			__magic_name__ = "pil" ,			__magic_name__ = True ,			)   ->							Union[Tuple, ImagePipelineOutput]:
      if isinstance(__magic_name__ ,			PIL.Image.Image  ):
           _a      =							1
      elif isinstance(__magic_name__ ,			torch.Tensor  ):
           _a      =							image.shape[0]
      else:
           raise ValueError(f'`image` has to be of type `PIL.Image.Image` or `torch.Tensor` but is {type(__magic_name__  )}'  )

      if isinstance(__magic_name__ ,			PIL.Image.Image  ):
           _a      =							preprocess(__magic_name__  )

      _a ,				_a      =							image.shape[-2:]

      # in_channels should be 6: 3 for latents, 3 for low resolution image
      _a      =							(batch_size, self.unet.config.in_channels // 2, height, width)
      _a      =							next(self.unet.parameters()  ).dtype

      _a      =							randn_tensor(__magic_name__ ,			generator=__magic_name__ ,			device=self.device ,			dtype=__magic_name__  )

      _a      =							image.to(device=self.device ,			dtype=__magic_name__  )

      # set timesteps and move to the correct device
      self.scheduler.set_timesteps(__magic_name__ ,			device=self.device  )
      _a      =							self.scheduler.timesteps

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

      for t in self.progress_bar(__magic_name__  ):
           # concat latents and low resolution image in the channel dimension.
           _a      =							torch.cat([latents, image] ,			dim=1  )
           _a      =							self.scheduler.scale_model_input(__magic_name__ ,			__magic_name__  )
           # predict the noise residual
           _a      =							self.unet(__magic_name__ ,			__magic_name__  ).sample
           # compute the previous noisy sample x_t -> x_t-1
           _a      =							self.scheduler.step(__magic_name__ ,			__magic_name__ ,			__magic_name__ ,			**__magic_name__  ).prev_sample

      # decode the image latents with the VQVAE
      _a      =							self.vqvae.decode(__magic_name__  ).sample
      _a      =							torch.clamp(__magic_name__ ,			-1.0 ,			1.0  )
      _a      =							image / 2 + 0.5
      _a      =							image.cpu().permute(0 ,			2 ,			3 ,			1  ).numpy()

      if output_type == "pil":
           _a      =							self.numpy_to_pil(__magic_name__  )

      if not return_dict:
           return (image,)

      return ImagePipelineOutput(images=__magic_name__  )





'''simple docstring'''






import unittest

from diffusers.pipelines.pipeline_utils import is_safetensors_compatible




class 	a       (       unittest.TestCase    ):
 def 							__UpperCAmelCase						(	self  )   ->							int:
      _a      =							[
          'safety_checker/pytorch_model.bin',
          'safety_checker/model.safetensors',
          'vae/diffusion_pytorch_model.bin',
          'vae/diffusion_pytorch_model.safetensors',
          'text_encoder/pytorch_model.bin',
          'text_encoder/model.safetensors',
          'unet/diffusion_pytorch_model.bin',
          'unet/diffusion_pytorch_model.safetensors',
      ]
      self.assertTrue(is_safetensors_compatible(__magic_name__  )  )
 def 							__UpperCAmelCase						(	self  )   ->							Dict:
      _a      =							[
          'unet/diffusion_pytorch_model.bin',
          'unet/diffusion_pytorch_model.safetensors',
      ]
      self.assertTrue(is_safetensors_compatible(__magic_name__  )  )
 def 							__UpperCAmelCase						(	self  )   ->							Optional[int]:
      _a      =							[
          'safety_checker/pytorch_model.bin',
          'safety_checker/model.safetensors',
          'vae/diffusion_pytorch_model.bin',
          'vae/diffusion_pytorch_model.safetensors',
          'text_encoder/pytorch_model.bin',
          'text_encoder/model.safetensors',
          'unet/diffusion_pytorch_model.bin',
          # Removed: 'unet/diffusion_pytorch_model.safetensors',
      ]
      self.assertFalse(is_safetensors_compatible(__magic_name__  )  )
 def 							__UpperCAmelCase						(	self  )   ->							Optional[Any]:
      _a      =							[
          'text_encoder/pytorch_model.bin',
          'text_encoder/model.safetensors',
      ]
      self.assertTrue(is_safetensors_compatible(__magic_name__  )  )
 def 							__UpperCAmelCase						(	self  )   ->							str:
      _a      =							[
          'safety_checker/pytorch_model.bin',
          'safety_checker/model.safetensors',
          'vae/diffusion_pytorch_model.bin',
          'vae/diffusion_pytorch_model.safetensors',
          'text_encoder/pytorch_model.bin',
          # Removed: 'text_encoder/model.safetensors',
          'unet/diffusion_pytorch_model.bin',
          'unet/diffusion_pytorch_model.safetensors',
      ]
      self.assertFalse(is_safetensors_compatible(__magic_name__  )  )
 def 							__UpperCAmelCase						(	self  )   ->							List[Any]:
      _a      =							[
          'safety_checker/pytorch_model.fp16.bin',
          'safety_checker/model.fp16.safetensors',
          'vae/diffusion_pytorch_model.fp16.bin',
          'vae/diffusion_pytorch_model.fp16.safetensors',
          'text_encoder/pytorch_model.fp16.bin',
          'text_encoder/model.fp16.safetensors',
          'unet/diffusion_pytorch_model.fp16.bin',
          'unet/diffusion_pytorch_model.fp16.safetensors',
      ]
      _a      =							'fp16'
      self.assertTrue(is_safetensors_compatible(__magic_name__ ,			variant=__magic_name__  )  )
 def 							__UpperCAmelCase						(	self  )   ->							Any:
      _a      =							[
          'unet/diffusion_pytorch_model.fp16.bin',
          'unet/diffusion_pytorch_model.fp16.safetensors',
      ]
      _a      =							'fp16'
      self.assertTrue(is_safetensors_compatible(__magic_name__ ,			variant=__magic_name__  )  )
 def 							__UpperCAmelCase						(	self  )   ->							int:
      # pass variant but use the non-variant filenames
      _a      =							[
          'unet/diffusion_pytorch_model.bin',
          'unet/diffusion_pytorch_model.safetensors',
      ]
      _a      =							'fp16'
      self.assertTrue(is_safetensors_compatible(__magic_name__ ,			variant=__magic_name__  )  )
 def 							__UpperCAmelCase						(	self  )   ->							List[str]:
      _a      =							[
          'safety_checker/pytorch_model.fp16.bin',
          'safety_checker/model.fp16.safetensors',
          'vae/diffusion_pytorch_model.fp16.bin',
          'vae/diffusion_pytorch_model.fp16.safetensors',
          'text_encoder/pytorch_model.fp16.bin',
          'text_encoder/model.fp16.safetensors',
          'unet/diffusion_pytorch_model.fp16.bin',
          # Removed: 'unet/diffusion_pytorch_model.fp16.safetensors',
      ]
      _a      =							'fp16'
      self.assertFalse(is_safetensors_compatible(__magic_name__ ,			variant=__magic_name__  )  )
 def 							__UpperCAmelCase						(	self  )   ->							Union[str, Any]:
      _a      =							[
          'text_encoder/pytorch_model.fp16.bin',
          'text_encoder/model.fp16.safetensors',
      ]
      _a      =							'fp16'
      self.assertTrue(is_safetensors_compatible(__magic_name__ ,			variant=__magic_name__  )  )
 def 							__UpperCAmelCase						(	self  )   ->							List[Any]:
      # pass variant but use the non-variant filenames
      _a      =							[
          'text_encoder/pytorch_model.bin',
          'text_encoder/model.safetensors',
      ]
      _a      =							'fp16'
      self.assertTrue(is_safetensors_compatible(__magic_name__ ,			variant=__magic_name__  )  )




 def 							__UpperCAmelCase						(	self  )   ->							List[Any]:
      _a      =							[
          'safety_checker/pytorch_model.fp16.bin',
          'safety_checker/model.fp16.safetensors',
          'vae/diffusion_pytorch_model.fp16.bin',
          'vae/diffusion_pytorch_model.fp16.safetensors',
          'text_encoder/pytorch_model.fp16.bin',
          # 'text_encoder/model.fp16.safetensors',
          'unet/diffusion_pytorch_model.fp16.bin',
          'unet/diffusion_pytorch_model.fp16.safetensors',
      ]
      _a      =							'fp16'
      self.assertFalse(is_safetensors_compatible(__magic_name__ ,			variant=__magic_name__  )  )

