




'''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      a_							(      __snake_case							:			Optional[int]   )							->						Dict:



       """simple docstring"""
       lowerCamelCase_,      lowerCamelCase_							=image.size
       lowerCamelCase_,      lowerCamelCase_							=(x - x % 32 for x in (w, h))  # resize to integer multiple of 32
       lowerCamelCase_							=image.resize((w, h)   ,       resample=PIL_INTERPOLATION['''lanczos''']   )
       lowerCamelCase_							=np.array(__snake_case   ).astype(np.floataa   ) / 2_5_5.0
       lowerCamelCase_							=image[None].transpose(0   ,       3   ,       1   ,       2   )
       lowerCamelCase_							=torch.from_numpy(__snake_case   )
       return 2.0 * image - 1.0


class   __UpperCamelCase       ( _lowerCamelCase				):






       def __init__(							self,		lowerCAmelCase,		lowerCAmelCase,		lowerCAmelCase,		):



              """simple docstring"""







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






       @torch.no_grad()
       def __call__(							self,		lowerCAmelCase = None,		lowerCAmelCase = 1,		lowerCAmelCase = 100,		lowerCAmelCase = 0.0,		lowerCAmelCase = None,		lowerCAmelCase = "pil",		lowerCAmelCase = True,		):



              """simple docstring"""







              if isinstance(_UpperCamelCase,		PIL.Image.Image  ):
                     lowerCamelCase_							=1
              elif isinstance(_UpperCamelCase,		torch.Tensor  ):
                     lowerCamelCase_							=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  ):
                     lowerCamelCase_							=preprocess(_UpperCamelCase  )

              lowerCamelCase_,      lowerCamelCase_							=image.shape[-2:]

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

              lowerCamelCase_							=randn_tensor(_UpperCamelCase,		generator=_UpperCamelCase,		device=self.device,		dtype=_UpperCamelCase  )

              lowerCamelCase_							=image.to(device=self.device,		dtype=_UpperCamelCase  )

              # set timesteps and move to the correct device
              self.scheduler.set_timesteps(_UpperCamelCase,		device=self.device  )
              lowerCamelCase_							=self.scheduler.timesteps

              # scale the initial noise by the standard deviation required by the scheduler
              lowerCamelCase_							=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]
              lowerCamelCase_							='''eta''' in set(inspect.signature(self.scheduler.step  ).parameters.keys()  )
              lowerCamelCase_							={}
              if accepts_eta:
                     lowerCamelCase_							=eta

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

              # decode the image latents with the VQVAE
              lowerCamelCase_							=self.vqvae.decode(_UpperCamelCase  ).sample
              lowerCamelCase_							=torch.clamp(_UpperCamelCase,		-1.0,		1.0  )
              lowerCamelCase_							=image / 2 + 0.5
              lowerCamelCase_							=image.cpu().permute(0,		2,		3,		1  ).numpy()

              if output_type == "pil":
                     lowerCamelCase_							=self.numpy_to_pil(_UpperCamelCase  )

              if not return_dict:
                     return (image,)

              return ImagePipelineOutput(images=_UpperCamelCase  )











import numpy as np
from PIL import Image



def _a				(	lowerCamelCase:   np.ndarray					,				lowerCamelCase:   int					,				lowerCamelCase:   int )	->						np.ndarray:







      '''simple docstring'''






      __A          =				np.array(lowerCamelCase )
      if arr.shape[0] != arr.shape[1]:
            raise ValueError('''The input array is not a square matrix''' )
      __A          =				0
      __A          =				0
      __A          =				0
      __A          =				0

      # compute the shape of the output matrix
      __A          =				(arr.shape[0] - size) // stride + 1
      # initialize the output matrix with zeros of shape maxpool_shape
      __A          =				np.zeros((maxpool_shape, maxpool_shape) )

      while i < arr.shape[0]:
            if i + size > arr.shape[0]:
                  # if the end of the matrix is reached, break
                  break
            while j < arr.shape[1]:
                  # if the end of the matrix is reached, break
                  if j + size > arr.shape[1]:
                        break
                  # compute the maximum of the pooling matrix
                  __A          =				np.max(arr[i : i + size, j : j + size] )
                  # shift the pooling matrix by stride of column pixels
                  j += stride
                  mat_j += 1

            # shift the pooling matrix by stride of row pixels
            i += stride
            mat_i += 1

            # reset the column index to 0
            __A          =				0
            __A          =				0

      return updated_arr



def _a				(	lowerCamelCase:   np.ndarray					,				lowerCamelCase:   int					,				lowerCamelCase:   int )	->						np.ndarray:







      '''simple docstring'''






      __A          =				np.array(lowerCamelCase )
      if arr.shape[0] != arr.shape[1]:
            raise ValueError('''The input array is not a square matrix''' )
      __A          =				0
      __A          =				0
      __A          =				0
      __A          =				0

      # compute the shape of the output matrix
      __A          =				(arr.shape[0] - size) // stride + 1
      # initialize the output matrix with zeros of shape avgpool_shape
      __A          =				np.zeros((avgpool_shape, avgpool_shape) )

      while i < arr.shape[0]:
            # if the end of the matrix is reached, break
            if i + size > arr.shape[0]:
                  break
            while j < arr.shape[1]:
                  # if the end of the matrix is reached, break
                  if j + size > arr.shape[1]:
                        break
                  # compute the average of the pooling matrix
                  __A          =				int(np.average(arr[i : i + size, j : j + size] ) )
                  # shift the pooling matrix by stride of column pixels
                  j += stride
                  mat_j += 1

            # shift the pooling matrix by stride of row pixels
            i += stride
            mat_i += 1
            # reset the column index to 0
            __A          =				0
            __A          =				0

      return updated_arr


# Main Function
if __name__ == "__main__":
       from doctest import testmod

       testmod(name='avgpooling', verbose=True)

       # Loading the image
       snake_case__					:							Optional[Any]													=     Image.open('path_to_image')

       # Converting the image to numpy array and maxpooling, displaying the result
       # Ensure that the image is a square matrix

       Image.fromarray(maxpooling(np.array(image), size=3, stride=2)).show()

       # Converting the image to numpy array and averagepooling, displaying the result
       # Ensure that the image is a square matrix

       Image.fromarray(avgpooling(np.array(image), size=3, stride=2)).show()





