




# limitations under the License.


from typing import Optional, Tuple, Union

import torch

from diffusers.pipeline_utils import DiffusionPipeline, ImagePipelineOutput




class    snake_case__							(_UpperCamelCase       ):
				"""simple docstring"""



				def __init__(				self    :					Union[str, Any] ,				__lowerCamelCase    :					str ,				__lowerCamelCase    :					Dict							) ->					str:
							super().__init__()
							self.register_modules(unet=__lowerCamelCase ,				scheduler=__lowerCamelCase							)





				@torch.no_grad()
				def __call__(				self    :					List[Any] ,				__lowerCamelCase    :					int = 1 ,				__lowerCamelCase    :					Optional[torch.Generator] = None ,				__lowerCamelCase    :					int = 50 ,				__lowerCamelCase    :					Optional[str] = "pil" ,				__lowerCamelCase    :					bool = True ,				**__lowerCamelCase    :					Optional[int] ,				) ->					Union[ImagePipelineOutput, Tuple]:
							a   =      torch.randn(
							    (batch_size, self.unet.config.in_channels, self.unet.config.sample_size, self.unet.config.sample_size) ,				generator=__lowerCamelCase ,				)
							a   =      image.to(self.device							)

							# set step values
							self.scheduler.set_timesteps(__lowerCamelCase							)

							for t in self.progress_bar(self.scheduler.timesteps							):
										# 1. predict noise model_output
										a   =      self.unet(__lowerCamelCase ,				__lowerCamelCase							).sample

										# 2. predict previous mean of image x_t-1 and add variance depending on eta
										# eta corresponds to η in paper and should be between [0, 1]
										# do x_t -> x_t-1
										a   =      self.scheduler.step(__lowerCamelCase ,				__lowerCamelCase ,				__lowerCamelCase							).prev_sample

							a   =      (image / 2 + 0.5).clamp(0 ,				1							)
							a   =      image.cpu().permute(0 ,				2 ,				3 ,				1							).numpy()
							if output_type == "pil":
										a   =      self.numpy_to_pil(__lowerCamelCase							)

							if not return_dict:
										return (image,), "This is a local test"

							return ImagePipelineOutput(images=__lowerCamelCase							), "This is a local test"












import torch
import torch.nn.functional as F

from diffusers import VQDiffusionScheduler

from .test_schedulers import SchedulerCommonTest




class    snake_case__							(_UpperCamelCase       ):
				"""simple docstring"""



				SCREAMING_SNAKE_CASE_	:							Dict										= (VQDiffusionScheduler,)





				def 							__UpperCAmelCase							(				self    :					Tuple ,				**__lowerCamelCase    :					Optional[int]							) ->					Optional[Any]:
							a   =      {
							    "num_vec_classes": 40_97,
							    "num_train_timesteps": 1_00,
							}

							config.update(**__lowerCamelCase							)
							return config





				def 							__UpperCAmelCase							(				self    :					List[str] ,				__lowerCamelCase    :					Any							) ->					Union[str, Any]:
							a   =      4
							a   =      8
							a   =      8

							a   =      torch.randint(0 ,				__lowerCamelCase ,				(batch_size, height * width)							)

							return sample





				@property
				def 							__UpperCAmelCase							(				self    :					List[str]							) ->					Optional[Any]:
							assert False





				def 							__UpperCAmelCase							(				self    :					Any ,				__lowerCamelCase    :					Optional[int]							) ->					int:
							def model(__lowerCamelCase    :					Tuple ,				__lowerCamelCase    :					Any ,				*__lowerCamelCase    :					Optional[Any]							):
										a   ,		a   =      sample.shape
										a   =      torch.rand((batch_size, num_vec_classes - 1, num_latent_pixels)							)
										a   =      F.log_softmax(logits.double() ,				dim=1							).float()
										return return_value

							return model





				def 							__UpperCAmelCase							(				self    :					Optional[Any]							) ->					Any:
							for timesteps in [2, 5, 1_00, 10_00]:
										self.check_over_configs(num_train_timesteps=__lowerCamelCase							)





				def 							__UpperCAmelCase							(				self    :					Any							) ->					List[Any]:
							for num_vec_classes in [5, 1_00, 10_00, 40_00]:
										self.check_over_configs(num_vec_classes=__lowerCamelCase							)





				def 							__UpperCAmelCase							(				self    :					Optional[Any]							) ->					List[Any]:
							for t in [0, 50, 99]:
										self.check_over_forward(time_step=__lowerCamelCase							)





				def 							__UpperCAmelCase							(				self    :					List[Any]							) ->					Optional[Any]:
							pass






