





"""simple docstring"""




import unittest

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

from diffusers import (
    AutoencoderKL,
    DDIMScheduler,
    DPMSolverMultistepScheduler,
    TextToVideoSDPipeline,
    UNetaDConditionModel,
)
from diffusers.utils import is_xformers_available, load_numpy, skip_mps, slow, torch_device
from diffusers.utils.testing_utils import enable_full_determinism

from ..pipeline_params import TEXT_TO_IMAGE_BATCH_PARAMS, TEXT_TO_IMAGE_PARAMS
from ..test_pipelines_common import PipelineTesterMixin


enable_full_determinism()



@skip_mps
class     _UpperCAmelCase   (   _UpperCamelCase		,   unittest.TestCase   ):

  """simple docstring"""


  a_      =			TextToVideoSDPipeline
  a_      =			TEXT_TO_IMAGE_PARAMS
  a_      =			TEXT_TO_IMAGE_BATCH_PARAMS
  # No `output_type`.
  a_      =			frozenset(
      [
          """num_inference_steps""",
          """generator""",
          """latents""",
          """return_dict""",
          """callback""",
          """callback_steps""",
      ]   )






  def     lowercase		(    self			:		Any					)							->						int:
     torch.manual_seed(0					)
     __lowerCAmelCase							     =     UNetaDConditionModel(
         block_out_channels=(3_2, 6_4, 6_4, 6_4) ,  layers_per_block=2 ,  sample_size=3_2 ,  in_channels=4 ,  out_channels=4 ,  down_block_types=('CrossAttnDownBlock3D', 'CrossAttnDownBlock3D', 'CrossAttnDownBlock3D', 'DownBlock3D') ,  up_block_types=('UpBlock3D', 'CrossAttnUpBlock3D', 'CrossAttnUpBlock3D', 'CrossAttnUpBlock3D') ,  cross_attention_dim=3_2 ,  attention_head_dim=4 ,  )
     __lowerCAmelCase							     =     DDIMScheduler(
         beta_start=0.0_00_85 ,  beta_end=0.0_12 ,  beta_schedule='scaled_linear' ,  clip_sample=lowerCAmelCase_ ,  set_alpha_to_one=lowerCAmelCase_ ,  )
     torch.manual_seed(0					)
     __lowerCAmelCase							     =     AutoencoderKL(
         block_out_channels=[3_2, 6_4] ,  in_channels=3 ,  out_channels=3 ,  down_block_types=['DownEncoderBlock2D', 'DownEncoderBlock2D'] ,  up_block_types=['UpDecoderBlock2D', 'UpDecoderBlock2D'] ,  latent_channels=4 ,  sample_size=1_2_8 ,  )
     torch.manual_seed(0					)
     __lowerCAmelCase							     =     CLIPTextConfig(
         bos_token_id=0 ,  eos_token_id=2 ,  hidden_size=3_2 ,  intermediate_size=3_7 ,  layer_norm_eps=1e-05 ,  num_attention_heads=4 ,  num_hidden_layers=5 ,  pad_token_id=1 ,  vocab_size=1_0_0_0 ,  hidden_act='gelu' ,  projection_dim=5_1_2 ,  )
     __lowerCAmelCase							     =     CLIPTextModel(lowerCAmelCase_					)
     __lowerCAmelCase							     =     CLIPTokenizer.from_pretrained('hf-internal-testing/tiny-random-clip'					)

     __lowerCAmelCase							     =     {
         'unet': unet,
         'scheduler': scheduler,
         'vae': vae,
         'text_encoder': text_encoder,
         'tokenizer': tokenizer,
     }
     return components






  def     lowercase		(    self			:		Union[str, Any] ,  lowerCAmelCase_			:		Any ,  lowerCAmelCase_			:		Any=0					)							->						Dict:
     if str(lowerCAmelCase_					).startswith('mps'					):
        __lowerCAmelCase							     =     torch.manual_seed(lowerCAmelCase_					)
     else:
        __lowerCAmelCase							     =     torch.Generator(device=lowerCAmelCase_					).manual_seed(lowerCAmelCase_					)
     __lowerCAmelCase							     =     {
         'prompt': 'A painting of a squirrel eating a burger',
         'generator': generator,
         'num_inference_steps': 2,
         'guidance_scale': 6.0,
         'output_type': 'pt',
     }
     return inputs






  def     lowercase		(    self			:		List[Any]					)							->						str:
     __lowerCAmelCase							     =     'cpu'  # ensure determinism for the device-dependent torch.Generator
     __lowerCAmelCase							     =     self.get_dummy_components()
     __lowerCAmelCase							     =     TextToVideoSDPipeline(**lowerCAmelCase_					)
     __lowerCAmelCase							     =     sd_pipe.to(lowerCAmelCase_					)
     sd_pipe.set_progress_bar_config(disable=lowerCAmelCase_					)

     __lowerCAmelCase							     =     self.get_dummy_inputs(lowerCAmelCase_					)
     __lowerCAmelCase							     =     'np'
     __lowerCAmelCase							     =     sd_pipe(**lowerCAmelCase_					).frames
     __lowerCAmelCase							     =     frames[0][-3:, -3:, -1]

     assert frames[0].shape == (6_4, 6_4, 3)
     __lowerCAmelCase							     =     np.array([1_5_8.0, 1_6_0.0, 1_5_3.0, 1_2_5.0, 1_0_0.0, 1_2_1.0, 1_1_1.0, 93.0, 1_1_3.0]					)

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






  def     lowercase		(    self			:		Union[str, Any]					)							->						Optional[Any]:
     self._test_attention_slicing_forward_pass(test_mean_pixel_difference=lowerCAmelCase_ ,  expected_max_diff=3e-3					)






  @unittest.skipIf(
      torch_device != 'cuda' or not is_xformers_available() ,  reason='XFormers attention is only available with CUDA and `xformers` installed' ,  )
  def     lowercase		(    self			:		Union[str, Any]					)							->						Any:
     self._test_xformers_attention_forwardGenerator_pass(test_mean_pixel_difference=lowerCAmelCase_ ,  expected_max_diff=1e-2					)






  @unittest.skip(reason='Batching needs to be properly figured out first for this pipeline.'					)
  def     lowercase		(    self			:		Tuple					)							->						Any:
     pass






  @unittest.skip(reason='Batching needs to be properly figured out first for this pipeline.'					)
  def     lowercase		(    self			:		Optional[int]					)							->						List[str]:
     pass






  @unittest.skip(reason='`num_images_per_prompt` argument is not supported for this pipeline.'					)
  def     lowercase		(    self			:		Union[str, Any]					)							->						Union[str, Any]:
     pass






  def     lowercase		(    self			:		List[Any]					)							->						Dict:
     return super().test_progress_bar()





@slow
@skip_mps
class     _UpperCAmelCase   (   unittest.TestCase   ):

  """simple docstring"""


  def     lowercase		(    self			:		Union[str, Any]					)							->						Dict:
     __lowerCAmelCase							     =     load_numpy(
         'https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/text_to_video/video.npy'					)

     __lowerCAmelCase							     =     TextToVideoSDPipeline.from_pretrained('damo-vilab/text-to-video-ms-1.7b'					)
     __lowerCAmelCase							     =     DPMSolverMultistepScheduler.from_config(pipe.scheduler.config					)
     __lowerCAmelCase							     =     pipe.to('cuda'					)

     __lowerCAmelCase							     =     'Spiderman is surfing'
     __lowerCAmelCase							     =     torch.Generator(device='cpu'					).manual_seed(0					)

     __lowerCAmelCase							     =     pipe(lowerCAmelCase_ ,  generator=lowerCAmelCase_ ,  num_inference_steps=2_5 ,  output_type='pt'					).frames
     __lowerCAmelCase							     =     video_frames.cpu().numpy()

     assert np.abs(expected_video - video					).mean() < 5e-2






  def     lowercase		(    self			:		Optional[int]					)							->						List[Any]:
     __lowerCAmelCase							     =     load_numpy(
         'https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/text_to_video/video_2step.npy'					)

     __lowerCAmelCase							     =     TextToVideoSDPipeline.from_pretrained('damo-vilab/text-to-video-ms-1.7b'					)
     __lowerCAmelCase							     =     pipe.to('cuda'					)

     __lowerCAmelCase							     =     'Spiderman is surfing'
     __lowerCAmelCase							     =     torch.Generator(device='cpu'					).manual_seed(0					)

     __lowerCAmelCase							     =     pipe(lowerCAmelCase_ ,  generator=lowerCAmelCase_ ,  num_inference_steps=2 ,  output_type='pt'					).frames
     __lowerCAmelCase							     =     video_frames.cpu().numpy()

     assert np.abs(expected_video - video					).mean() < 5e-2








from __future__ import annotations

import time
from math import sqrt

# 1 for manhattan, 0 for euclidean
_snake_case  :   Dict						   =  0

_snake_case  :   Dict						   =  [
    [0, 0, 0, 0, 0, 0, 0],
    [0, 1, 0, 0, 0, 0, 0],  # 0 are free path whereas 1's are obstacles
    [0, 0, 0, 0, 0, 0, 0],
    [0, 0, 1, 0, 0, 0, 0],
    [1, 0, 1, 0, 0, 0, 0],
    [0, 0, 0, 0, 0, 0, 0],
    [0, 0, 0, 0, 1, 0, 0],
]

_snake_case  :   List[str]						   =  [[-1, 0], [0, -1], [1, 0], [0, 1]]  # up, left, down, right

_snake_case  :   Tuple						   =  tuple[int, int]



class     _UpperCAmelCase   :

  """simple docstring"""


  def __init__(    self			:		Dict ,  lowerCAmelCase_			:		int ,  lowerCAmelCase_			:		int ,  lowerCAmelCase_			:		int ,  lowerCAmelCase_			:		int ,  lowerCAmelCase_			:		int ,  lowerCAmelCase_			:		Node | None ,  )							->						None:
     __lowerCAmelCase							     =     pos_x
     __lowerCAmelCase							     =     pos_y
     __lowerCAmelCase							     =     (pos_y, pos_x)
     __lowerCAmelCase							     =     goal_x
     __lowerCAmelCase							     =     goal_y
     __lowerCAmelCase							     =     g_cost
     __lowerCAmelCase							     =     parent
     __lowerCAmelCase							     =     self.calculate_heuristic()
     __lowerCAmelCase							     =     self.g_cost + self.h_cost






  def     lowercase		(    self			:		Any					)							->						float:
     __lowerCAmelCase							     =     self.pos_x - self.goal_x
     __lowerCAmelCase							     =     self.pos_y - self.goal_y
     if HEURISTIC == 1:
        return abs(lowerCAmelCase_					) + abs(lowerCAmelCase_					)
     else:
        return sqrt(dy**2 + dx**2					)






  def __lt__(    self			:		Union[str, Any] ,  lowerCAmelCase_			:		Node					)							->						bool:
     return self.f_cost < other.f_cost





class     _UpperCAmelCase   :

  """simple docstring"""


  def __init__(    self			:		List[str] ,  lowerCAmelCase_			:		TPosition ,  lowerCAmelCase_			:		TPosition					)							->						Tuple:
     __lowerCAmelCase							     =     Node(start[1] ,  start[0] ,  goal[1] ,  goal[0] ,  0 ,  lowerCAmelCase_					)
     __lowerCAmelCase							     =     Node(goal[1] ,  goal[0] ,  goal[1] ,  goal[0] ,  9_9_9_9_9 ,  lowerCAmelCase_					)

     __lowerCAmelCase							     =     [self.start]
     __lowerCAmelCase							     =     []

     __lowerCAmelCase							     =     False






  def     lowercase		(    self			:		str					)							->						list[TPosition]:
     while self.open_nodes:
        # Open Nodes are sorted using __lt__
        self.open_nodes.sort()
        __lowerCAmelCase							     =     self.open_nodes.pop(0					)

        if current_node.pos == self.target.pos:
           return self.retrace_path(lowerCAmelCase_					)

        self.closed_nodes.append(lowerCAmelCase_					)
        __lowerCAmelCase							     =     self.get_successors(lowerCAmelCase_					)

        for child_node in successors:
           if child_node in self.closed_nodes:
              continue

           if child_node not in self.open_nodes:
              self.open_nodes.append(lowerCAmelCase_					)
           else:
              # retrieve the best current path
              __lowerCAmelCase							     =     self.open_nodes.pop(self.open_nodes.index(lowerCAmelCase_					)					)

              if child_node.g_cost < better_node.g_cost:
                 self.open_nodes.append(lowerCAmelCase_					)
              else:
                 self.open_nodes.append(lowerCAmelCase_					)

     return [self.start.pos]






  def     lowercase		(    self			:		Optional[Any] ,  lowerCAmelCase_			:		Node					)							->						list[Node]:
     __lowerCAmelCase							     =     []
     for action in delta:
        __lowerCAmelCase							     =     parent.pos_x + action[1]
        __lowerCAmelCase							     =     parent.pos_y + action[0]
        if not (0 <= pos_x <= len(grid[0]					) - 1 and 0 <= pos_y <= len(lowerCAmelCase_					) - 1):
           continue

        if grid[pos_y][pos_x] != 0:
           continue

        successors.append(
            Node(
                lowerCAmelCase_ ,  lowerCAmelCase_ ,  self.target.pos_y ,  self.target.pos_x ,  parent.g_cost + 1 ,  lowerCAmelCase_ ,  )					)
     return successors






  def     lowercase		(    self			:		Tuple ,  lowerCAmelCase_			:		Node | None					)							->						list[TPosition]:
     __lowerCAmelCase							     =     node
     __lowerCAmelCase							     =     []
     while current_node is not None:
        path.append((current_node.pos_y, current_node.pos_x)					)
        __lowerCAmelCase							     =     current_node.parent
     path.reverse()
     return path





class     _UpperCAmelCase   :

  """simple docstring"""


  def __init__(    self			:		int ,  lowerCAmelCase_			:		TPosition ,  lowerCAmelCase_			:		TPosition					)							->						None:
     __lowerCAmelCase							     =     AStar(lowerCAmelCase_ ,  lowerCAmelCase_					)
     __lowerCAmelCase							     =     AStar(lowerCAmelCase_ ,  lowerCAmelCase_					)
     __lowerCAmelCase							     =     False






  def     lowercase		(    self			:		Dict					)							->						list[TPosition]:
     while self.fwd_astar.open_nodes or self.bwd_astar.open_nodes:
        self.fwd_astar.open_nodes.sort()
        self.bwd_astar.open_nodes.sort()
        __lowerCAmelCase							     =     self.fwd_astar.open_nodes.pop(0					)
        __lowerCAmelCase							     =     self.bwd_astar.open_nodes.pop(0					)

        if current_bwd_node.pos == current_fwd_node.pos:
           return self.retrace_bidirectional_path(
               lowerCAmelCase_ ,  lowerCAmelCase_					)

        self.fwd_astar.closed_nodes.append(lowerCAmelCase_					)
        self.bwd_astar.closed_nodes.append(lowerCAmelCase_					)

        __lowerCAmelCase							     =     current_bwd_node
        __lowerCAmelCase							     =     current_fwd_node

        __lowerCAmelCase							     =     {
            self.fwd_astar: self.fwd_astar.get_successors(lowerCAmelCase_					),
            self.bwd_astar: self.bwd_astar.get_successors(lowerCAmelCase_					),
        }

        for astar in [self.fwd_astar, self.bwd_astar]:
           for child_node in successors[astar]:
              if child_node in astar.closed_nodes:
                 continue

              if child_node not in astar.open_nodes:
                 astar.open_nodes.append(lowerCAmelCase_					)
              else:
                 # retrieve the best current path
                 __lowerCAmelCase							     =     astar.open_nodes.pop(
                     astar.open_nodes.index(lowerCAmelCase_					)					)

                 if child_node.g_cost < better_node.g_cost:
                    astar.open_nodes.append(lowerCAmelCase_					)
                 else:
                    astar.open_nodes.append(lowerCAmelCase_					)

     return [self.fwd_astar.start.pos]






  def     lowercase		(    self			:		Dict ,  lowerCAmelCase_			:		Node ,  lowerCAmelCase_			:		Node					)							->						list[TPosition]:
     __lowerCAmelCase							     =     self.fwd_astar.retrace_path(lowerCAmelCase_					)
     __lowerCAmelCase							     =     self.bwd_astar.retrace_path(lowerCAmelCase_					)
     bwd_path.pop()
     bwd_path.reverse()
     __lowerCAmelCase							     =     fwd_path + bwd_path
     return path


if __name__ == "__main__":
     # all coordinates are given in format [y,x]
     _snake_case  :   List[Any]						   =  (0, 0)
     _snake_case  :   Union[str, Any]						   =  (len(grid) - 1, len(grid[0]) - 1)
     for elem in grid:
          print(elem)

     _snake_case  :   int						   =  time.time()
     _snake_case  :   Optional[int]						   =  AStar(init, goal)
     _snake_case  :   int						   =  a_star.search()
     _snake_case  :   Union[str, Any]						   =  time.time() - start_time
     print(F"""AStar execution time = {end_time:f} seconds""")

     _snake_case  :   Any						   =  time.time()
     _snake_case  :   Dict						   =  BidirectionalAStar(init, goal)
     _snake_case  :   Optional[int]						   =  time.time() - bd_start_time
     print(F"""BidirectionalAStar execution time = {bd_end_time:f} seconds""")

