
import random
import unittest

import numpy as np
import torch

from diffusers import DDIMScheduler, LDMSuperResolutionPipeline, UNetaDModel, VQModel
from diffusers.utils import PIL_INTERPOLATION, floats_tensor, load_image, slow, torch_device
from diffusers.utils.testing_utils import enable_full_determinism, require_torch


enable_full_determinism()


class   __lowerCAmelCase		(		unittest.TestCase ):



      """simple docstring"""







      @property
      def 						snake_case_     ( self						:    Tuple			):
          __lowercase						:     str		      =    1
          __lowercase						:     Dict		      =    3
          __lowercase						:     Optional[Any]		      =    (32, 32)

          __lowercase						:     Dict		      =    floats_tensor((batch_size, num_channels) + sizes			,    rng=random.Random(0			)			).to(_snake_case			)
          return image




      @property
      def 						snake_case_     ( self						:    int			):
          torch.manual_seed(0			)
          __lowercase						:     List[str]		      =    UNetaDModel(
              block_out_channels=(32, 64)			,    layers_per_block=2			,    sample_size=32			,    in_channels=6			,    out_channels=3			,    down_block_types=('''DownBlock2D''', '''AttnDownBlock2D''')			,    up_block_types=('''AttnUpBlock2D''', '''UpBlock2D''')			,    )
          return model




      @property
      def 						snake_case_     ( self						:    Union[str, Any]			):
          torch.manual_seed(0			)
          __lowercase						:     List[Any]		      =    VQModel(
              block_out_channels=[32, 64]			,    in_channels=3			,    out_channels=3			,    down_block_types=['''DownEncoderBlock2D''', '''DownEncoderBlock2D''']			,    up_block_types=['''UpDecoderBlock2D''', '''UpDecoderBlock2D''']			,    latent_channels=3			,    )
          return model




      def 						snake_case_     ( self						:    int			):
          __lowercase						:     List[Any]		      =    '''cpu'''
          __lowercase						:     Optional[int]		      =    self.dummy_uncond_unet
          __lowercase						:     List[str]		      =    DDIMScheduler()
          __lowercase						:     List[Any]		      =    self.dummy_vq_model

          __lowercase						:     Optional[int]		      =    LDMSuperResolutionPipeline(unet=_snake_case			,    vqvae=_snake_case			,    scheduler=_snake_case			)
          ldm.to(_snake_case			)
          ldm.set_progress_bar_config(disable=_snake_case			)

          __lowercase						:     Union[str, Any]		      =    self.dummy_image.to(_snake_case			)

          __lowercase						:     Tuple		      =    torch.Generator(device=_snake_case			).manual_seed(0			)
          __lowercase						:     Union[str, Any]		      =    ldm(image=_snake_case			,    generator=_snake_case			,    num_inference_steps=2			,    output_type='''numpy'''			).images

          __lowercase						:     Optional[int]		      =    image[0, -3:, -3:, -1]

          assert image.shape == (1, 64, 64, 3)
          __lowercase						:     Tuple		      =    np.array([0.86_78, 0.82_45, 0.63_81, 0.68_30, 0.43_85, 0.55_99, 0.46_41, 0.62_01, 0.51_50]			)

          assert np.abs(image_slice.flatten() - expected_slice			).max() < 1E-2




      @unittest.skipIf(torch_device != '''cuda'''			,    '''This test requires a GPU'''			)
      def 						snake_case_     ( self						:    List[Any]			):
          __lowercase						:     Union[str, Any]		      =    self.dummy_uncond_unet
          __lowercase						:     Union[str, Any]		      =    DDIMScheduler()
          __lowercase						:     Union[str, Any]		      =    self.dummy_vq_model

          # put models in fp16
          __lowercase						:     Any		      =    unet.half()
          __lowercase						:     str		      =    vqvae.half()

          __lowercase						:     List[str]		      =    LDMSuperResolutionPipeline(unet=_snake_case			,    vqvae=_snake_case			,    scheduler=_snake_case			)
          ldm.to(_snake_case			)
          ldm.set_progress_bar_config(disable=_snake_case			)

          __lowercase						:     Tuple		      =    self.dummy_image.to(_snake_case			)

          __lowercase						:     Optional[Any]		      =    ldm(_snake_case			,    num_inference_steps=2			,    output_type='''numpy'''			).images

          assert image.shape == (1, 64, 64, 3)






@slow
@require_torch
class   __lowerCAmelCase		(		unittest.TestCase ):



      """simple docstring"""







      def 						snake_case_     ( self						:    Union[str, Any]			):
          __lowercase						:     Any		      =    load_image(
              '''https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main'''
              '''/vq_diffusion/teddy_bear_pool.png'''			)
          __lowercase						:     Optional[int]		      =    init_image.resize((64, 64)			,    resample=PIL_INTERPOLATION['''lanczos''']			)

          __lowercase						:     Any		      =    LDMSuperResolutionPipeline.from_pretrained('''duongna/ldm-super-resolution'''			,    device_map='''auto'''			)
          ldm.set_progress_bar_config(disable=_snake_case			)

          __lowercase						:     List[str]		      =    torch.manual_seed(0			)
          __lowercase						:     Optional[int]		      =    ldm(image=_snake_case			,    generator=_snake_case			,    num_inference_steps=20			,    output_type='''numpy'''			).images

          __lowercase						:     Any		      =    image[0, -3:, -3:, -1]

          assert image.shape == (1, 256, 256, 3)
          __lowercase						:     Dict		      =    np.array([0.76_44, 0.76_79, 0.76_42, 0.76_33, 0.76_66, 0.75_60, 0.74_25, 0.72_57, 0.69_07]			)

          assert np.abs(image_slice.flatten() - expected_slice			).max() < 1E-2




from typing import Dict, List, Optional, Union

import numpy as np

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


__lowerCAmelCase   :     Dict      =						logging.get_logger(__name__)


class   __lowerCAmelCase		(		lowerCAmelCase_ ):



      """simple docstring"""







      A__		:     Union[str, Any]       				=      ['''pixel_values''']




      def __init__( self						:    int			,    _snake_case						:    bool = True			,    _snake_case						:    Optional[Dict[str, int]] = None			,    _snake_case						:    PILImageResampling = PILImageResampling.BILINEAR			,    _snake_case						:    bool = True			,    _snake_case						:    Union[int, float] = 1 / 255			,    _snake_case						:    bool = True			,    _snake_case						:    Optional[Union[float, List[float]]] = None			,    _snake_case						:    Optional[Union[float, List[float]]] = None			,    **_snake_case						:    List[str]			,    ):
          super().__init__(**_snake_case			)
          __lowercase						:     Tuple		      =    size if size is not None else {'''height''': 224, '''width''': 224}
          __lowercase						:     List[str]		      =    get_size_dict(_snake_case			)
          __lowercase						:     Union[str, Any]		      =    do_resize
          __lowercase						:     int		      =    do_rescale
          __lowercase						:     List[Any]		      =    do_normalize
          __lowercase						:     Dict		      =    size
          __lowercase						:     Any		      =    resample
          __lowercase						:     int		      =    rescale_factor
          __lowercase						:     Optional[Any]		      =    image_mean if image_mean is not None else IMAGENET_STANDARD_MEAN
          __lowercase						:     Optional[Any]		      =    image_std if image_std is not None else IMAGENET_STANDARD_STD




      def 						snake_case_     ( self						:    int			,    _snake_case						:    np.ndarray			,    _snake_case						:    Dict[str, int]			,    _snake_case						:    PILImageResampling = PILImageResampling.BILINEAR			,    _snake_case						:    Optional[Union[str, ChannelDimension]] = None			,    **_snake_case						:    Optional[int]			,    ):
          __lowercase						:     List[Any]		      =    get_size_dict(_snake_case			)
          if "height" not in size or "width" not in size:
              raise ValueError(F'The `size` dictionary must contain the keys `height` and `width`. Got {size.keys()}'			)
          return resize(
              _snake_case			,    size=(size['''height'''], size['''width'''])			,    resample=_snake_case			,    data_format=_snake_case			,    **_snake_case			)




      def 						snake_case_     ( self						:    Union[str, Any]			,    _snake_case						:    np.ndarray			,    _snake_case						:    float			,    _snake_case						:    Optional[Union[str, ChannelDimension]] = None			,    **_snake_case						:    Dict			):
          return rescale(_snake_case			,    scale=_snake_case			,    data_format=_snake_case			,    **_snake_case			)




      def 						snake_case_     ( self						:    Any			,    _snake_case						:    np.ndarray			,    _snake_case						:    Union[float, List[float]]			,    _snake_case						:    Union[float, List[float]]			,    _snake_case						:    Optional[Union[str, ChannelDimension]] = None			,    **_snake_case						:    Any			,    ):
          return normalize(_snake_case			,    mean=_snake_case			,    std=_snake_case			,    data_format=_snake_case			,    **_snake_case			)




      def 						snake_case_     ( self						:    Dict			,    _snake_case						:    ImageInput			,    _snake_case						:    Optional[bool] = None			,    _snake_case						:    Dict[str, int] = None			,    _snake_case						:    PILImageResampling = None			,    _snake_case						:    Optional[bool] = None			,    _snake_case						:    Optional[float] = None			,    _snake_case						:    Optional[bool] = None			,    _snake_case						:    Optional[Union[float, List[float]]] = None			,    _snake_case						:    Optional[Union[float, List[float]]] = None			,    _snake_case						:    Optional[Union[str, TensorType]] = None			,    _snake_case						:    Union[str, ChannelDimension] = ChannelDimension.FIRST			,    **_snake_case						:    str			,    ):
          __lowercase						:     List[Any]		      =    do_resize if do_resize is not None else self.do_resize
          __lowercase						:     Any		      =    do_rescale if do_rescale is not None else self.do_rescale
          __lowercase						:     Optional[Any]		      =    do_normalize if do_normalize is not None else self.do_normalize
          __lowercase						:     Optional[Any]		      =    resample if resample is not None else self.resample
          __lowercase						:     Tuple		      =    rescale_factor if rescale_factor is not None else self.rescale_factor
          __lowercase						:     Dict		      =    image_mean if image_mean is not None else self.image_mean
          __lowercase						:     Any		      =    image_std if image_std is not None else self.image_std

          __lowercase						:     Optional[int]		      =    size if size is not None else self.size
          __lowercase						:     Tuple		      =    get_size_dict(_snake_case			)

          __lowercase						:     Optional[int]		      =    make_list_of_images(_snake_case			)

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

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

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

          # All transformations expect numpy arrays.
          __lowercase						:     Optional[Any]		      =    [to_numpy_array(_snake_case			) for image in images]

          if do_resize:
              __lowercase						:     List[Any]		      =    [self.resize(image=_snake_case			,    size=_snake_case			,    resample=_snake_case			) for image in images]

          if do_rescale:
              __lowercase						:     Tuple		      =    [self.rescale(image=_snake_case			,    scale=_snake_case			) for image in images]

          if do_normalize:
              __lowercase						:     Union[str, Any]		      =    [self.normalize(image=_snake_case			,    mean=_snake_case			,    std=_snake_case			) for image in images]

          __lowercase						:     Tuple		      =    [to_channel_dimension_format(_snake_case			,    _snake_case			) for image in images]

          __lowercase						:     List[str]		      =    {'''pixel_values''': images}
          return BatchFeature(data=_snake_case			,    tensor_type=_snake_case			)


