




"""simple docstring"""






from typing import Dict, List, Optional, Union

import numpy as np

from transformers.utils import is_vision_available
from transformers.utils.generic import TensorType

from ...image_processing_utils import BaseImageProcessor, BatchFeature, get_size_dict
from ...image_transforms import (
    center_crop,
    get_resize_output_image_size,
    normalize,
    rescale,
    resize,
    to_channel_dimension_format,
)
from ...image_utils import (
    IMAGENET_STANDARD_MEAN,
    IMAGENET_STANDARD_STD,
    ChannelDimension,
    ImageInput,
    PILImageResampling,
    is_valid_image,
    to_numpy_array,
    valid_images,
)
from ...utils import logging


if is_vision_available():
				import PIL

UpperCamelCase__						=			logging.get_logger(__name__)

def 							UpperCAmelCase		(		snake_case			:					List[Any]    ):
					if isinstance(snake_case      ,       (list, tuple)    ) and isinstance(videos[0]      ,       (list, tuple)    ) and is_valid_image(videos[0][0]    ):
										return videos

					elif isinstance(snake_case      ,       (list, tuple)    ) and is_valid_image(videos[0]    ):
										return [videos]

					elif is_valid_image(snake_case    ):
										return [[videos]]

					raise ValueError(F'Could not make batched video from {videos}'    )





class   a__			(       UpperCamelCase_  ):
						snake_case__             = ['''pixel_values''']





						def __init__(							self      :	List[Any]     ,a__      :	bool = True     ,a__      :	Dict[str, int] = None     ,a__      :	PILImageResampling = PILImageResampling.BILINEAR     ,a__      :	bool = True     ,a__      :	Dict[str, int] = None     ,a__      :	bool = True     ,a__      :	Union[int, float] = 1 / 255     ,a__      :	bool = True     ,a__      :	bool = True     ,a__      :	Optional[Union[float, List[float]]] = None     ,a__      :	Optional[Union[float, List[float]]] = None     ,**a__      :	Optional[int]     ,)						->       None:




											"""simple docstring"""







											super().__init__(**a__)
											_lowerCAmelCase:List[Any]									=  size if size is not None else {'''shortest_edge''': 256}
											_lowerCAmelCase:Tuple									=  get_size_dict(a__     ,default_to_square=a__)
											_lowerCAmelCase:Optional[Any]									=  crop_size if crop_size is not None else {'''height''': 224, '''width''': 224}
											_lowerCAmelCase:Optional[int]									=  get_size_dict(a__     ,param_name='''crop_size''')

											_lowerCAmelCase:Optional[int]									=  do_resize
											_lowerCAmelCase:str									=  size
											_lowerCAmelCase:Union[str, Any]									=  do_center_crop
											_lowerCAmelCase:Dict									=  crop_size
											_lowerCAmelCase:Any									=  resample
											_lowerCAmelCase:Union[str, Any]									=  do_rescale
											_lowerCAmelCase:Dict									=  rescale_factor
											_lowerCAmelCase:Any									=  offset
											_lowerCAmelCase:Union[str, Any]									=  do_normalize
											_lowerCAmelCase:Union[str, Any]									=  image_mean if image_mean is not None else IMAGENET_STANDARD_MEAN
											_lowerCAmelCase:Union[str, Any]									=  image_std if image_std is not None else IMAGENET_STANDARD_STD





						def       __UpperCamelCase     (							self      :	str     ,a__      :	np.ndarray     ,a__      :	Dict[str, int]     ,a__      :	PILImageResampling = PILImageResampling.BILINEAR     ,a__      :	Optional[Union[str, ChannelDimension]] = None     ,**a__      :	Optional[Any]     ,)						->       np.ndarray:




											"""simple docstring"""







											_lowerCAmelCase:Optional[Any]									=  get_size_dict(a__     ,default_to_square=a__)
											if "shortest_edge" in size:
																_lowerCAmelCase:Union[str, Any]									=  get_resize_output_image_size(a__     ,size['''shortest_edge''']     ,default_to_square=a__)
											elif "height" in size and "width" in size:
																_lowerCAmelCase:List[Any]									=  (size['''height'''], size['''width'''])
											else:
																raise ValueError(F'Size must have \'height\' and \'width\' or \'shortest_edge\' as keys. Got {size.keys()}')
											return resize(a__     ,size=a__     ,resample=a__     ,data_format=a__     ,**a__)





						def       __UpperCamelCase     (							self      :	List[Any]     ,a__      :	np.ndarray     ,a__      :	Dict[str, int]     ,a__      :	Optional[Union[str, ChannelDimension]] = None     ,**a__      :	str     ,)						->       np.ndarray:




											"""simple docstring"""







											_lowerCAmelCase:Dict									=  get_size_dict(a__)
											if "height" not in size or "width" not in size:
																raise ValueError(F'Size must have \'height\' and \'width\' as keys. Got {size.keys()}')
											return center_crop(a__     ,size=(size['''height'''], size['''width'''])     ,data_format=a__     ,**a__)





						def       __UpperCamelCase     (							self      :	List[str]     ,a__      :	np.ndarray     ,a__      :	Union[int, float]     ,a__      :	bool = True     ,a__      :	Optional[Union[str, ChannelDimension]] = None     ,**a__      :	List[str]     ,)						->       List[Any]:




											"""simple docstring"""







											_lowerCAmelCase:Optional[Any]									=  image.astype(np.floataa)
											if offset:
																_lowerCAmelCase:Any									=  image - (scale / 2)
											return rescale(a__     ,scale=a__     ,data_format=a__     ,**a__)





						def       __UpperCamelCase     (							self      :	Dict     ,a__      :	np.ndarray     ,a__      :	Union[float, List[float]]     ,a__      :	Union[float, List[float]]     ,a__      :	Optional[Union[str, ChannelDimension]] = None     ,**a__      :	Tuple     ,)						->       np.ndarray:




											"""simple docstring"""







											return normalize(a__     ,mean=a__     ,std=a__     ,data_format=a__     ,**a__)





						def       __UpperCamelCase     (							self      :	Union[str, Any]     ,a__      :	ImageInput     ,a__      :	bool = None     ,a__      :	Dict[str, int] = None     ,a__      :	PILImageResampling = None     ,a__      :	bool = None     ,a__      :	Dict[str, int] = None     ,a__      :	bool = None     ,a__      :	float = None     ,a__      :	bool = None     ,a__      :	bool = None     ,a__      :	Optional[Union[float, List[float]]] = None     ,a__      :	Optional[Union[float, List[float]]] = None     ,a__      :	Optional[ChannelDimension] = ChannelDimension.FIRST     ,)						->       np.ndarray:




											"""simple docstring"""







											if do_resize and size is None or resample is None:
																raise ValueError('''Size and resample must be specified if do_resize is True.''')

											if do_center_crop and crop_size is None:
																raise ValueError('''Crop size must be specified if do_center_crop is True.''')

											if do_rescale and rescale_factor is None:
																raise ValueError('''Rescale factor must be specified if do_rescale is True.''')

											if do_normalize and (image_mean is None or image_std is None):
																raise ValueError('''Image mean and std must be specified if do_normalize is True.''')

											if offset and not do_rescale:
																raise ValueError('''For offset, do_rescale must also be set to True.''')

											# All transformations expect numpy arrays.
											_lowerCAmelCase:Dict									=  to_numpy_array(a__)

											if do_resize:
																_lowerCAmelCase:Tuple									=  self.resize(image=a__     ,size=a__     ,resample=a__)

											if do_center_crop:
																_lowerCAmelCase:Optional[Any]									=  self.center_crop(a__     ,size=a__)

											if do_rescale:
																_lowerCAmelCase:Optional[int]									=  self.rescale(image=a__     ,scale=a__     ,offset=a__)

											if do_normalize:
																_lowerCAmelCase:Any									=  self.normalize(image=a__     ,mean=a__     ,std=a__)

											_lowerCAmelCase:List[Any]									=  to_channel_dimension_format(a__     ,a__)
											return image







						def       __UpperCamelCase     (							self      :	str     ,a__      :	ImageInput     ,a__      :	bool = None     ,a__      :	Dict[str, int] = None     ,a__      :	PILImageResampling = None     ,a__      :	bool = None     ,a__      :	Dict[str, int] = None     ,a__      :	bool = None     ,a__      :	float = None     ,a__      :	bool = None     ,a__      :	bool = None     ,a__      :	Optional[Union[float, List[float]]] = None     ,a__      :	Optional[Union[float, List[float]]] = None     ,a__      :	Optional[Union[str, TensorType]] = None     ,a__      :	ChannelDimension = ChannelDimension.FIRST     ,**a__      :	Any     ,)						->       PIL.Image.Image:




											"""simple docstring"""







											_lowerCAmelCase:List[str]									=  do_resize if do_resize is not None else self.do_resize
											_lowerCAmelCase:str									=  resample if resample is not None else self.resample
											_lowerCAmelCase:List[Any]									=  do_center_crop if do_center_crop is not None else self.do_center_crop
											_lowerCAmelCase:str									=  do_rescale if do_rescale is not None else self.do_rescale
											_lowerCAmelCase:Optional[int]									=  rescale_factor if rescale_factor is not None else self.rescale_factor
											_lowerCAmelCase:Any									=  offset if offset is not None else self.offset
											_lowerCAmelCase:Tuple									=  do_normalize if do_normalize is not None else self.do_normalize
											_lowerCAmelCase:Any									=  image_mean if image_mean is not None else self.image_mean
											_lowerCAmelCase:Optional[Any]									=  image_std if image_std is not None else self.image_std

											_lowerCAmelCase:Optional[int]									=  size if size is not None else self.size
											_lowerCAmelCase:Tuple									=  get_size_dict(a__     ,default_to_square=a__)
											_lowerCAmelCase:Any									=  crop_size if crop_size is not None else self.crop_size
											_lowerCAmelCase:Union[str, Any]									=  get_size_dict(a__     ,param_name='''crop_size''')

											if not valid_images(a__):
																raise ValueError(
																    '''Invalid image type. Must be of type PIL.Image.Image, numpy.ndarray, '''
																    '''torch.Tensor, tf.Tensor or jax.ndarray.''')

											_lowerCAmelCase:Union[str, Any]									=  make_batched(a__)

											_lowerCAmelCase:Any									=  [
											    [
											        self._preprocess_image(
											            image=a__     ,do_resize=a__     ,size=a__     ,resample=a__     ,do_center_crop=a__     ,crop_size=a__     ,do_rescale=a__     ,rescale_factor=a__     ,offset=a__     ,do_normalize=a__     ,image_mean=a__     ,image_std=a__     ,data_format=a__     ,)
											        for img in video
											    ]
											    for video in videos
											]

											_lowerCAmelCase:Dict									=  {'''pixel_values''': videos}
											return BatchFeature(data=a__     ,tensor_type=a__)












"""simple docstring"""






import torch

from diffusers import KDPMaDiscreteScheduler
from diffusers.utils import torch_device

from .test_schedulers import SchedulerCommonTest




class   a__			(       UpperCamelCase_  ):
						snake_case__             = (KDPMaDiscreteScheduler,)
						snake_case__             = 1_0





						def       __UpperCamelCase     (							self      :	Dict     ,**a__      :	str)						->       int:




											"""simple docstring"""







											_lowerCAmelCase:Optional[int]									=  {
											    '''num_train_timesteps''': 1100,
											    '''beta_start''': 0.0001,
											    '''beta_end''': 0.02,
											    '''beta_schedule''': '''linear''',
											}

											config.update(**a__)
											return config





						def       __UpperCamelCase     (							self      :	Optional[int])						->       Optional[int]:




											"""simple docstring"""







											for timesteps in [10, 50, 100, 1000]:
																self.check_over_configs(num_train_timesteps=a__)





						def       __UpperCamelCase     (							self      :	Union[str, Any])						->       Optional[int]:




											"""simple docstring"""







											for beta_start, beta_end in zip([0.00001, 0.0001, 0.001]     ,[0.0002, 0.002, 0.02]):
																self.check_over_configs(beta_start=a__     ,beta_end=a__)





						def       __UpperCamelCase     (							self      :	Optional[Any])						->       Union[str, Any]:




											"""simple docstring"""







											for schedule in ["linear", "scaled_linear"]:
																self.check_over_configs(beta_schedule=a__)





						def       __UpperCamelCase     (							self      :	str)						->       str:




											"""simple docstring"""







											for prediction_type in ["epsilon", "v_prediction"]:
																self.check_over_configs(prediction_type=a__)





						def       __UpperCamelCase     (							self      :	Tuple)						->       Tuple:




											"""simple docstring"""







											_lowerCAmelCase:Union[str, Any]									=  self.scheduler_classes[0]
											_lowerCAmelCase:Union[str, Any]									=  self.get_scheduler_config(prediction_type='''v_prediction''')
											_lowerCAmelCase:Tuple									=  scheduler_class(**a__)

											scheduler.set_timesteps(self.num_inference_steps)

											_lowerCAmelCase:Optional[Any]									=  self.dummy_model()
											_lowerCAmelCase:List[Any]									=  self.dummy_sample_deter * scheduler.init_noise_sigma
											_lowerCAmelCase:Optional[Any]									=  sample.to(a__)

											for i, t in enumerate(scheduler.timesteps):
																_lowerCAmelCase:Tuple									=  scheduler.scale_model_input(a__     ,a__)

																_lowerCAmelCase:Any									=  model(a__     ,a__)

																_lowerCAmelCase:Union[str, Any]									=  scheduler.step(a__     ,a__     ,a__)
																_lowerCAmelCase:Optional[int]									=  output.prev_sample

											_lowerCAmelCase:str									=  torch.sum(torch.abs(a__))
											_lowerCAmelCase:Union[str, Any]									=  torch.mean(torch.abs(a__))

											if torch_device in ["cpu", "mps"]:
																assert abs(result_sum.item() - 4.6_934E-07) < 1E-2
																assert abs(result_mean.item() - 6.1_112E-10) < 1E-3
											else:
																# CUDA
																assert abs(result_sum.item() - 4.693_428_650_170_972E-07) < 1E-2
																assert abs(result_mean.item() - 0.0002) < 1E-3





						def       __UpperCamelCase     (							self      :	Tuple)						->       Dict:




											"""simple docstring"""







											if torch_device == "mps":
																return
											_lowerCAmelCase:Tuple									=  self.scheduler_classes[0]
											_lowerCAmelCase:List[str]									=  self.get_scheduler_config()
											_lowerCAmelCase:Union[str, Any]									=  scheduler_class(**a__)

											scheduler.set_timesteps(self.num_inference_steps)

											_lowerCAmelCase:Any									=  self.dummy_model()
											_lowerCAmelCase:str									=  self.dummy_sample_deter * scheduler.init_noise_sigma
											_lowerCAmelCase:List[str]									=  sample.to(a__)

											for i, t in enumerate(scheduler.timesteps):
																_lowerCAmelCase:Tuple									=  scheduler.scale_model_input(a__     ,a__)

																_lowerCAmelCase:List[str]									=  model(a__     ,a__)

																_lowerCAmelCase:int									=  scheduler.step(a__     ,a__     ,a__)
																_lowerCAmelCase:List[Any]									=  output.prev_sample

											_lowerCAmelCase:Tuple									=  torch.sum(torch.abs(a__))
											_lowerCAmelCase:Dict									=  torch.mean(torch.abs(a__))

											if torch_device in ["cpu", "mps"]:
																assert abs(result_sum.item() - 20.4125) < 1E-2
																assert abs(result_mean.item() - 0.0266) < 1E-3
											else:
																# CUDA
																assert abs(result_sum.item() - 20.4125) < 1E-2
																assert abs(result_mean.item() - 0.0266) < 1E-3







						def       __UpperCamelCase     (							self      :	Union[str, Any])						->       List[Any]:




											"""simple docstring"""







											if torch_device == "mps":
																return
											_lowerCAmelCase:str									=  self.scheduler_classes[0]
											_lowerCAmelCase:Optional[Any]									=  self.get_scheduler_config()
											_lowerCAmelCase:Any									=  scheduler_class(**a__)

											scheduler.set_timesteps(self.num_inference_steps     ,device=a__)

											_lowerCAmelCase:Optional[int]									=  self.dummy_model()
											_lowerCAmelCase:Optional[Any]									=  self.dummy_sample_deter.to(a__) * scheduler.init_noise_sigma

											for t in scheduler.timesteps:
																_lowerCAmelCase:Union[str, Any]									=  scheduler.scale_model_input(a__     ,a__)

																_lowerCAmelCase:Tuple									=  model(a__     ,a__)

																_lowerCAmelCase:str									=  scheduler.step(a__     ,a__     ,a__)
																_lowerCAmelCase:Dict									=  output.prev_sample

											_lowerCAmelCase:Any									=  torch.sum(torch.abs(a__))
											_lowerCAmelCase:str									=  torch.mean(torch.abs(a__))

											if str(a__).startswith('''cpu'''):
																# The following sum varies between 148 and 156 on mps. Why?
																assert abs(result_sum.item() - 20.4125) < 1E-2
																assert abs(result_mean.item() - 0.0266) < 1E-3
											else:
																# CUDA
																assert abs(result_sum.item() - 20.4125) < 1E-2
																assert abs(result_mean.item() - 0.0266) < 1E-3






