def    lowerCamelCase_   (  _lowercase = "mytext.txt" ,     _lowercase = False    )							->	None:
		import socket

		__A							:   List[Any]	  =				12_312  # Reserve a port for your service.
		__A							:   Tuple	  =				socket.socket()  # Create a socket object
		__A							:   Any	  =				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:
				__A  ,       __A							:   Union[str, Any]	  =				sock.accept()  # Establish connection with client.
				print(F"Got connection from {addr}"    )
				__A							:   Optional[Any]	  =				conn.recv(1_024    )
				print(F"Server received: {data = }"    )

				with open(_lowercase ,     "rb"    ) as in_file:
						__A							:   List[str]	  =				in_file.read(1_024    )
						while data:
								conn.send(_lowercase    )
								print(F"Sent {data!r}"    )
								__A							:   str	  =				in_file.read(1_024    )

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

		sock.shutdown(1    )
		sock.close()


if __name__ == "__main__":
					send_file()






import gc
import os
import tempfile
import unittest

import torch
import torch.nn as nn
import torch.nn.functional as F
from transformers import CLIPTextConfig, CLIPTextModel, CLIPTokenizer

from diffusers import AutoencoderKL, DDIMScheduler, StableDiffusionPipeline, UNetaDConditionModel
from diffusers.loaders import AttnProcsLayers, LoraLoaderMixin
from diffusers.models.attention_processor import (
    Attention,
    AttnProcessor,
    AttnProcessor2_0,
    LoRAAttnProcessor,
    LoRAAttnProcessor2_0,
    LoRAXFormersAttnProcessor,
    XFormersAttnProcessor,
)
from diffusers.utils import TEXT_ENCODER_ATTN_MODULE, floats_tensor, torch_device



def    lowerCamelCase_   (  _lowercase    )							->	Optional[Any]:
		__A							:   int	  =				{}
		for name in unet.attn_processors.keys():
				__A							:   Optional[int]	  =				None if name.endswith("attn1.processor"    ) else unet.config.cross_attention_dim
				if name.startswith("mid_block"    ):
						__A							:   Optional[Any]	  =				unet.config.block_out_channels[-1]
				elif name.startswith("up_blocks"    ):
						__A							:   Any	  =				int(name[len("up_blocks."    )]    )
						__A							:   List[str]	  =				list(reversed(unet.config.block_out_channels    )    )[block_id]
				elif name.startswith("down_blocks"    ):
						__A							:   List[str]	  =				int(name[len("down_blocks."    )]    )
						__A							:   int	  =				unet.config.block_out_channels[block_id]
				__A							:   str	  =				(
				    LoRAAttnProcessor2_0 if hasattr(_lowercase ,     "scaled_dot_product_attention"    ) else LoRAAttnProcessor
				)
				__A							:   int	  =				lora_attn_processor_class(
				    hidden_size=_lowercase ,     cross_attention_dim=_lowercase    )
		__A							:   Tuple	  =				AttnProcsLayers(_lowercase    )
		return lora_attn_procs, unet_lora_layers



def    lowerCamelCase_   (  _lowercase    )							->	Optional[int]:
		__A							:   int	  =				{}
		__A							:   int	  =				(
		    LoRAAttnProcessor2_0 if hasattr(_lowercase ,     "scaled_dot_product_attention"    ) else LoRAAttnProcessor
		)
		for name, module in text_encoder.named_modules():
				if name.endswith(_lowercase    ):
						__A							:   Any	  =				lora_attn_processor_class(
						    hidden_size=module.out_proj.out_features ,     cross_attention_dim=_lowercase    )
		return text_lora_attn_procs



def    lowerCamelCase_   (  _lowercase    )							->	Optional[int]:
		__A							:   List[Any]	  =				create_text_encoder_lora_attn_procs(_lowercase    )
		__A							:   Optional[int]	  =				AttnProcsLayers(_lowercase    )
		return text_encoder_lora_layers



def    lowerCamelCase_   (  _lowercase ,     _lowercase=False    )							->	str:
		for _, attn_proc in text_lora_attn_procs.items():
				# set up.weights
				for layer_name, layer_module in attn_proc.named_modules():
						if layer_name.endswith("_lora"    ):
								__A							:   Optional[int]	  =				(
								    torch.randn_like(layer_module.up.weight    )
								    if randn_weight
								    else torch.zeros_like(layer_module.up.weight    )
								)
								__A							:   List[Any]	  =				torch.nn.Parameter(_lowercase    )





class _a       (	unittest.TestCase ):

	'''simple docstring'''



	def 					__UpperCAmelCase(       self   ):
			torch.manual_seed(0   )
			__A							:   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							,			)
			__A							:   List[str]	  =				DDIMScheduler(
			    beta_start=0.0_00_85							,			beta_end=0.0_12							,			beta_schedule="scaled_linear"							,			clip_sample=__UpperCAmelCase							,			set_alpha_to_one=__UpperCAmelCase							,			steps_offset=1							,			)
			torch.manual_seed(0   )
			__A							:   int	  =				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							,			)
			__A							:   Dict	  =				CLIPTextConfig(
			    bos_token_id=0							,			eos_token_id=2							,			hidden_size=32							,			intermediate_size=37							,			layer_norm_eps=1e-05							,			num_attention_heads=4							,			num_hidden_layers=5							,			pad_token_id=1							,			vocab_size=1_000							,			)
			__A							:   List[str]	  =				CLIPTextModel(__UpperCAmelCase   )
			__A							:   Optional[int]	  =				CLIPTokenizer.from_pretrained("hf-internal-testing/tiny-random-clip"   )

			__A  ,       __A							:   List[str]	  =				create_unet_lora_layers(__UpperCAmelCase   )
			__A							:   Union[str, Any]	  =				create_text_encoder_lora_layers(__UpperCAmelCase   )

			__A							:   Dict	  =				{
			    "unet": unet,
			    "scheduler": scheduler,
			    "vae": vae,
			    "text_encoder": text_encoder,
			    "tokenizer": tokenizer,
			    "safety_checker": None,
			    "feature_extractor": None,
			}
			__A							:   Union[str, Any]	  =				{
			    "unet_lora_layers": unet_lora_layers,
			    "text_encoder_lora_layers": text_encoder_lora_layers,
			    "unet_lora_attn_procs": unet_lora_attn_procs,
			}
			return pipeline_components, lora_components

	def 					__UpperCAmelCase(       self   ):
			__A							:   Optional[Any]	  =				1
			__A							:   Any	  =				10
			__A							:   Tuple	  =				4
			__A							:   str	  =				(32, 32)

			__A							:   int	  =				torch.manual_seed(0   )
			__A							:   Optional[Any]	  =				floats_tensor((batch_size, num_channels) + sizes   )
			__A							:   int	  =				torch.randint(1							,			__UpperCAmelCase							,			size=(batch_size, sequence_length)							,			generator=__UpperCAmelCase   )

			__A							:   Any	  =				{
			    "prompt": "A painting of a squirrel eating a burger",
			    "generator": generator,
			    "num_inference_steps": 2,
			    "guidance_scale": 6.0,
			    "output_type": "numpy",
			}

			return noise, input_ids, pipeline_inputs

			# copied from: https://colab.research.google.com/gist/sayakpaul/df2ef6e1ae6d8c10a49d859883b10860/scratchpad.ipynb

	def 					__UpperCAmelCase(       self   ):
			__A							:   List[str]	  =				77

			__A							:   List[Any]	  =				torch.randint(2							,			56							,			size=(1, max_seq_length)							,			generator=torch.manual_seed(0   )   )

			__A							:   Optional[Any]	  =				{}
			__A							:   Optional[int]	  =				inputs
			return prepared_inputs

	def 					__UpperCAmelCase(       self							,			__UpperCAmelCase   ):
			__A  ,       __A							:   Optional[Any]	  =				self.get_dummy_components()
			LoraLoaderMixin.save_lora_weights(
			    save_directory=__UpperCAmelCase							,			unet_lora_layers=lora_components["unet_lora_layers"]							,			text_encoder_lora_layers=lora_components["text_encoder_lora_layers"]							,			)
			self.assertTrue(os.path.isfile(os.path.join(__UpperCAmelCase							,			"pytorch_lora_weights.bin"   )   )   )

	def 					__UpperCAmelCase(       self   ):
			__A  ,       __A							:   Tuple	  =				self.get_dummy_components()
			__A							:   Optional[Any]	  =				StableDiffusionPipeline(**__UpperCAmelCase   )
			__A							:   Dict	  =				sd_pipe.to(__UpperCAmelCase   )
			sd_pipe.set_progress_bar_config(disable=__UpperCAmelCase   )

			__A  ,       __A  ,       __A							:   Dict	  =				self.get_dummy_inputs()

			__A							:   Any	  =				sd_pipe(**__UpperCAmelCase   ).images
			__A							:   Tuple	  =				original_images[0, -3:, -3:, -1]

			with tempfile.TemporaryDirectory() as tmpdirname:
					LoraLoaderMixin.save_lora_weights(
					    save_directory=__UpperCAmelCase							,			unet_lora_layers=lora_components["unet_lora_layers"]							,			text_encoder_lora_layers=lora_components["text_encoder_lora_layers"]							,			)
					self.assertTrue(os.path.isfile(os.path.join(__UpperCAmelCase							,			"pytorch_lora_weights.bin"   )   )   )
					sd_pipe.load_lora_weights(__UpperCAmelCase   )

			__A							:   str	  =				sd_pipe(**__UpperCAmelCase   ).images
			__A							:   int	  =				lora_images[0, -3:, -3:, -1]

			# Outputs shouldn't match.
			self.assertFalse(torch.allclose(torch.from_numpy(__UpperCAmelCase   )							,			torch.from_numpy(__UpperCAmelCase   )   )   )

	def 					__UpperCAmelCase(       self   ):
			__A  ,       __A							:   Dict	  =				self.get_dummy_components()
			__A							:   int	  =				StableDiffusionPipeline(**__UpperCAmelCase   )
			__A							:   Tuple	  =				sd_pipe.to(__UpperCAmelCase   )
			sd_pipe.set_progress_bar_config(disable=__UpperCAmelCase   )

			__A  ,       __A  ,       __A							:   str	  =				self.get_dummy_inputs()

			__A							:   List[str]	  =				sd_pipe(**__UpperCAmelCase   ).images
			__A							:   Tuple	  =				original_images[0, -3:, -3:, -1]

			with tempfile.TemporaryDirectory() as tmpdirname:
					LoraLoaderMixin.save_lora_weights(
					    save_directory=__UpperCAmelCase							,			unet_lora_layers=lora_components["unet_lora_layers"]							,			text_encoder_lora_layers=lora_components["text_encoder_lora_layers"]							,			safe_serialization=__UpperCAmelCase							,			)
					self.assertTrue(os.path.isfile(os.path.join(__UpperCAmelCase							,			"pytorch_lora_weights.safetensors"   )   )   )
					sd_pipe.load_lora_weights(__UpperCAmelCase   )

			__A							:   Any	  =				sd_pipe(**__UpperCAmelCase   ).images
			__A							:   List[str]	  =				lora_images[0, -3:, -3:, -1]

			# Outputs shouldn't match.
			self.assertFalse(torch.allclose(torch.from_numpy(__UpperCAmelCase   )							,			torch.from_numpy(__UpperCAmelCase   )   )   )

	def 					__UpperCAmelCase(       self   ):
			__A  ,       __A							:   Any	  =				self.get_dummy_components()
			__A							:   Optional[Any]	  =				lora_components["unet_lora_attn_procs"]
			__A							:   Optional[Any]	  =				StableDiffusionPipeline(**__UpperCAmelCase   )
			__A							:   Any	  =				sd_pipe.to(__UpperCAmelCase   )
			sd_pipe.set_progress_bar_config(disable=__UpperCAmelCase   )

			__A  ,       __A  ,       __A							:   List[str]	  =				self.get_dummy_inputs()

			__A							:   str	  =				sd_pipe(**__UpperCAmelCase   ).images
			__A							:   List[Any]	  =				original_images[0, -3:, -3:, -1]

			with tempfile.TemporaryDirectory() as tmpdirname:
					__A							:   Dict	  =				sd_pipe.unet
					unet.set_attn_processor(__UpperCAmelCase   )
					unet.save_attn_procs(__UpperCAmelCase   )
					self.assertTrue(os.path.isfile(os.path.join(__UpperCAmelCase							,			"pytorch_lora_weights.bin"   )   )   )
					sd_pipe.load_lora_weights(__UpperCAmelCase   )

			__A							:   Optional[Any]	  =				sd_pipe(**__UpperCAmelCase   ).images
			__A							:   Tuple	  =				lora_images[0, -3:, -3:, -1]

			# Outputs shouldn't match.
			self.assertFalse(torch.allclose(torch.from_numpy(__UpperCAmelCase   )							,			torch.from_numpy(__UpperCAmelCase   )   )   )

	def 					__UpperCAmelCase(       self   ):
			__A  ,       __A							:   List[str]	  =				self.get_dummy_components()
			__A							:   Dict	  =				StableDiffusionPipeline(**__UpperCAmelCase   )

			__A							:   int	  =				self.get_dummy_tokens()

			# inference without lora
			__A							:   Dict	  =				pipe.text_encoder(**__UpperCAmelCase   )[0]
			assert outputs_without_lora.shape == (1, 77, 32)

			# create lora_attn_procs with zeroed out up.weights
			__A							:   Any	  =				create_text_encoder_lora_attn_procs(pipe.text_encoder   )
			set_lora_up_weights(__UpperCAmelCase							,			randn_weight=__UpperCAmelCase   )

			# monkey patch
			pipe._modify_text_encoder(__UpperCAmelCase   )

			# verify that it's okay to release the text_attn_procs which holds the LoRAAttnProcessor.
			del text_attn_procs
			gc.collect()

			# inference with lora
			__A							:   Union[str, Any]	  =				pipe.text_encoder(**__UpperCAmelCase   )[0]
			assert outputs_with_lora.shape == (1, 77, 32)

			assert torch.allclose(
			    __UpperCAmelCase							,			__UpperCAmelCase   ), "lora_up_weight are all zero, so the lora outputs should be the same to without lora outputs"

			# create lora_attn_procs with randn up.weights
			__A							:   Optional[Any]	  =				create_text_encoder_lora_attn_procs(pipe.text_encoder   )
			set_lora_up_weights(__UpperCAmelCase							,			randn_weight=__UpperCAmelCase   )

			# monkey patch
			pipe._modify_text_encoder(__UpperCAmelCase   )

			# verify that it's okay to release the text_attn_procs which holds the LoRAAttnProcessor.
			del text_attn_procs
			gc.collect()

			# inference with lora
			__A							:   Tuple	  =				pipe.text_encoder(**__UpperCAmelCase   )[0]
			assert outputs_with_lora.shape == (1, 77, 32)

			assert not torch.allclose(
			    __UpperCAmelCase							,			__UpperCAmelCase   ), "lora_up_weight are not zero, so the lora outputs should be different to without lora outputs"

	def 					__UpperCAmelCase(       self   ):
			__A  ,       __A							:   Any	  =				self.get_dummy_components()
			__A							:   Any	  =				StableDiffusionPipeline(**__UpperCAmelCase   )

			__A							:   str	  =				self.get_dummy_tokens()

			# inference without lora
			__A							:   Tuple	  =				pipe.text_encoder(**__UpperCAmelCase   )[0]
			assert outputs_without_lora.shape == (1, 77, 32)

			# create lora_attn_procs with randn up.weights
			__A							:   Union[str, Any]	  =				create_text_encoder_lora_attn_procs(pipe.text_encoder   )
			set_lora_up_weights(__UpperCAmelCase							,			randn_weight=__UpperCAmelCase   )

			# monkey patch
			pipe._modify_text_encoder(__UpperCAmelCase   )

			# verify that it's okay to release the text_attn_procs which holds the LoRAAttnProcessor.
			del text_attn_procs
			gc.collect()

			# inference with lora
			__A							:   List[str]	  =				pipe.text_encoder(**__UpperCAmelCase   )[0]
			assert outputs_with_lora.shape == (1, 77, 32)

			assert not torch.allclose(
			    __UpperCAmelCase							,			__UpperCAmelCase   ), "lora outputs should be different to without lora outputs"

			# remove monkey patch
			pipe._remove_text_encoder_monkey_patch()

			# inference with removed lora
			__A							:   Optional[int]	  =				pipe.text_encoder(**__UpperCAmelCase   )[0]
			assert outputs_without_lora_removed.shape == (1, 77, 32)

			assert torch.allclose(
			    __UpperCAmelCase							,			__UpperCAmelCase   ), "remove lora monkey patch should restore the original outputs"

	def 					__UpperCAmelCase(       self   ):
			__A  ,       __A							:   List[str]	  =				self.get_dummy_components()
			__A							:   Tuple	  =				StableDiffusionPipeline(**__UpperCAmelCase   )
			__A							:   Optional[Any]	  =				sd_pipe.to(__UpperCAmelCase   )
			sd_pipe.set_progress_bar_config(disable=__UpperCAmelCase   )

			__A  ,       __A  ,       __A							:   Tuple	  =				self.get_dummy_inputs()

			with tempfile.TemporaryDirectory() as tmpdirname:
					LoraLoaderMixin.save_lora_weights(
					    save_directory=__UpperCAmelCase							,			unet_lora_layers=lora_components["unet_lora_layers"]							,			text_encoder_lora_layers=lora_components["text_encoder_lora_layers"]							,			)
					self.assertTrue(os.path.isfile(os.path.join(__UpperCAmelCase							,			"pytorch_lora_weights.bin"   )   )   )
					sd_pipe.load_lora_weights(__UpperCAmelCase   )

			__A							:   int	  =				sd_pipe(**__UpperCAmelCase   ).images
			__A							:   List[str]	  =				lora_images[0, -3:, -3:, -1]

			__A							:   Dict	  =				sd_pipe(**__UpperCAmelCase							,			cross_attention_kwargs={"scale": 0.5}   ).images
			__A							:   Optional[int]	  =				lora_images_with_scale[0, -3:, -3:, -1]

			# Outputs shouldn't match.
			self.assertFalse(
			    torch.allclose(torch.from_numpy(__UpperCAmelCase   )							,			torch.from_numpy(__UpperCAmelCase   )   )   )

	def 					__UpperCAmelCase(       self   ):
			with tempfile.TemporaryDirectory() as tmpdirname:
					self.create_lora_weight_file(__UpperCAmelCase   )

					__A  ,       __A							:   Optional[Any]	  =				self.get_dummy_components()
					__A							:   int	  =				StableDiffusionPipeline(**__UpperCAmelCase   )
					__A							:   List[Any]	  =				sd_pipe.to(__UpperCAmelCase   )
					sd_pipe.set_progress_bar_config(disable=__UpperCAmelCase   )

					# check if vanilla attention processors are used
					for _, module in sd_pipe.unet.named_modules():
							if isinstance(__UpperCAmelCase							,			__UpperCAmelCase   ):
									self.assertIsInstance(module.processor							,			(AttnProcessor, AttnProcessor2_0)   )

            # load LoRA weight file
					sd_pipe.load_lora_weights(__UpperCAmelCase   )

					# check if lora attention processors are used
					for _, module in sd_pipe.unet.named_modules():
							if isinstance(__UpperCAmelCase							,			__UpperCAmelCase   ):
									__A							:   Optional[Any]	  =				(
									    LoRAAttnProcessor2_0 if hasattr(__UpperCAmelCase							,			"scaled_dot_product_attention"   ) else LoRAAttnProcessor
									)
									self.assertIsInstance(module.processor							,			__UpperCAmelCase   )

	@unittest.skipIf(torch_device != "cuda"							,			"This test is supposed to run on GPU"   )
	def 					__UpperCAmelCase(       self   ):
			with tempfile.TemporaryDirectory() as tmpdirname:
					self.create_lora_weight_file(__UpperCAmelCase   )

					__A  ,       __A							:   Optional[int]	  =				self.get_dummy_components()
					__A							:   int	  =				StableDiffusionPipeline(**__UpperCAmelCase   )
					__A							:   Optional[int]	  =				sd_pipe.to(__UpperCAmelCase   )
					sd_pipe.set_progress_bar_config(disable=__UpperCAmelCase   )

					# enable XFormers
					sd_pipe.enable_xformers_memory_efficient_attention()

					# check if xFormers attention processors are used
					for _, module in sd_pipe.unet.named_modules():
							if isinstance(__UpperCAmelCase							,			__UpperCAmelCase   ):
									self.assertIsInstance(module.processor							,			__UpperCAmelCase   )

            # load LoRA weight file
					sd_pipe.load_lora_weights(__UpperCAmelCase   )

					# check if lora attention processors are used
					for _, module in sd_pipe.unet.named_modules():
							if isinstance(__UpperCAmelCase							,			__UpperCAmelCase   ):
									self.assertIsInstance(module.processor							,			__UpperCAmelCase   )







	@unittest.skipIf(torch_device != "cuda"							,			"This test is supposed to run on GPU"   )
	def 					__UpperCAmelCase(       self   ):
			__A  ,       __A							:   List[str]	  =				self.get_dummy_components()
			__A							:   Tuple	  =				StableDiffusionPipeline(**__UpperCAmelCase   )
			__A							:   Any	  =				sd_pipe.to(__UpperCAmelCase   )
			sd_pipe.set_progress_bar_config(disable=__UpperCAmelCase   )

			__A  ,       __A  ,       __A							:   Optional[int]	  =				self.get_dummy_inputs()

			# enable XFormers
			sd_pipe.enable_xformers_memory_efficient_attention()

			__A							:   Union[str, Any]	  =				sd_pipe(**__UpperCAmelCase   ).images
			__A							:   Tuple	  =				original_images[0, -3:, -3:, -1]

			with tempfile.TemporaryDirectory() as tmpdirname:
					LoraLoaderMixin.save_lora_weights(
					    save_directory=__UpperCAmelCase							,			unet_lora_layers=lora_components["unet_lora_layers"]							,			text_encoder_lora_layers=lora_components["text_encoder_lora_layers"]							,			)
					self.assertTrue(os.path.isfile(os.path.join(__UpperCAmelCase							,			"pytorch_lora_weights.bin"   )   )   )
					sd_pipe.load_lora_weights(__UpperCAmelCase   )

			__A							:   Optional[Any]	  =				sd_pipe(**__UpperCAmelCase   ).images
			__A							:   Tuple	  =				lora_images[0, -3:, -3:, -1]

			# Outputs shouldn't match.
			self.assertFalse(torch.allclose(torch.from_numpy(__UpperCAmelCase   )							,			torch.from_numpy(__UpperCAmelCase   )   )   )





