






"""simple docstring"""



# This model implementation is heavily inspired by https://github.com/haofanwang/ControlNet-for-Diffusers/

import gc
import random
import tempfile
import unittest

import numpy as np
import torch
from PIL import Image
from transformers import CLIPTextConfig, CLIPTextModel, CLIPTokenizer

from diffusers import (
    AutoencoderKL,
    ControlNetModel,
    DDIMScheduler,
    StableDiffusionControlNetImgaImgPipeline,
    UNetaDConditionModel,
)
from diffusers.pipelines.stable_diffusion.pipeline_stable_diffusion_controlnet import MultiControlNetModel
from diffusers.utils import floats_tensor, load_image, load_numpy, randn_tensor, slow, torch_device
from diffusers.utils.import_utils import is_xformers_available
from diffusers.utils.testing_utils import enable_full_determinism, require_torch_gpu

from ..pipeline_params import (
    IMAGE_TO_IMAGE_IMAGE_PARAMS,
    TEXT_GUIDED_IMAGE_VARIATION_BATCH_PARAMS,
    TEXT_GUIDED_IMAGE_VARIATION_PARAMS,
)
from ..test_pipelines_common import (
    PipelineKarrasSchedulerTesterMixin,
    PipelineLatentTesterMixin,
    PipelineTesterMixin,
)


enable_full_determinism()






class 							__SCREAMING_SNAKE_CASE  (			UpperCamelCase    ,						UpperCamelCase    ,						UpperCamelCase    ,						unittest.TestCase						):
							snake_case_       =							StableDiffusionControlNetImgaImgPipeline
							snake_case_       =							TEXT_GUIDED_IMAGE_VARIATION_PARAMS - {'height', 'width'}
							snake_case_       =							TEXT_GUIDED_IMAGE_VARIATION_BATCH_PARAMS
							snake_case_       =							IMAGE_TO_IMAGE_IMAGE_PARAMS.union({'control_image'}						)
							snake_case_       =							IMAGE_TO_IMAGE_IMAGE_PARAMS


							def   _UpperCamelCase    (     self :  Tuple							):
													'''simple docstring'''





													torch.manual_seed(0							)
													A__  :					Optional[int]											=					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     ,      )
													torch.manual_seed(0							)
													A__  :					Optional[int]											=					ControlNetModel(
													    block_out_channels=(32, 64)     ,      layers_per_block=2     ,      in_channels=4     ,      down_block_types=("""DownBlock2D""", """CrossAttnDownBlock2D""")     ,      cross_attention_dim=32     ,      conditioning_embedding_out_channels=(16, 32)     ,      )
													torch.manual_seed(0							)
													A__  :					List[Any]											=					DDIMScheduler(
													    beta_start=0.00085     ,      beta_end=0.012     ,      beta_schedule="""scaled_linear"""     ,      clip_sample=snake_case     ,      set_alpha_to_one=snake_case     ,      )
													torch.manual_seed(0							)
													A__  :					List[str]											=					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							)
													A__  :					Optional[Any]											=					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=1000     ,      )
													A__  :					Optional[int]											=					CLIPTextModel(snake_case							)
													A__  :					Dict											=					CLIPTokenizer.from_pretrained("""hf-internal-testing/tiny-random-clip"""							)

													A__  :					Optional[Any]											=					{
													    """unet""": unet,
													    """controlnet""": controlnet,
													    """scheduler""": scheduler,
													    """vae""": vae,
													    """text_encoder""": text_encoder,
													    """tokenizer""": tokenizer,
													    """safety_checker""": None,
													    """feature_extractor""": None,
													}
													return components


							def   _UpperCamelCase    (     self :  Optional[int]     ,      snake_case :  List[str]     ,      snake_case :  Tuple=0							):
													'''simple docstring'''





													if str(snake_case							).startswith("""mps"""							):
																			A__  :					List[str]											=					torch.manual_seed(snake_case							)
													else:
																			A__  :					Optional[Any]											=					torch.Generator(device=snake_case							).manual_seed(snake_case							)

													A__  :					Tuple											=					2
													A__  :					List[str]											=					randn_tensor(
													    (1, 3, 32 * controlnet_embedder_scale_factor, 32 * controlnet_embedder_scale_factor)     ,      generator=snake_case     ,      device=torch.device(snake_case							)     ,      )
													A__  :					Dict											=					floats_tensor(control_image.shape     ,      rng=random.Random(snake_case							)							).to(snake_case							)
													A__  :					Any											=					image.cpu().permute(0     ,      2     ,      3     ,      1							)[0]
													A__  :					int											=					Image.fromarray(np.uinta(snake_case							)							).convert("""RGB"""							).resize((64, 64)							)
													A__  :					Any											=					{
													    """prompt""": """A painting of a squirrel eating a burger""",
													    """generator""": generator,
													    """num_inference_steps""": 2,
													    """guidance_scale""": 6.0,
													    """output_type""": """numpy""",
													    """image""": image,
													    """control_image""": control_image,
													}

													return inputs


							def   _UpperCamelCase    (     self :  Union[str, Any]							):
													'''simple docstring'''





													return self._test_attention_slicing_forward_pass(expected_max_diff=2e-3							)


							@unittest.skipIf(
							    torch_device != """cuda""" or not is_xformers_available()     ,      reason="""XFormers attention is only available with CUDA and `xformers` installed"""     ,      )
							def   _UpperCamelCase    (     self :  Dict							):
													'''simple docstring'''





													self._test_xformers_attention_forwardGenerator_pass(expected_max_diff=2e-3							)


							def   _UpperCamelCase    (     self :  List[str]							):
													'''simple docstring'''





													self._test_inference_batch_single_identical(expected_max_diff=2e-3							)






class 							__SCREAMING_SNAKE_CASE  (			UpperCamelCase    ,						UpperCamelCase    ,						unittest.TestCase						):
							snake_case_       =							StableDiffusionControlNetImgaImgPipeline
							snake_case_       =							TEXT_GUIDED_IMAGE_VARIATION_PARAMS - {'height', 'width'}
							snake_case_       =							TEXT_GUIDED_IMAGE_VARIATION_BATCH_PARAMS
							snake_case_       =							frozenset([]						)  # TO_DO: add image_params once refactored VaeImageProcessor.preprocess


							def   _UpperCamelCase    (     self :  Any							):
													'''simple docstring'''





													torch.manual_seed(0							)
													A__  :					Tuple											=					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     ,      )
													torch.manual_seed(0							)

													def init_weights(snake_case :  str							):
																			if isinstance(snake_case     ,      torch.nn.Convad							):
																									torch.nn.init.normal(m.weight							)
																									m.bias.data.fill_(1.0							)

													A__  :					Union[str, Any]											=					ControlNetModel(
													    block_out_channels=(32, 64)     ,      layers_per_block=2     ,      in_channels=4     ,      down_block_types=("""DownBlock2D""", """CrossAttnDownBlock2D""")     ,      cross_attention_dim=32     ,      conditioning_embedding_out_channels=(16, 32)     ,      )
													controlneta.controlnet_down_blocks.apply(snake_case							)

													torch.manual_seed(0							)
													A__  :					Union[str, Any]											=					ControlNetModel(
													    block_out_channels=(32, 64)     ,      layers_per_block=2     ,      in_channels=4     ,      down_block_types=("""DownBlock2D""", """CrossAttnDownBlock2D""")     ,      cross_attention_dim=32     ,      conditioning_embedding_out_channels=(16, 32)     ,      )
													controlneta.controlnet_down_blocks.apply(snake_case							)

													torch.manual_seed(0							)
													A__  :					Tuple											=					DDIMScheduler(
													    beta_start=0.00085     ,      beta_end=0.012     ,      beta_schedule="""scaled_linear"""     ,      clip_sample=snake_case     ,      set_alpha_to_one=snake_case     ,      )
													torch.manual_seed(0							)
													A__  :					Optional[Any]											=					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							)
													A__  :					str											=					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=1000     ,      )
													A__  :					int											=					CLIPTextModel(snake_case							)
													A__  :					Tuple											=					CLIPTokenizer.from_pretrained("""hf-internal-testing/tiny-random-clip"""							)

													A__  :					Tuple											=					MultiControlNetModel([controlneta, controlneta]							)

													A__  :					List[str]											=					{
													    """unet""": unet,
													    """controlnet""": controlnet,
													    """scheduler""": scheduler,
													    """vae""": vae,
													    """text_encoder""": text_encoder,
													    """tokenizer""": tokenizer,
													    """safety_checker""": None,
													    """feature_extractor""": None,
													}
													return components


							def   _UpperCamelCase    (     self :  List[Any]     ,      snake_case :  str     ,      snake_case :  Any=0							):
													'''simple docstring'''





													if str(snake_case							).startswith("""mps"""							):
																			A__  :					Optional[int]											=					torch.manual_seed(snake_case							)
													else:
																			A__  :					List[Any]											=					torch.Generator(device=snake_case							).manual_seed(snake_case							)

													A__  :					Optional[int]											=					2

													A__  :					Optional[int]											=					[
													    randn_tensor(
													        (1, 3, 32 * controlnet_embedder_scale_factor, 32 * controlnet_embedder_scale_factor)     ,      generator=snake_case     ,      device=torch.device(snake_case							)     ,      ),
													    randn_tensor(
													        (1, 3, 32 * controlnet_embedder_scale_factor, 32 * controlnet_embedder_scale_factor)     ,      generator=snake_case     ,      device=torch.device(snake_case							)     ,      ),
													]

													A__  :					Optional[int]											=					floats_tensor(control_image[0].shape     ,      rng=random.Random(snake_case							)							).to(snake_case							)
													A__  :					Tuple											=					image.cpu().permute(0     ,      2     ,      3     ,      1							)[0]
													A__  :					str											=					Image.fromarray(np.uinta(snake_case							)							).convert("""RGB"""							).resize((64, 64)							)
													A__  :					Optional[Any]											=					{
													    """prompt""": """A painting of a squirrel eating a burger""",
													    """generator""": generator,
													    """num_inference_steps""": 2,
													    """guidance_scale""": 6.0,
													    """output_type""": """numpy""",
													    """image""": image,
													    """control_image""": control_image,
													}

													return inputs


							def   _UpperCamelCase    (     self :  Optional[int]							):
													'''simple docstring'''





													A__  :					str											=					self.get_dummy_components()
													A__  :					Optional[Any]											=					self.pipeline_class(**snake_case							)
													pipe.to(snake_case							)

													A__  :					str											=					10.0
													A__  :					Dict											=					4

													A__  :					Any											=					self.get_dummy_inputs(snake_case							)
													A__  :					str											=					steps
													A__  :					List[Any]											=					scale
													A__  :					int											=					pipe(**snake_case							)[0]

													A__  :					Optional[int]											=					self.get_dummy_inputs(snake_case							)
													A__  :					Optional[Any]											=					steps
													A__  :					int											=					scale
													A__  :					Optional[int]											=					pipe(**snake_case     ,      control_guidance_start=0.1     ,      control_guidance_end=0.2							)[0]

													A__  :					str											=					self.get_dummy_inputs(snake_case							)
													A__  :					Dict											=					steps
													A__  :					Dict											=					scale
													A__  :					int											=					pipe(**snake_case     ,      control_guidance_start=[0.1, 0.3]     ,      control_guidance_end=[0.2, 0.7]							)[0]

													A__  :					str											=					self.get_dummy_inputs(snake_case							)
													A__  :					List[Any]											=					steps
													A__  :					Optional[int]											=					scale
													A__  :					Any											=					pipe(**snake_case     ,      control_guidance_start=0.4     ,      control_guidance_end=[0.5, 0.8]							)[0]

													# make sure that all outputs are different
													assert np.sum(np.abs(output_a - output_a							)							) > 1e-3
													assert np.sum(np.abs(output_a - output_a							)							) > 1e-3
													assert np.sum(np.abs(output_a - output_a							)							) > 1e-3


							def   _UpperCamelCase    (     self :  int							):
													'''simple docstring'''





													return self._test_attention_slicing_forward_pass(expected_max_diff=2e-3							)


							@unittest.skipIf(
							    torch_device != """cuda""" or not is_xformers_available()     ,      reason="""XFormers attention is only available with CUDA and `xformers` installed"""     ,      )
							def   _UpperCamelCase    (     self :  str							):
													'''simple docstring'''





													self._test_xformers_attention_forwardGenerator_pass(expected_max_diff=2e-3							)


							def   _UpperCamelCase    (     self :  Union[str, Any]							):
													'''simple docstring'''





													self._test_inference_batch_single_identical(expected_max_diff=2e-3							)


							def   _UpperCamelCase    (     self :  str							):
													'''simple docstring'''





													A__  :					List[Any]											=					self.get_dummy_components()
													A__  :					Any											=					self.pipeline_class(**snake_case							)
													pipe.to(snake_case							)
													pipe.set_progress_bar_config(disable=snake_case							)
													with tempfile.TemporaryDirectory() as tmpdir:
																			try:
																									# save_pretrained is not implemented for Multi-ControlNet
																									pipe.save_pretrained(snake_case							)
																			except NotImplementedError:
																									pass






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


							def   _UpperCamelCase    (     self :  int							):
													'''simple docstring'''





													super().tearDown()
													gc.collect()
													torch.cuda.empty_cache()


							def   _UpperCamelCase    (     self :  str							):
													'''simple docstring'''





													A__  :					Optional[Any]											=					ControlNetModel.from_pretrained("""lllyasviel/sd-controlnet-canny"""							)

													A__  :					Any											=					StableDiffusionControlNetImgaImgPipeline.from_pretrained(
													    """runwayml/stable-diffusion-v1-5"""     ,      safety_checker=snake_case     ,      controlnet=snake_case							)
													pipe.enable_model_cpu_offload()
													pipe.set_progress_bar_config(disable=snake_case							)

													A__  :					str											=					torch.Generator(device="""cpu"""							).manual_seed(0							)
													A__  :					Optional[Any]											=					"""evil space-punk bird"""
													A__  :					List[Any]											=					load_image(
													    """https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/sd_controlnet/bird_canny.png"""							).resize((512, 512)							)
													A__  :					Optional[int]											=					load_image(
													    """https://huggingface.co/lllyasviel/sd-controlnet-canny/resolve/main/images/bird.png"""							).resize((512, 512)							)

													A__  :					Union[str, Any]											=					pipe(
													    snake_case     ,      snake_case     ,      control_image=snake_case     ,      generator=snake_case     ,      output_type="""np"""     ,      num_inference_steps=50     ,      strength=0.6     ,      )

													A__  :					Dict											=					output.images[0]

													assert image.shape == (512, 512, 3)

													A__  :					int											=					load_numpy(
													    """https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/sd_controlnet/img2img.npy"""							)

													assert np.abs(expected_image - image							).max() < 9e-2














"""simple docstring"""



import numpy as np






class 							__SCREAMING_SNAKE_CASE  :


							def __init__(     self :  List[str]							):
													'''simple docstring'''





													A__  :					Optional[int]											=					(0, 0)
													A__  :					Dict											=					None
													A__  :					int											=					0
													A__  :					str											=					0
													A__  :					Optional[Any]											=					0


							def __eq__(     self :  str     ,      snake_case :  Optional[int]							):
													'''simple docstring'''





													return self.position == cell.position


							def   _UpperCamelCase    (     self :  List[str]							):
													'''simple docstring'''





													print(self.position							)






class 							__SCREAMING_SNAKE_CASE  :


							def __init__(     self :  int     ,      snake_case :  Any=(5, 5)							):
													'''simple docstring'''





													A__  :					Optional[int]											=					np.zeros(snake_case							)
													A__  :					List[Any]											=					world_size[0]
													A__  :					Dict											=					world_size[1]


							def   _UpperCamelCase    (     self :  Any							):
													'''simple docstring'''





													print(self.w							)


							def   _UpperCamelCase    (     self :  Optional[int]     ,      snake_case :  List[Any]							):
													'''simple docstring'''





													A__  :					int											=					[
													    (-1, -1),
													    (-1, 0),
													    (-1, 1),
													    (0, -1),
													    (0, 1),
													    (1, -1),
													    (1, 0),
													    (1, 1),
													]
													A__  :					int											=					cell.position[0]
													A__  :					str											=					cell.position[1]
													A__  :					Any											=					[]
													for n in neughbour_cord:
																			A__  :					List[Any]											=					current_x + n[0]
																			A__  :					Tuple											=					current_y + n[1]
																			if 0 <= x < self.world_x_limit and 0 <= y < self.world_y_limit:
																									A__  :					List[Any]											=					Cell()
																									A__  :					str											=					(x, y)
																									A__  :					Optional[Any]											=					cell
																									neighbours.append(snake_case							)
													return neighbours




def        _lowerCAmelCase						(							UpperCAmelCase__    :   List[str],  UpperCAmelCase__    :   Optional[Any],  UpperCAmelCase__    :   Dict )						->Dict:
						A__  :					Union[str, Any]											=					[]
						A__  :					Optional[int]											=					[]
						_open.append(UpperCAmelCase__ )

						while _open:
												A__  :					List[Any]											=					np.argmin([n.f for n in _open] )
												A__  :					Union[str, Any]											=					_open[min_f]
												_closed.append(_open.pop(UpperCAmelCase__ ) )
												if current == goal:
																		break
												for n in world.get_neigbours(UpperCAmelCase__ ):
																		for c in _closed:
																								if c == n:
																														continue
																		A__  :					Dict											=					current.g + 1
																		A__					,					A__  :					int											=					n.position
																		A__					,					A__  :					Optional[int]											=					goal.position
																		A__  :					Union[str, Any]											=					(ya - ya) ** 2 + (xa - xa) ** 2
																		A__  :					Optional[int]											=					n.h + n.g

																		for c in _open:
																								if c == n and c.f < n.f:
																														continue
																		_open.append(UpperCAmelCase__ )
						A__  :					List[str]											=					[]
						while current.parent is not None:
												path.append(current.position )
												A__  :					Union[str, Any]											=					current.parent
						path.append(current.position )
						return path[::-1]


if __name__ == "__main__":
						A_      =							Gridworld()
						# Start position and goal
						A_      =							Cell()
						A_      =							(0, 0)
						A_      =							Cell()
						A_      =							(4, 4)
						print(F'path from {start.position} to {goal.position}')
						A_      =							astar(world, start, goal)
						#   Just for visual reasons.
						for i in s:
												A_      =							1
						print(world.w)






