



import gc
import unittest

import numpy as np
import torch
from torch.backends.cuda import sdp_kernel

from diffusers import (
    CMStochasticIterativeScheduler,
    ConsistencyModelPipeline,
    UNetaDModel,
)
from diffusers.utils import randn_tensor, slow, torch_device
from diffusers.utils.testing_utils import enable_full_determinism, require_torch_a, require_torch_gpu

from ..pipeline_params import UNCONDITIONAL_IMAGE_GENERATION_BATCH_PARAMS, UNCONDITIONAL_IMAGE_GENERATION_PARAMS
from ..test_pipelines_common import PipelineTesterMixin


enable_full_determinism()

class 				snake_case__(A__    ,      unittest.TestCase      ):

					"""simple docstring"""






					lowercase_								=		ConsistencyModelPipeline
					lowercase_								=		UNCONDITIONAL_IMAGE_GENERATION_PARAMS
					lowercase_								=		UNCONDITIONAL_IMAGE_GENERATION_BATCH_PARAMS

					# Override required_optional_params to remove num_images_per_prompt
					lowercase_								=		frozenset(
					    [
					        """num_inference_steps""",
					        """generator""",
					        """latents""",
					        """output_type""",
					        """return_dict""",
					        """callback""",
					        """callback_steps""",
					    ]      )







					@property
					def 			snake_case					(							self :     str					):
									lowercase__					:   int													=					UNetaDModel.from_pretrained(
									    "diffusers/consistency-models-test"				,       subfolder="test_unet"				,       )
									return unet







					@property
					def 			snake_case					(							self :     str					):
									lowercase__					:   Optional[int]													=					UNetaDModel.from_pretrained(
									    "diffusers/consistency-models-test"				,       subfolder="test_unet_class_cond"				,       )
									return unet







					def 			snake_case					(							self :     Optional[int]				,       SCREAMING_SNAKE_CASE :     Any=False					):
									if class_cond:
													lowercase__					:   int													=					self.dummy_cond_unet
									else:
													lowercase__					:   int													=					self.dummy_uncond_unet

									# Default to CM multistep sampler
									lowercase__					:   Dict													=					CMStochasticIterativeScheduler(
									    num_train_timesteps=40				,       sigma_min=0.002				,       sigma_max=80.0				,       )

									lowercase__					:   str													=					{
									    """unet""": unet,
									    """scheduler""": scheduler,
									}

									return components







					def 			snake_case					(							self :     Optional[int]				,       SCREAMING_SNAKE_CASE :     Union[str, Any]				,       SCREAMING_SNAKE_CASE :     int=0					):
									if str(__A					).startswith("mps"					):
													lowercase__					:   Union[str, Any]													=					torch.manual_seed(__A					)
									else:
													lowercase__					:   Any													=					torch.Generator(device=__A					).manual_seed(__A					)

									lowercase__					:   List[str]													=					{
									    """batch_size""": 1,
									    """num_inference_steps""": None,
									    """timesteps""": [22, 0],
									    """generator""": generator,
									    """output_type""": """np""",
									}

									return inputs







					def 			snake_case					(							self :     Optional[Any]					):
									lowercase__					:   str													=					"""cpu"""  # ensure determinism for the device-dependent torch.Generator
									lowercase__					:   List[Any]													=					self.get_dummy_components()
									lowercase__					:   int													=					ConsistencyModelPipeline(**__A					)
									lowercase__					:   Any													=					pipe.to(__A					)
									pipe.set_progress_bar_config(disable=__A					)

									lowercase__					:   List[Any]													=					self.get_dummy_inputs(__A					)
									lowercase__					:   Optional[int]													=					pipe(**__A					).images
									assert image.shape == (1, 32, 32, 3)

									lowercase__					:   Optional[int]													=					image[0, -3:, -3:, -1]
									lowercase__					:   str													=					np.array([0.3_572, 0.6_273, 0.4_031, 0.3_961, 0.4_321, 0.5_730, 0.5_266, 0.4_780, 0.5_004]					)

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







					def 			snake_case					(							self :     Union[str, Any]					):
									lowercase__					:   Union[str, Any]													=					"""cpu"""  # ensure determinism for the device-dependent torch.Generator
									lowercase__					:   Optional[int]													=					self.get_dummy_components(class_cond=__A					)
									lowercase__					:   Optional[Any]													=					ConsistencyModelPipeline(**__A					)
									lowercase__					:   Tuple													=					pipe.to(__A					)
									pipe.set_progress_bar_config(disable=__A					)

									lowercase__					:   Any													=					self.get_dummy_inputs(__A					)
									lowercase__					:   int													=					0
									lowercase__					:   Optional[Any]													=					pipe(**__A					).images
									assert image.shape == (1, 32, 32, 3)

									lowercase__					:   Tuple													=					image[0, -3:, -3:, -1]
									lowercase__					:   List[Any]													=					np.array([0.3_572, 0.6_273, 0.4_031, 0.3_961, 0.4_321, 0.5_730, 0.5_266, 0.4_780, 0.5_004]					)

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







					def 			snake_case					(							self :     Optional[int]					):
									lowercase__					:   Union[str, Any]													=					"""cpu"""  # ensure determinism for the device-dependent torch.Generator
									lowercase__					:   Any													=					self.get_dummy_components()
									lowercase__					:   str													=					ConsistencyModelPipeline(**__A					)
									lowercase__					:   Optional[int]													=					pipe.to(__A					)
									pipe.set_progress_bar_config(disable=__A					)

									lowercase__					:   List[Any]													=					self.get_dummy_inputs(__A					)
									lowercase__					:   Optional[Any]													=					1
									lowercase__					:   Any													=					None
									lowercase__					:   str													=					pipe(**__A					).images
									assert image.shape == (1, 32, 32, 3)

									lowercase__					:   Dict													=					image[0, -3:, -3:, -1]
									lowercase__					:   Dict													=					np.array([0.5_004, 0.5_004, 0.4_994, 0.5_008, 0.4_976, 0.5_018, 0.4_990, 0.4_982, 0.4_987]					)

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







					def 			snake_case					(							self :     Union[str, Any]					):
									lowercase__					:   Dict													=					"""cpu"""  # ensure determinism for the device-dependent torch.Generator
									lowercase__					:   List[Any]													=					self.get_dummy_components(class_cond=__A					)
									lowercase__					:   Union[str, Any]													=					ConsistencyModelPipeline(**__A					)
									lowercase__					:   Union[str, Any]													=					pipe.to(__A					)
									pipe.set_progress_bar_config(disable=__A					)

									lowercase__					:   Optional[Any]													=					self.get_dummy_inputs(__A					)
									lowercase__					:   Tuple													=					1
									lowercase__					:   Dict													=					None
									lowercase__					:   int													=					0
									lowercase__					:   Optional[Any]													=					pipe(**__A					).images
									assert image.shape == (1, 32, 32, 3)

									lowercase__					:   List[str]													=					image[0, -3:, -3:, -1]
									lowercase__					:   Any													=					np.array([0.5_004, 0.5_004, 0.4_994, 0.5_008, 0.4_976, 0.5_018, 0.4_990, 0.4_982, 0.4_987]					)

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






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

					"""simple docstring"""






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







					def 			snake_case					(							self :     Tuple				,       SCREAMING_SNAKE_CASE :     List[str]=0				,       SCREAMING_SNAKE_CASE :     Optional[int]=False				,       SCREAMING_SNAKE_CASE :     Dict="cpu"				,       SCREAMING_SNAKE_CASE :     int=torch.floataa				,       SCREAMING_SNAKE_CASE :     List[str]=(1, 3, 64, 64)					):
									lowercase__					:   Any													=					torch.manual_seed(__A					)

									lowercase__					:   Optional[int]													=					{
									    """num_inference_steps""": None,
									    """timesteps""": [22, 0],
									    """class_labels""": 0,
									    """generator""": generator,
									    """output_type""": """np""",
									}

									if get_fixed_latents:
													lowercase__					:   Dict													=					self.get_fixed_latents(seed=__A				,       device=__A				,       dtype=__A				,       shape=__A					)
													lowercase__					:   Tuple													=					latents

									return inputs







					def 			snake_case					(							self :     str				,       SCREAMING_SNAKE_CASE :     str=0				,       SCREAMING_SNAKE_CASE :     Optional[Any]="cpu"				,       SCREAMING_SNAKE_CASE :     List[str]=torch.floataa				,       SCREAMING_SNAKE_CASE :     str=(1, 3, 64, 64)					):
									if type(__A					) == str:
													lowercase__					:   int													=					torch.device(__A					)
									lowercase__					:   str													=					torch.Generator(device=__A					).manual_seed(__A					)
									lowercase__					:   Union[str, Any]													=					randn_tensor(__A				,       generator=__A				,       device=__A				,       dtype=__A					)
									return latents







					def 			snake_case					(							self :     str					):
									lowercase__					:   Optional[Any]													=					UNetaDModel.from_pretrained("diffusers/consistency_models"				,       subfolder="diffusers_cd_imagenet64_l2"					)
									lowercase__					:   int													=					CMStochasticIterativeScheduler(
									    num_train_timesteps=40				,       sigma_min=0.002				,       sigma_max=80.0				,       )
									lowercase__					:   Optional[int]													=					ConsistencyModelPipeline(unet=__A				,       scheduler=__A					)
									pipe.to(torch_device=__A					)
									pipe.set_progress_bar_config(disable=__A					)

									lowercase__					:   Optional[Any]													=					self.get_inputs()
									lowercase__					:   Union[str, Any]													=					pipe(**__A					).images
									assert image.shape == (1, 64, 64, 3)

									lowercase__					:   Any													=					image[0, -3:, -3:, -1]

									lowercase__					:   Union[str, Any]													=					np.array([0.0_888, 0.0_881, 0.0_666, 0.0_479, 0.0_292, 0.0_195, 0.0_201, 0.0_163, 0.0_254]					)

									assert np.abs(image_slice.flatten() - expected_slice					).max() < 2E-2







					def 			snake_case					(							self :     Dict					):
									lowercase__					:   Dict													=					UNetaDModel.from_pretrained("diffusers/consistency_models"				,       subfolder="diffusers_cd_imagenet64_l2"					)
									lowercase__					:   Tuple													=					CMStochasticIterativeScheduler(
									    num_train_timesteps=40				,       sigma_min=0.002				,       sigma_max=80.0				,       )
									lowercase__					:   List[str]													=					ConsistencyModelPipeline(unet=__A				,       scheduler=__A					)
									pipe.to(torch_device=__A					)
									pipe.set_progress_bar_config(disable=__A					)

									lowercase__					:   str													=					self.get_inputs()
									lowercase__					:   List[str]													=					1
									lowercase__					:   Optional[int]													=					None
									lowercase__					:   str													=					pipe(**__A					).images
									assert image.shape == (1, 64, 64, 3)

									lowercase__					:   str													=					image[0, -3:, -3:, -1]

									lowercase__					:   Union[str, Any]													=					np.array([0.0_340, 0.0_152, 0.0_063, 0.0_267, 0.0_221, 0.0_107, 0.0_416, 0.0_186, 0.0_217]					)

									assert np.abs(image_slice.flatten() - expected_slice					).max() < 2E-2







					@require_torch_a
					def 			snake_case					(							self :     str					):
									lowercase__					:   int													=					UNetaDModel.from_pretrained("diffusers/consistency_models"				,       subfolder="diffusers_cd_imagenet64_l2"					)
									lowercase__					:   List[str]													=					CMStochasticIterativeScheduler(
									    num_train_timesteps=40				,       sigma_min=0.002				,       sigma_max=80.0				,       )
									lowercase__					:   Dict													=					ConsistencyModelPipeline(unet=__A				,       scheduler=__A					)
									pipe.to(torch_device=__A				,       torch_dtype=torch.floataa					)
									pipe.set_progress_bar_config(disable=__A					)

									lowercase__					:   List[Any]													=					self.get_inputs(get_fixed_latents=__A				,       device=__A					)
									# Ensure usage of flash attention in torch 2.0
									with sdp_kernel(enable_flash=__A				,       enable_math=__A				,       enable_mem_efficient=__A					):
													lowercase__					:   Tuple													=					pipe(**__A					).images
									assert image.shape == (1, 64, 64, 3)

									lowercase__					:   Union[str, Any]													=					image[0, -3:, -3:, -1]

									lowercase__					:   Optional[int]													=					np.array([0.1_875, 0.1_428, 0.1_289, 0.2_151, 0.2_092, 0.1_477, 0.1_877, 0.1_641, 0.1_353]					)

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







					@require_torch_a
					def 			snake_case					(							self :     str					):
									lowercase__					:   List[Any]													=					UNetaDModel.from_pretrained("diffusers/consistency_models"				,       subfolder="diffusers_cd_imagenet64_l2"					)
									lowercase__					:   Union[str, Any]													=					CMStochasticIterativeScheduler(
									    num_train_timesteps=40				,       sigma_min=0.002				,       sigma_max=80.0				,       )
									lowercase__					:   Dict													=					ConsistencyModelPipeline(unet=__A				,       scheduler=__A					)
									pipe.to(torch_device=__A				,       torch_dtype=torch.floataa					)
									pipe.set_progress_bar_config(disable=__A					)

									lowercase__					:   Dict													=					self.get_inputs(get_fixed_latents=__A				,       device=__A					)
									lowercase__					:   Any													=					1
									lowercase__					:   List[Any]													=					None
									# Ensure usage of flash attention in torch 2.0
									with sdp_kernel(enable_flash=__A				,       enable_math=__A				,       enable_mem_efficient=__A					):
													lowercase__					:   Any													=					pipe(**__A					).images
									assert image.shape == (1, 64, 64, 3)

									lowercase__					:   Union[str, Any]													=					image[0, -3:, -3:, -1]

									lowercase__					:   List[str]													=					np.array([0.1_663, 0.1_948, 0.2_275, 0.1_680, 0.1_204, 0.1_245, 0.1_858, 0.1_338, 0.2_095]					)

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











from __future__ import annotations

lowerCAmelCase__					  =  [
    [-1, 0],  # left
    [0, -1],  # down
    [1, 0],  # right
    [0, 1],  # up
]


def  __lowerCamelCase (       lowerCamelCase__					,			lowerCamelCase__					,			lowerCamelCase__					,			lowerCamelCase__					,			lowerCamelCase__					,			):

				"""simple docstring"""


				lowercase__					:   Optional[int]													=					[
				    [0 for col in range(len(grid[0]						)						)] for row in range(len(lowerCamelCase__						)						)
				]  # the reference grid
				lowercase__					:   List[Any]													=					1
				lowercase__					:   Tuple													=					[
				    [0 for col in range(len(grid[0]						)						)] for row in range(len(lowerCamelCase__						)						)
				]  # the action grid

				lowercase__					:   Union[str, Any]													=					init[0]
				lowercase__					:   List[str]													=					init[1]
				lowercase__					:   Optional[Any]													=					0
				lowercase__					:   Optional[int]													=					g + heuristic[x][y]  # cost from starting cell to destination cell
				lowercase__					:   Tuple													=					[[f, g, x, y]]

				lowercase__					:   Union[str, Any]													=					False  # flag that is set when search is complete
				lowercase__					:   Any													=					False  # flag set if we can't find expand

				while not found and not resign:
								if len(lowerCamelCase__						) == 0:
												raise ValueError("Algorithm is unable to find solution"						)
								else:  # to choose the least costliest action so as to move closer to the goal
												cell.sort()
												cell.reverse()
												lowercase__					:   Tuple													=					cell.pop()
												lowercase__					:   Optional[Any]													=					next_cell[2]
												lowercase__					:   int													=					next_cell[3]
												lowercase__					:   Union[str, Any]													=					next_cell[1]

												if x == goal[0] and y == goal[1]:
																lowercase__					:   Tuple													=					True
												else:
																for i in range(len(lowerCamelCase__						)						):  # to try out different valid actions
																				lowercase__					:   Tuple													=					x + DIRECTIONS[i][0]
																				lowercase__					:   str													=					y + DIRECTIONS[i][1]
																				if xa >= 0 and xa < len(lowerCamelCase__						) and ya >= 0 and ya < len(grid[0]						):
																								if closed[xa][ya] == 0 and grid[xa][ya] == 0:
																												lowercase__					:   List[Any]													=					g + cost
																												lowercase__					:   Tuple													=					ga + heuristic[xa][ya]
																												cell.append([fa, ga, xa, ya]						)
																												lowercase__					:   Dict													=					1
																												lowercase__					:   Union[str, Any]													=					i
				lowercase__					:   Optional[int]													=					[]
				lowercase__					:   List[Any]													=					goal[0]
				lowercase__					:   Optional[int]													=					goal[1]
				invpath.append([x, y]						)  # we get the reverse path from here
				while x != init[0] or y != init[1]:
								lowercase__					:   int													=					x - DIRECTIONS[action[x][y]][0]
								lowercase__					:   List[Any]													=					y - DIRECTIONS[action[x][y]][1]
								lowercase__					:   Optional[Any]													=					xa
								lowercase__					:   Dict													=					ya
								invpath.append([x, y]						)

				lowercase__					:   List[str]													=					[]
				for i in range(len(lowerCamelCase__						)						):
								path.append(invpath[len(lowerCamelCase__						) - 1 - i]						)
				return path, action


if __name__ == "__main__":
				lowerCAmelCase__					  =  [
				    [0, 1, 0, 0, 0, 0],
				    [0, 1, 0, 0, 0, 0],  # 0 are free path whereas 1's are obstacles
				    [0, 1, 0, 0, 0, 0],
				    [0, 1, 0, 0, 1, 0],
				    [0, 0, 0, 0, 1, 0],
				]

				lowerCAmelCase__					  =  [0, 0]
				# all coordinates are given in format [y,x]
				lowerCAmelCase__					  =  [len(grid) - 1, len(grid[0]) - 1]
				lowerCAmelCase__					  =  1

				# the cost map which pushes the path closer to the goal
				lowerCAmelCase__					  =  [[0 for row in range(len(grid[0]))] for col in range(len(grid))]
				for i in range(len(grid)):
								for j in range(len(grid[0])):
												lowerCAmelCase__					  =  abs(i - goal[0]) + abs(j - goal[1])
												if grid[i][j] == 1:
																# added extra penalty in the heuristic map
																lowerCAmelCase__					  =  9_9

				lowerCAmelCase__  ,						lowerCAmelCase__					  =  search(grid, init, goal, cost, heuristic)

				print('''ACTION MAP''')
				for i in range(len(action)):
								print(action[i])

				for i in range(len(path)):
								print(path[i])




