



'''simple docstring'''





from __future__ import annotations

import time
from collections.abc import Sequence
from random import randint

from matplotlib import pyplot as plt

def        a_     (							_lowerCAmelCase       ,_lowerCAmelCase       ,_lowerCAmelCase       )      ->      tuple[int | None, int | None, float]:
  if not arr:
    return None, None, 0
  if low == high:
    return low, high, arr[low]

  __lowerCamelCase       :       Optional[Any]           = (low + high) // 2
  __lowerCamelCase       ,__lowerCamelCase       ,__lowerCamelCase       :       Dict           = max_subarray(_lowerCAmelCase       ,_lowerCAmelCase       ,_lowerCAmelCase       )
  __lowerCamelCase       ,__lowerCamelCase       ,__lowerCamelCase       :       Optional[Any]           = max_subarray(_lowerCAmelCase       ,mid + 1       ,_lowerCAmelCase       )
  __lowerCamelCase       ,__lowerCamelCase       ,__lowerCamelCase       :       List[str]           = max_cross_sum(_lowerCAmelCase       ,_lowerCAmelCase       ,_lowerCAmelCase       ,_lowerCAmelCase       )
  if left_sum >= right_sum and left_sum >= cross_sum:
    return left_low, left_high, left_sum
  elif right_sum >= left_sum and right_sum >= cross_sum:
    return right_low, right_high, right_sum
  return cross_left, cross_right, cross_sum

def        a_     (							_lowerCAmelCase       ,_lowerCAmelCase       ,_lowerCAmelCase       ,_lowerCAmelCase       )      ->      tuple[int, int, float]:
  __lowerCamelCase       ,__lowerCamelCase       :       str           = float('-inf'       ), -1
  __lowerCamelCase       ,__lowerCamelCase       :       Any           = float('-inf'       ), -1

  __lowerCamelCase       :       int | float           = 0
  for i in range(_lowerCAmelCase       ,low - 1       ,-1       ):
    summ += arr[i]
    if summ > left_sum:
      __lowerCamelCase       :       List[str]           = summ
      __lowerCamelCase       :       Optional[int]           = i

  __lowerCamelCase       :       List[str]           = 0
  for i in range(mid + 1       ,high + 1       ):
    summ += arr[i]
    if summ > right_sum:
      __lowerCamelCase       :       Dict           = summ
      __lowerCamelCase       :       List[str]           = i

  return max_left, max_right, (left_sum + right_sum)

def        a_     (							_lowerCAmelCase       )      ->      float:
  __lowerCamelCase       :       Tuple           = [randint(1       ,_lowerCAmelCase       ) for _ in range(_lowerCAmelCase       )]
  __lowerCamelCase       :       List[Any]           = time.time()
  max_subarray(_lowerCAmelCase       ,0       ,input_size - 1       )
  __lowerCamelCase       :       Tuple           = time.time()
  return end - start

def        a_     (							)      ->      None:
  __lowerCamelCase       :       Optional[int]           = [10, 100, 1000, 10000, 50000, 100000, 200000, 300000, 400000, 500000]
  __lowerCamelCase       :       str           = [time_max_subarray(_lowerCAmelCase       ) for input_size in input_sizes]
  print('No of Inputs\t\tTime Taken'       )
  for input_size, runtime in zip(_lowerCAmelCase       ,_lowerCAmelCase       ):
    print(_lowerCAmelCase       ,'\t\t'       ,_lowerCAmelCase       )
  plt.plot(_lowerCAmelCase       ,_lowerCAmelCase       )
  plt.xlabel('Number of Inputs'       )
  plt.ylabel('Time taken in seconds'       )
  plt.show()


if __name__ == "__main__":
   from doctest import testmod

   testmod()











'''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_     (							_lowerCAmelCase       )      ->      str:
  __lowerCamelCase       ,__lowerCamelCase       :       List[Any]           = image.size
  __lowerCamelCase       ,__lowerCamelCase       :       List[str]           = (x - x % 32 for x in (w, h))  # resize to integer multiple of 32
  __lowerCamelCase       :       Optional[Any]           = image.resize((w, h)       ,resample=PIL_INTERPOLATION['lanczos']       )
  __lowerCamelCase       :       List[Any]           = np.array(_lowerCAmelCase       ).astype(np.floataa       ) / 255.0
  __lowerCamelCase       :       Optional[Any]           = image[None].transpose(0       ,3       ,1       ,2       )
  __lowerCamelCase       :       int           = torch.from_numpy(_lowerCAmelCase       )
  return 2.0 * image - 1.0






class 					lowerCamelCase_    (				SCREAMING_SNAKE_CASE_     ):
  """simple docstring"""




  def __init__( self      :	str			,	_a      :	VQModel			,	_a      :	UNetaDModel			,	_a      :	Union[
          DDIMScheduler,
          PNDMScheduler,
          LMSDiscreteScheduler,
          EulerDiscreteScheduler,
          EulerAncestralDiscreteScheduler,
          DPMSolverMultistepScheduler,
      ]			,	)					->					Optional[Any]:
    super().__init__()
    self.register_modules(vqvae=_a			,	unet=_a			,	scheduler=_a   )




  @torch.no_grad()
  def __call__( self      :	List[Any]			,	_a      :	Union[torch.Tensor, PIL.Image.Image] = None			,	_a      :	Optional[int] = 1			,	_a      :	Optional[int] = 100			,	_a      :	Optional[float] = 0.0			,	_a      :	Optional[Union[torch.Generator, List[torch.Generator]]] = None			,	_a      :	Optional[str] = "pil"			,	_a      :	bool = True			,	)					->					Union[Tuple, ImagePipelineOutput]:
    if isinstance(_a			,	PIL.Image.Image   ):
      __lowerCamelCase       :       Any           = 1
    elif isinstance(_a			,	torch.Tensor   ):
      __lowerCamelCase       :       Any           = image.shape[0]
    else:
      raise ValueError(f'`image` has to be of type `PIL.Image.Image` or `torch.Tensor` but is {type(_a   )}'   )

    if isinstance(_a			,	PIL.Image.Image   ):
      __lowerCamelCase       :       List[str]           = preprocess(_a   )

    __lowerCamelCase       ,__lowerCamelCase       :       List[str]           = image.shape[-2:]

    # in_channels should be 6: 3 for latents, 3 for low resolution image
    __lowerCamelCase       :       Union[str, Any]           = (batch_size, self.unet.config.in_channels // 2, height, width)
    __lowerCamelCase       :       Tuple           = next(self.unet.parameters()   ).dtype

    __lowerCamelCase       :       Optional[int]           = randn_tensor(_a			,	generator=_a			,	device=self.device			,	dtype=_a   )

    __lowerCamelCase       :       Optional[int]           = image.to(device=self.device			,	dtype=_a   )

    # set timesteps and move to the correct device
    self.scheduler.set_timesteps(_a			,	device=self.device   )
    __lowerCamelCase       :       Union[str, Any]           = self.scheduler.timesteps

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

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

    # decode the image latents with the VQVAE
    __lowerCamelCase       :       Union[str, Any]           = self.vqvae.decode(_a   ).sample
    __lowerCamelCase       :       Union[str, Any]           = torch.clamp(_a			,	-1.0			,	1.0   )
    __lowerCamelCase       :       List[str]           = image / 2 + 0.5
    __lowerCamelCase       :       Optional[Any]           = image.cpu().permute(0			,	2			,	3			,	1   ).numpy()

    if output_type == "pil":
      __lowerCamelCase       :       Tuple           = self.numpy_to_pil(_a   )

    if not return_dict:
      return (image,)

    return ImagePipelineOutput(images=_a   )






