
'''simple docstring'''

import gc
import random
import unittest

import numpy as np
import torch
from PIL import Image
from transformers import CLIPImageProcessor, CLIPVisionConfig, CLIPVisionModelWithProjection

from diffusers import (
    AutoencoderKL,
    DPMSolverMultistepScheduler,
    PNDMScheduler,
    StableDiffusionImageVariationPipeline,
    UNetaDConditionModel,
)
from diffusers.utils import floats_tensor, load_image, load_numpy, nightly, slow, torch_device
from diffusers.utils.testing_utils import enable_full_determinism, require_torch_gpu

from ..pipeline_params import IMAGE_VARIATION_BATCH_PARAMS, IMAGE_VARIATION_PARAMS
from ..test_pipelines_common import PipelineKarrasSchedulerTesterMixin, PipelineLatentTesterMixin, PipelineTesterMixin


enable_full_determinism()


class 		_a     (							SCREAMING_SNAKE_CASE			,    SCREAMING_SNAKE_CASE			,    SCREAMING_SNAKE_CASE			,    unittest.TestCase  ):







   """simple docstring"""




   A		= StableDiffusionImageVariationPipeline
   A		= IMAGE_VARIATION_PARAMS
   A		= IMAGE_VARIATION_BATCH_PARAMS
   A		= frozenset([]  )
   # TO-DO: update image_params once pipeline is refactored with VaeImageProcessor.preprocess
   A		= frozenset([]  )






   def 					__a			(       self      ):
    torch.manual_seed(0      )
    SCREAMING_SNAKE_CASE  :  Any								=		UNetaDConditionModel(
        block_out_channels=(32, 64)    ,layers_per_block=2    ,sample_size=32    ,in_channels=4    ,out_channels=4    ,down_block_types=('DownBlock2D', 'CrossAttnDownBlock2D')    ,up_block_types=('CrossAttnUpBlock2D', 'UpBlock2D')    ,cross_attention_dim=32    ,)
    SCREAMING_SNAKE_CASE  :  Optional[Any]								=		PNDMScheduler(skip_prk_steps=__SCREAMING_SNAKE_CASE      )
    torch.manual_seed(0      )
    SCREAMING_SNAKE_CASE  :  Dict								=		AutoencoderKL(
        block_out_channels=[32, 64]    ,in_channels=3    ,out_channels=3    ,down_block_types=['DownEncoderBlock2D', 'DownEncoderBlock2D']    ,up_block_types=['UpDecoderBlock2D', 'UpDecoderBlock2D']    ,latent_channels=4    ,)
    torch.manual_seed(0      )
    SCREAMING_SNAKE_CASE  :  Tuple								=		CLIPVisionConfig(
        hidden_size=32    ,projection_dim=32    ,intermediate_size=37    ,layer_norm_eps=1e-05    ,num_attention_heads=4    ,num_hidden_layers=5    ,image_size=32    ,patch_size=4    ,)
    SCREAMING_SNAKE_CASE  :  Optional[Any]								=		CLIPVisionModelWithProjection(__SCREAMING_SNAKE_CASE      )
    SCREAMING_SNAKE_CASE  :  List[str]								=		CLIPImageProcessor(crop_size=32    ,size=32      )

    SCREAMING_SNAKE_CASE  :  Optional[Any]								=		{
        'unet': unet,
        'scheduler': scheduler,
        'vae': vae,
        'image_encoder': image_encoder,
        'feature_extractor': feature_extractor,
        'safety_checker': None,
    }
    return components






   def 					__a			(       self    ,__SCREAMING_SNAKE_CASE    ,__SCREAMING_SNAKE_CASE=0      ):
    SCREAMING_SNAKE_CASE  :  Union[str, Any]								=		floats_tensor((1, 3, 32, 32)    ,rng=random.Random(__SCREAMING_SNAKE_CASE      )      )
    SCREAMING_SNAKE_CASE  :  List[Any]								=		image.cpu().permute(0    ,2    ,3    ,1      )[0]
    SCREAMING_SNAKE_CASE  :  str								=		Image.fromarray(np.uinta(__SCREAMING_SNAKE_CASE      )      ).convert('RGB'      ).resize((32, 32)      )
    if str(__SCREAMING_SNAKE_CASE      ).startswith('mps'      ):
     SCREAMING_SNAKE_CASE  :  List[Any]								=		torch.manual_seed(__SCREAMING_SNAKE_CASE      )
    else:
     SCREAMING_SNAKE_CASE  :  int								=		torch.Generator(device=__SCREAMING_SNAKE_CASE      ).manual_seed(__SCREAMING_SNAKE_CASE      )
    SCREAMING_SNAKE_CASE  :  Union[str, Any]								=		{
        'image': image,
        'generator': generator,
        'num_inference_steps': 2,
        'guidance_scale': 6.0,
        'output_type': 'numpy',
    }
    return inputs






   def 					__a			(       self      ):
    SCREAMING_SNAKE_CASE  :  Union[str, Any]								=		'cpu'  # ensure determinism for the device-dependent torch.Generator
    SCREAMING_SNAKE_CASE  :  Optional[int]								=		self.get_dummy_components()
    SCREAMING_SNAKE_CASE  :  Optional[Any]								=		StableDiffusionImageVariationPipeline(**__SCREAMING_SNAKE_CASE      )
    SCREAMING_SNAKE_CASE  :  Tuple								=		sd_pipe.to(__SCREAMING_SNAKE_CASE      )
    sd_pipe.set_progress_bar_config(disable=__SCREAMING_SNAKE_CASE      )

    SCREAMING_SNAKE_CASE  :  List[Any]								=		self.get_dummy_inputs(__SCREAMING_SNAKE_CASE      )
    SCREAMING_SNAKE_CASE  :  Union[str, Any]								=		sd_pipe(**__SCREAMING_SNAKE_CASE      ).images
    SCREAMING_SNAKE_CASE  :  List[Any]								=		image[0, -3:, -3:, -1]

    assert image.shape == (1, 64, 64, 3)
    SCREAMING_SNAKE_CASE  :  Tuple								=		np.array([0.5239, 0.5723, 0.4796, 0.5049, 0.5550, 0.4685, 0.5329, 0.4891, 0.4921]      )

    assert np.abs(image_slice.flatten() - expected_slice      ).max() < 1e-3






   def 					__a			(       self      ):
    SCREAMING_SNAKE_CASE  :  Any								=		'cpu'  # ensure determinism for the device-dependent torch.Generator
    SCREAMING_SNAKE_CASE  :  Optional[Any]								=		self.get_dummy_components()
    SCREAMING_SNAKE_CASE  :  int								=		StableDiffusionImageVariationPipeline(**__SCREAMING_SNAKE_CASE      )
    SCREAMING_SNAKE_CASE  :  List[Any]								=		sd_pipe.to(__SCREAMING_SNAKE_CASE      )
    sd_pipe.set_progress_bar_config(disable=__SCREAMING_SNAKE_CASE      )

    SCREAMING_SNAKE_CASE  :  Tuple								=		self.get_dummy_inputs(__SCREAMING_SNAKE_CASE      )
    SCREAMING_SNAKE_CASE  :  Optional[Any]								=		2 * [inputs['image']]
    SCREAMING_SNAKE_CASE  :  List[Any]								=		sd_pipe(**__SCREAMING_SNAKE_CASE      )

    SCREAMING_SNAKE_CASE  :  Optional[int]								=		output.images

    SCREAMING_SNAKE_CASE  :  int								=		image[-1, -3:, -3:, -1]

    assert image.shape == (2, 64, 64, 3)
    SCREAMING_SNAKE_CASE  :  List[Any]								=		np.array([0.6892, 0.5637, 0.5836, 0.5771, 0.6254, 0.6409, 0.5580, 0.5569, 0.5289]      )

    assert np.abs(image_slice.flatten() - expected_slice      ).max() < 1e-3






   def 					__a			(       self      ):
    super().test_inference_batch_single_identical(expected_max_diff=3e-3      )



@slow
@require_torch_gpu
class 		_a     (							unittest.TestCase  ):







   """simple docstring"""




   def 					__a			(       self      ):
    super().tearDown()
    gc.collect()
    torch.cuda.empty_cache()






   def 					__a			(       self    ,__SCREAMING_SNAKE_CASE    ,__SCREAMING_SNAKE_CASE="cpu"    ,__SCREAMING_SNAKE_CASE=torch.floataa    ,__SCREAMING_SNAKE_CASE=0      ):
    SCREAMING_SNAKE_CASE  :  Union[str, Any]								=		torch.Generator(device=__SCREAMING_SNAKE_CASE      ).manual_seed(__SCREAMING_SNAKE_CASE      )
    SCREAMING_SNAKE_CASE  :  int								=		load_image(
        'https://huggingface.co/datasets/diffusers/test-arrays/resolve/main'
        '/stable_diffusion_imgvar/input_image_vermeer.png'      )
    SCREAMING_SNAKE_CASE  :  List[str]								=		np.random.RandomState(__SCREAMING_SNAKE_CASE      ).standard_normal((1, 4, 64, 64)      )
    SCREAMING_SNAKE_CASE  :  Dict								=		torch.from_numpy(__SCREAMING_SNAKE_CASE      ).to(device=__SCREAMING_SNAKE_CASE    ,dtype=__SCREAMING_SNAKE_CASE      )
    SCREAMING_SNAKE_CASE  :  Any								=		{
        'image': init_image,
        'latents': latents,
        'generator': generator,
        'num_inference_steps': 3,
        'guidance_scale': 7.5,
        'output_type': 'numpy',
    }
    return inputs






   def 					__a			(       self      ):
    SCREAMING_SNAKE_CASE  :  Any								=		StableDiffusionImageVariationPipeline.from_pretrained(
        'lambdalabs/sd-image-variations-diffusers'    ,safety_checker=__SCREAMING_SNAKE_CASE      )
    SCREAMING_SNAKE_CASE  :  Optional[int]								=		sd_pipe.to(__SCREAMING_SNAKE_CASE      )
    sd_pipe.set_progress_bar_config(disable=__SCREAMING_SNAKE_CASE      )

    SCREAMING_SNAKE_CASE  :  List[str]								=		self.get_inputs(__SCREAMING_SNAKE_CASE      )
    SCREAMING_SNAKE_CASE  :  Dict								=		sd_pipe(**__SCREAMING_SNAKE_CASE      ).images
    SCREAMING_SNAKE_CASE  :  Any								=		image[0, -3:, -3:, -1].flatten()

    assert image.shape == (1, 512, 512, 3)
    SCREAMING_SNAKE_CASE  :  Union[str, Any]								=		np.array([0.8_4491, 0.9_0789, 0.7_5708, 0.7_8734, 0.8_3485, 0.7_0099, 0.6_6938, 0.6_8727, 0.6_1379]      )
    assert np.abs(image_slice - expected_slice      ).max() < 6e-3






   def 					__a			(       self      ):
    SCREAMING_SNAKE_CASE  :  Optional[Any]								=		0

    def callback_fn(__SCREAMING_SNAKE_CASE    ,__SCREAMING_SNAKE_CASE    ,__SCREAMING_SNAKE_CASE      ) -> None:
     SCREAMING_SNAKE_CASE  :  Any								=		True
     nonlocal number_of_steps
     number_of_steps += 1
     if step == 1:
      SCREAMING_SNAKE_CASE  :  Dict								=		latents.detach().cpu().numpy()
      assert latents.shape == (1, 4, 64, 64)
      SCREAMING_SNAKE_CASE  :  Union[str, Any]								=		latents[0, -3:, -3:, -1]
      SCREAMING_SNAKE_CASE  :  Tuple								=		np.array(
          [-0.1621, 0.2837, -0.7979, -0.1221, -1.3057, 0.7681, -2.1191, 0.0464, 1.6309]      )

      assert np.abs(latents_slice.flatten() - expected_slice      ).max() < 5e-2
     elif step == 2:
      SCREAMING_SNAKE_CASE  :  Optional[Any]								=		latents.detach().cpu().numpy()
      assert latents.shape == (1, 4, 64, 64)
      SCREAMING_SNAKE_CASE  :  str								=		latents[0, -3:, -3:, -1]
      SCREAMING_SNAKE_CASE  :  int								=		np.array([0.6299, 1.7500, 1.1992, -2.1582, -1.8994, 0.7334, -0.7090, 1.0137, 1.5273]      )

      assert np.abs(latents_slice.flatten() - expected_slice      ).max() < 5e-2

    SCREAMING_SNAKE_CASE  :  List[Any]								=		False

    SCREAMING_SNAKE_CASE  :  Optional[int]								=		StableDiffusionImageVariationPipeline.from_pretrained(
        'fusing/sd-image-variations-diffusers'    ,safety_checker=__SCREAMING_SNAKE_CASE    ,torch_dtype=torch.floataa    ,)
    pipe.to(__SCREAMING_SNAKE_CASE      )
    pipe.set_progress_bar_config(disable=__SCREAMING_SNAKE_CASE      )
    pipe.enable_attention_slicing()

    SCREAMING_SNAKE_CASE  :  str								=		self.get_inputs(__SCREAMING_SNAKE_CASE    ,dtype=torch.floataa      )
    pipe(**__SCREAMING_SNAKE_CASE    ,callback=__SCREAMING_SNAKE_CASE    ,callback_steps=1      )
    assert callback_fn.has_been_called
    assert number_of_steps == inputs["num_inference_steps"]






   def 					__a			(       self      ):
    torch.cuda.empty_cache()
    torch.cuda.reset_max_memory_allocated()
    torch.cuda.reset_peak_memory_stats()

    SCREAMING_SNAKE_CASE  :  Optional[int]								=		'fusing/sd-image-variations-diffusers'
    SCREAMING_SNAKE_CASE  :  List[str]								=		StableDiffusionImageVariationPipeline.from_pretrained(
        __SCREAMING_SNAKE_CASE    ,safety_checker=__SCREAMING_SNAKE_CASE    ,torch_dtype=torch.floataa      )
    SCREAMING_SNAKE_CASE  :  List[str]								=		pipe.to(__SCREAMING_SNAKE_CASE      )
    pipe.set_progress_bar_config(disable=__SCREAMING_SNAKE_CASE      )
    pipe.enable_attention_slicing(1      )
    pipe.enable_sequential_cpu_offload()

    SCREAMING_SNAKE_CASE  :  Tuple								=		self.get_inputs(__SCREAMING_SNAKE_CASE    ,dtype=torch.floataa      )
    SCREAMING_SNAKE_CASE  :  Union[str, Any]								=		pipe(**__SCREAMING_SNAKE_CASE      )

    SCREAMING_SNAKE_CASE  :  Any								=		torch.cuda.max_memory_allocated()
    # make sure that less than 2.6 GB is allocated
    assert mem_bytes < 2.6 * 10**9



@nightly
@require_torch_gpu
class 		_a     (							unittest.TestCase  ):







   """simple docstring"""




   def 					__a			(       self      ):
    super().tearDown()
    gc.collect()
    torch.cuda.empty_cache()






   def 					__a			(       self    ,__SCREAMING_SNAKE_CASE    ,__SCREAMING_SNAKE_CASE="cpu"    ,__SCREAMING_SNAKE_CASE=torch.floataa    ,__SCREAMING_SNAKE_CASE=0      ):
    SCREAMING_SNAKE_CASE  :  List[Any]								=		torch.Generator(device=__SCREAMING_SNAKE_CASE      ).manual_seed(__SCREAMING_SNAKE_CASE      )
    SCREAMING_SNAKE_CASE  :  str								=		load_image(
        'https://huggingface.co/datasets/diffusers/test-arrays/resolve/main'
        '/stable_diffusion_imgvar/input_image_vermeer.png'      )
    SCREAMING_SNAKE_CASE  :  Optional[Any]								=		np.random.RandomState(__SCREAMING_SNAKE_CASE      ).standard_normal((1, 4, 64, 64)      )
    SCREAMING_SNAKE_CASE  :  Dict								=		torch.from_numpy(__SCREAMING_SNAKE_CASE      ).to(device=__SCREAMING_SNAKE_CASE    ,dtype=__SCREAMING_SNAKE_CASE      )
    SCREAMING_SNAKE_CASE  :  Optional[Any]								=		{
        'image': init_image,
        'latents': latents,
        'generator': generator,
        'num_inference_steps': 50,
        'guidance_scale': 7.5,
        'output_type': 'numpy',
    }
    return inputs






   def 					__a			(       self      ):
    SCREAMING_SNAKE_CASE  :  Any								=		StableDiffusionImageVariationPipeline.from_pretrained('fusing/sd-image-variations-diffusers'      )
    sd_pipe.to(__SCREAMING_SNAKE_CASE      )
    sd_pipe.set_progress_bar_config(disable=__SCREAMING_SNAKE_CASE      )

    SCREAMING_SNAKE_CASE  :  Optional[Any]								=		self.get_inputs(__SCREAMING_SNAKE_CASE      )
    SCREAMING_SNAKE_CASE  :  List[Any]								=		sd_pipe(**__SCREAMING_SNAKE_CASE      ).images[0]

    SCREAMING_SNAKE_CASE  :  Tuple								=		load_numpy(
        'https://huggingface.co/datasets/diffusers/test-arrays/resolve/main'
        '/stable_diffusion_imgvar/lambdalabs_variations_pndm.npy'      )
    SCREAMING_SNAKE_CASE  :  Any								=		np.abs(expected_image - image      ).max()
    assert max_diff < 1e-3






   def 					__a			(       self      ):
    SCREAMING_SNAKE_CASE  :  int								=		StableDiffusionImageVariationPipeline.from_pretrained('fusing/sd-image-variations-diffusers'      )
    SCREAMING_SNAKE_CASE  :  str								=		DPMSolverMultistepScheduler.from_config(sd_pipe.scheduler.config      )
    sd_pipe.to(__SCREAMING_SNAKE_CASE      )
    sd_pipe.set_progress_bar_config(disable=__SCREAMING_SNAKE_CASE      )

    SCREAMING_SNAKE_CASE  :  List[Any]								=		self.get_inputs(__SCREAMING_SNAKE_CASE      )
    SCREAMING_SNAKE_CASE  :  Dict								=		25
    SCREAMING_SNAKE_CASE  :  Optional[Any]								=		sd_pipe(**__SCREAMING_SNAKE_CASE      ).images[0]

    SCREAMING_SNAKE_CASE  :  Optional[Any]								=		load_numpy(
        'https://huggingface.co/datasets/diffusers/test-arrays/resolve/main'
        '/stable_diffusion_imgvar/lambdalabs_variations_dpm_multi.npy'      )
    SCREAMING_SNAKE_CASE  :  Optional[Any]								=		np.abs(expected_image - image      ).max()
    assert max_diff < 1e-3




'''simple docstring'''

def        SCREAMING_SNAKE_CASE_  (	snake_case_				:	str = "mytext.txt"   ,     snake_case_				:	bool = False							)       ->	None:
 import socket

 SCREAMING_SNAKE_CASE  :  Optional[int]								=		1_2312  # Reserve a port for your service.
 SCREAMING_SNAKE_CASE  :  Any								=		socket.socket()  # Create a socket object
 SCREAMING_SNAKE_CASE  :  int								=		socket.gethostname()  # Get local machine name
 sock.bind((host, port)							)  # Bind to the port
 sock.listen(5							)  # Now wait for client connection.

 print('Server listening....'							)

 while True:
  SCREAMING_SNAKE_CASE				, SCREAMING_SNAKE_CASE  :  Optional[Any]								=		sock.accept()  # Establish connection with client.
  print(f"""Got connection from {addr}"""							)
  SCREAMING_SNAKE_CASE  :  Optional[Any]								=		conn.recv(1024							)
  print(f"""Server received: {data = }"""							)

  with open(snake_case_   ,     'rb'							) as in_file:
   SCREAMING_SNAKE_CASE  :  Optional[Any]								=		in_file.read(1024							)
   while data:
    conn.send(snake_case_							)
    print(f"""Sent {data!r}"""							)
    SCREAMING_SNAKE_CASE  :  List[Any]								=		in_file.read(1024							)

  print('Done sending'							)
  conn.close()
  if testing:  # Allow the test to complete
   break

 sock.shutdown(1							)
 sock.close()


if __name__ == "__main__":
   send_file()


