



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






class 		_UpperCamelCase						(							lowerCamelCase__					):


			'''simple docstring'''

			def __init__(							self  :				Optional[Any]							,  lowerCAmelCase__  :				VQModel							,  lowerCAmelCase__  :				UNetaDModel							,  lowerCAmelCase__  :				Union[
			        DDIMScheduler,
			        PNDMScheduler,
			        LMSDiscreteScheduler,
			        EulerDiscreteScheduler,
			        EulerAncestralDiscreteScheduler,
			        DPMSolverMultistepScheduler,
			    ]							,  ):





					"""simple docstring"""






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



			@torch.no_grad()
			def __call__(							self  :				Optional[Any]							,  lowerCAmelCase__  :				Union[torch.Tensor, PIL.Image.Image] = None							,  lowerCAmelCase__  :				Optional[int] = 1							,  lowerCAmelCase__  :				Optional[int] = 1_0_0							,  lowerCAmelCase__  :				Optional[float] = 0.0							,  lowerCAmelCase__  :				Optional[Union[torch.Generator, List[torch.Generator]]] = None							,  lowerCAmelCase__  :				Optional[str] = "pil"							,  lowerCAmelCase__  :				bool = True							,  ):





					"""simple docstring"""






					if isinstance(lowerCAmelCase__							,  PIL.Image.Image		):
							__SCREAMING_SNAKE_CASE :	Dict					 =	1
					elif isinstance(lowerCAmelCase__							,  torch.Tensor		):
							__SCREAMING_SNAKE_CASE :	Tuple					 =	image.shape[0]
					else:
							raise ValueError(F"`image` has to be of type `PIL.Image.Image` or `torch.Tensor` but is {type(lowerCAmelCase__		)}"		)

					if isinstance(lowerCAmelCase__							,  PIL.Image.Image		):
							__SCREAMING_SNAKE_CASE :	int					 =	preprocess(lowerCAmelCase__		)

					__SCREAMING_SNAKE_CASE ,		__SCREAMING_SNAKE_CASE :	Union[str, Any]					 =	image.shape[-2:]

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

					__SCREAMING_SNAKE_CASE :	int					 =	randn_tensor(lowerCAmelCase__							,  generator=lowerCAmelCase__							,  device=self.device							,  dtype=lowerCAmelCase__		)

					__SCREAMING_SNAKE_CASE :	Union[str, Any]					 =	image.to(device=self.device							,  dtype=lowerCAmelCase__		)

					# set timesteps and move to the correct device
					self.scheduler.set_timesteps(lowerCAmelCase__							,  device=self.device		)
					__SCREAMING_SNAKE_CASE :	Any					 =	self.scheduler.timesteps

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

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

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

					if output_type == "pil":
							__SCREAMING_SNAKE_CASE :	Tuple					 =	self.numpy_to_pil(lowerCAmelCase__		)

					if not return_dict:
							return (image,)

					return ImagePipelineOutput(images=lowerCAmelCase__		)




'''simple docstring'''




from unittest.mock import Mock, patch

from file_transfer.send_file import send_file


@patch("""socket.socket"""      )
@patch("""builtins.open"""      )
def        lowerCAmelCase_     (			_lowerCamelCase:     Dict ,   _lowerCamelCase:     List[Any]      ):
		# ===== initialization =====
		__SCREAMING_SNAKE_CASE :	Optional[int]					 =	Mock()
		__SCREAMING_SNAKE_CASE :	Optional[int]					 =	conn, Mock()
		__SCREAMING_SNAKE_CASE :	str					 =	iter([1, None]      )
		__SCREAMING_SNAKE_CASE :	Any					 =	lambda _lowerCamelCase      : next(_lowerCamelCase      )

		# ===== invoke =====
		send_file(filename="""mytext.txt""" ,   testing=_lowerCamelCase      )

		# ===== ensurance =====
		sock.assert_called_once()
		sock.return_value.bind.assert_called_once()
		sock.return_value.listen.assert_called_once()
		sock.return_value.accept.assert_called_once()
		conn.recv.assert_called_once()

		file.return_value.__enter__.assert_called_once()
		file.return_value.__enter__.return_value.read.assert_called()

		conn.send.assert_called_once()
		conn.close.assert_called_once()
		sock.return_value.shutdown.assert_called_once()
		sock.return_value.close.assert_called_once()