


'''simple docstring'''







import random
from typing import List, Optional, Union

import numpy as np
import torch

from ..state import AcceleratorState
from .constants import CUDA_DISTRIBUTED_TYPES
from .dataclasses import DistributedType, RNGType
from .imports import is_npu_available, is_tpu_available, is_xpu_available


if is_tpu_available(check_device=False):
					import torch_xla.core.xla_model as xm





def     lowerCamelCase_ (					SCREAMING_SNAKE_CASE__ :     int,     SCREAMING_SNAKE_CASE__ :     bool = False				)	->		int:
						if device_specific:
												seed += AcceleratorState().process_index
						random.seed(SCREAMING_SNAKE_CASE__				)
						np.random.seed(SCREAMING_SNAKE_CASE__				)
						torch.manual_seed(SCREAMING_SNAKE_CASE__				)
						if is_xpu_available():
												torch.xpu.manual_seed_all(SCREAMING_SNAKE_CASE__				)
						elif is_npu_available():
												torch.npu.manual_seed_all(SCREAMING_SNAKE_CASE__				)
						else:
												torch.cuda.manual_seed_all(SCREAMING_SNAKE_CASE__				)
						# ^^ safe to call this function even if cuda is not available
						if is_tpu_available():
												xm.set_rng_state(SCREAMING_SNAKE_CASE__				)





def     lowerCamelCase_ (					SCREAMING_SNAKE_CASE__ :     Optional[RNGType] = None,     SCREAMING_SNAKE_CASE__ :     Optional[torch.Generator] = None				)	->		Any:
						# Get the proper rng state
						if rng_type == RNGType.TORCH:
												UpperCAmelCase_     : Union[str, Any]         =					torch.get_rng_state()
						elif rng_type == RNGType.CUDA:
												UpperCAmelCase_     : List[str]         =					torch.cuda.get_rng_state()
						elif rng_type == RNGType.XLA:
												assert is_tpu_available(), "Can't synchronize XLA seeds on an environment without TPUs."
												UpperCAmelCase_     : List[Any]         =					torch.tensor(xm.get_rng_state()				)
						elif rng_type == RNGType.NPU:
												assert is_npu_available(), "Can't synchronize NPU seeds on an environment without NPUs."
												UpperCAmelCase_     : List[Any]         =					torch.npu.get_rng_state()
						elif rng_type == RNGType.XPU:
												assert is_xpu_available(), "Can't synchronize XPU seeds on an environment without XPUs."
												UpperCAmelCase_     : int         =					torch.xpu.get_rng_state()
						elif rng_type == RNGType.GENERATOR:
												assert generator is not None, "Need a generator to synchronize its seed."
												UpperCAmelCase_     : Optional[Any]         =					generator.get_state()

						# Broadcast the rng state from device 0 to other devices
						UpperCAmelCase_     : Union[str, Any]         =					AcceleratorState()
						if state.distributed_type == DistributedType.TPU:
												UpperCAmelCase_     : Tuple         =					rng_state.to(xm.xla_device()				)
												xm.collective_broadcast([rng_state]				)
												xm.mark_step()
												UpperCAmelCase_     : str         =					rng_state.cpu()
						elif (
						    state.distributed_type in CUDA_DISTRIBUTED_TYPES
						    or state.distributed_type == DistributedType.MULTI_NPU
						    or state.distributed_type == DistributedType.MULTI_XPU
						):
												UpperCAmelCase_     : Dict         =					rng_state.to(state.device				)
												torch.distributed.broadcast(SCREAMING_SNAKE_CASE__,     0				)
												UpperCAmelCase_     : List[str]         =					rng_state.cpu()
						elif state.distributed_type == DistributedType.MULTI_CPU:
												torch.distributed.broadcast(SCREAMING_SNAKE_CASE__,     0				)

						# Set the broadcast rng state
						if rng_type == RNGType.TORCH:
												torch.set_rng_state(SCREAMING_SNAKE_CASE__				)
						elif rng_type == RNGType.CUDA:
												torch.cuda.set_rng_state(SCREAMING_SNAKE_CASE__				)
						elif rng_type == RNGType.NPU:
												torch.npu.set_rng_state(SCREAMING_SNAKE_CASE__				)
						elif rng_type == RNGType.XPU:
												torch.xpu.set_rng_state(SCREAMING_SNAKE_CASE__				)
						elif rng_type == RNGType.XLA:
												xm.set_rng_state(rng_state.item()				)
						elif rng_type == RNGType.GENERATOR:
												generator.set_state(SCREAMING_SNAKE_CASE__				)





def     lowerCamelCase_ (					SCREAMING_SNAKE_CASE__ :     List[Union[str, RNGType]],     SCREAMING_SNAKE_CASE__ :     Optional[torch.Generator] = None				)	->		int:
						for rng_type in rng_types:
												synchronize_rng_state(RNGType(SCREAMING_SNAKE_CASE__				),     generator=SCREAMING_SNAKE_CASE__				)












'''simple docstring'''







from typing import List, Optional, Tuple, Union

import torch

from ...models import UNetaDModel
from ...schedulers import PNDMScheduler
from ...utils import randn_tensor
from ..pipeline_utils import DiffusionPipeline, ImagePipelineOutput


class    __a (lowerCamelCase       ):
							__a   :				UNetaDModel
							__a   :				PNDMScheduler

							def __init__(							self							:       Optional[Any]    ,       __magic_name__							:       UNetaDModel    ,       __magic_name__							:       PNDMScheduler				)   ->    Tuple:







													"""simple docstring"""
													super().__init__()

													UpperCAmelCase_     : Optional[int]         =					PNDMScheduler.from_config(scheduler.config				)

													self.register_modules(unet=__magic_name__    ,       scheduler=__magic_name__				)



							@torch.no_grad()
							def __call__(							self							:       Union[str, Any]    ,       __magic_name__							:       int = 1    ,       __magic_name__							:       int = 50    ,       __magic_name__							:       Optional[Union[torch.Generator, List[torch.Generator]]] = None    ,       __magic_name__							:       Optional[str] = "pil"    ,       __magic_name__							:       bool = True    ,       **__magic_name__							:       Dict    ,       )   ->    Union[ImagePipelineOutput, Tuple]:







													"""simple docstring"""
													UpperCAmelCase_     : Dict         =					randn_tensor(
													    (batch_size, self.unet.config.in_channels, self.unet.config.sample_size, self.unet.config.sample_size)    ,       generator=__magic_name__    ,       device=self.device    ,       )

													self.scheduler.set_timesteps(__magic_name__				)
													for t in self.progress_bar(self.scheduler.timesteps				):
																			UpperCAmelCase_     : Optional[int]         =					self.unet(__magic_name__    ,       __magic_name__				).sample

																			UpperCAmelCase_     : str         =					self.scheduler.step(__magic_name__    ,       __magic_name__    ,       __magic_name__				).prev_sample

													UpperCAmelCase_     : int         =					(image / 2 + 0.5).clamp(0    ,       1				)
													UpperCAmelCase_     : Any         =					image.cpu().permute(0    ,       2    ,       3    ,       1				).numpy()
													if output_type == "pil":
																			UpperCAmelCase_     : Optional[int]         =					self.numpy_to_pil(__magic_name__				)

													if not return_dict:
																			return (image,)

													return ImagePipelineOutput(images=__magic_name__				)





