






'''simple docstring'''





from manim import *


class  snake_case__      (					snake_case_):




					def 		A      (  self  :		Any      )    ->							Any:
								UpperCAmelCase_      :       List[str]          =       Rectangle(height=0.5				, width=0.5      )
								UpperCAmelCase_      :       Dict          =       Rectangle(height=0.46				, width=0.46      ).set_stroke(width=0      )

								UpperCAmelCase_      :       Tuple          =       [mem.copy() for i in range(6      )]
								UpperCAmelCase_      :       Any          =       [mem.copy() for i in range(6      )]
								UpperCAmelCase_      :       Tuple          =       VGroup(*UpperCamelCase__      ).arrange(UpperCamelCase__				, buff=0      )
								UpperCAmelCase_      :       Optional[Any]          =       VGroup(*UpperCamelCase__      ).arrange(UpperCamelCase__				, buff=0      )
								UpperCAmelCase_      :       Optional[Any]          =       VGroup(UpperCamelCase__				, UpperCamelCase__      ).arrange(UpperCamelCase__				, buff=0      )
								UpperCAmelCase_      :       Optional[Any]          =       Text('''CPU'''				, font_size=24      )
								UpperCAmelCase_      :       Optional[int]          =       Group(UpperCamelCase__				, UpperCamelCase__      ).arrange(UpperCamelCase__				, buff=0.5				, aligned_edge=UpperCamelCase__      )
								cpu.move_to([-2.5, -0.5, 0]      )
								self.add(UpperCamelCase__      )

								UpperCAmelCase_      :       Optional[Any]          =       [mem.copy() for i in range(1      )]
								UpperCAmelCase_      :       List[str]          =       VGroup(*UpperCamelCase__      ).arrange(UpperCamelCase__				, buff=0      )
								UpperCAmelCase_      :       Optional[int]          =       Text('''GPU'''				, font_size=24      )
								UpperCAmelCase_      :       Tuple          =       Group(UpperCamelCase__				, UpperCamelCase__      ).arrange(UpperCamelCase__				, buff=0.5				, aligned_edge=UpperCamelCase__      )
								gpu.align_to(UpperCamelCase__				, UpperCamelCase__      )
								gpu.set_x(gpu.get_x() - 1      )

								self.add(UpperCamelCase__      )

								UpperCAmelCase_      :       Any          =       [mem.copy() for i in range(6      )]
								UpperCAmelCase_      :       Any          =       VGroup(*UpperCamelCase__      ).arrange(UpperCamelCase__				, buff=0      )

								UpperCAmelCase_      :       Dict          =       Text('''Model'''				, font_size=24      )
								UpperCAmelCase_      :       Optional[Any]          =       Group(UpperCamelCase__				, UpperCamelCase__      ).arrange(UpperCamelCase__				, buff=0.5				, aligned_edge=UpperCamelCase__      )
								model.move_to([3, -1.0, 0]      )

								self.play(
								    Create(UpperCamelCase__				, run_time=1      )				, Create(UpperCamelCase__				, run_time=1      )				, Create(UpperCamelCase__				, run_time=1      )				, )

								UpperCAmelCase_      :       Optional[Any]          =       MarkupText(
								    F"First, an empty model skeleton is loaded\ninto <span fgcolor=\'{YELLOW}\'>memory</span> without using much RAM."				, font_size=24				, )

								UpperCAmelCase_      :       Any          =       Square(side_length=2.2      )
								key.move_to([-5, 2, 0]      )

								UpperCAmelCase_      :       Any          =       MarkupText(
								    F"<b>Key:</b>\n\n<span fgcolor=\'{YELLOW}\'>●</span> Empty Model"				, font_size=18				, )

								key_text.move_to([-5, 2.4, 0]      )

								step_a.move_to([2, 2, 0]      )
								self.play(Write(UpperCamelCase__				, run_time=2.5      )				, Write(UpperCamelCase__      )				, Write(UpperCamelCase__      )      )

								self.add(UpperCamelCase__      )

								UpperCAmelCase_      :       Optional[Any]          =       []
								UpperCAmelCase_      :       Dict          =       []
								UpperCAmelCase_      :       Union[str, Any]          =       []
								for i, rect in enumerate(UpperCamelCase__      ):
											UpperCAmelCase_      :       Any          =       Rectangle(height=0.46				, width=0.46      ).set_stroke(width=0.0      ).set_fill(UpperCamelCase__				, opacity=0.7      )
											cpu_target.move_to(UpperCamelCase__      )
											cpu_target.generate_target()
											UpperCAmelCase_      :       Optional[int]          =       0.46 / 4
											UpperCAmelCase_      :       Any          =       0.46 / 3

											if i == 0:
														cpu_target.target.next_to(cpu_left_col_base[0].get_corner(DOWN + LEFT      )				, buff=0.02				, direction=UpperCamelCase__      )
														cpu_target.target.set_x(cpu_target.target.get_x() + 0.1      )
											elif i == 3:
														cpu_target.target.next_to(cpu_targs[0].target				, direction=UpperCamelCase__				, buff=0.0      )
											else:
														cpu_target.target.next_to(cpu_targs[i - 1].target				, direction=UpperCamelCase__				, buff=0.0      )
											cpu_targs.append(UpperCamelCase__      )

											first_animations.append(rect.animate(run_time=0.5      ).set_stroke(UpperCamelCase__      )      )
											second_animations.append(MoveToTarget(UpperCamelCase__				, run_time=1.5      )      )

								self.play(*UpperCamelCase__      )
								self.play(*UpperCamelCase__      )

								self.wait()






"""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     __lowerCAmelCase		(		lowercase    :    Optional[int]					)      ->		Union[str, Any]:


    """simple docstring"""





    snake_case   ,snake_case       : List[str]    =       image.size
    snake_case   ,snake_case       : int    =       (x - x % 32 for x in (w, h))  # resize to integer multiple of 32
    snake_case       : Any    =       image.resize((w, h)		,				resample=PIL_INTERPOLATION["lanczos"]					)
    snake_case       : Dict    =       np.array(lowercase					).astype(np.floataa					) / 255.0
    snake_case       : List[str]    =       image[None].transpose(0		,				3		,				1		,				2					)
    snake_case       : List[Any]    =       torch.from_numpy(lowercase					)
    return 2.0 * image - 1.0







class _lowerCAmelCase							( snake_case_      ):







   def __init__(      self		,      UpperCamelCase__		,      UpperCamelCase__		,      UpperCamelCase__		,      )      ->						Optional[Any]:

       '''simple docstring'''




       super().__init__()
       self.register_modules(vqvae=UpperCamelCase__		,      unet=UpperCamelCase__		,      scheduler=UpperCamelCase__		)







   @torch.no_grad()
   def __call__(      self		,      UpperCamelCase__ = None		,      UpperCamelCase__ = 1		,      UpperCamelCase__ = 100		,      UpperCamelCase__ = 0.0		,      UpperCamelCase__ = None		,      UpperCamelCase__ = "pil"		,      UpperCamelCase__ = True		,      )      ->						Union[Tuple, ImagePipelineOutput]:

       '''simple docstring'''




       if isinstance(UpperCamelCase__		,      PIL.Image.Image		):
           snake_case       : Tuple    =       1
       elif isinstance(UpperCamelCase__		,      torch.Tensor		):
           snake_case       : str    =       image.shape[0]
       else:
           raise ValueError(F'`image` has to be of type `PIL.Image.Image` or `torch.Tensor` but is {type(UpperCamelCase__		)}'		)

       if isinstance(UpperCamelCase__		,      PIL.Image.Image		):
           snake_case       : Optional[int]    =       preprocess(UpperCamelCase__		)

       snake_case   ,snake_case       : List[str]    =       image.shape[-2:]

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

       snake_case       : int    =       randn_tensor(UpperCamelCase__		,      generator=UpperCamelCase__		,      device=self.device		,      dtype=UpperCamelCase__		)

       snake_case       : List[Any]    =       image.to(device=self.device		,      dtype=UpperCamelCase__		)

       # set timesteps and move to the correct device
       self.scheduler.set_timesteps(UpperCamelCase__		,      device=self.device		)
       snake_case       : Optional[Any]    =       self.scheduler.timesteps

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

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

       # decode the image latents with the VQVAE
       snake_case       : Optional[int]    =       self.vqvae.decode(UpperCamelCase__		).sample
       snake_case       : Tuple    =       torch.clamp(UpperCamelCase__		,      -1.0		,      1.0		)
       snake_case       : Union[str, Any]    =       image / 2 + 0.5
       snake_case       : Optional[Any]    =       image.cpu().permute(0		,      2		,      3		,      1		).numpy()

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

       if not return_dict:
           return (image,)

       return ImagePipelineOutput(images=UpperCamelCase__		)





