

from typing import SupportsIndex

import numpy as np
from scipy.ndimage import convolve





def    _snake_case		(snake_case_				:    np.ndarray						,					snake_case_				:    np.ndarray						,					snake_case_				:    np.ndarray     )       ->		np.ndarray:
  __UpperCAmelCase  :     Any		=    np.stack((horizontal_flow, vertical_flow)						,					2     )

  # Create a grid of all pixel coordinates and subtract the flow to get the
  # target pixels coordinates
  __UpperCAmelCase  :     List[Any]		=    np.stack(
      np.meshgrid(np.arange(0						,					image.shape[1]     )						,					np.arange(0						,					image.shape[0]     )     )						,					2     )
  __UpperCAmelCase  :     List[Any]		=    np.round(grid - flow     ).astype(np.intaa     )

  # Find the locations outside of the original image
  __UpperCAmelCase  :     int		=    (grid < 0) | (grid >= np.array([image.shape[1], image.shape[0]]     ))
  __UpperCAmelCase  :     List[Any]		=    0

  __UpperCAmelCase  :     Union[str, Any]		=    image[grid[:, :, 1], grid[:, :, 0]]

  # Set pixels at invalid locations to 0
  __UpperCAmelCase  :     Dict		=    0

  return warped





def    _snake_case		(snake_case_				:    np.ndarray						,					snake_case_				:    np.ndarray						,					snake_case_				:    SupportsIndex						,					snake_case_				:    float | None = None						,					)       ->		tuple[np.ndarray, np.ndarray]:
  if alpha is None:
    __UpperCAmelCase  :     Tuple		=    0.1

  # Initialize flow
  __UpperCAmelCase  :     Any		=    np.zeros_like(snake_case_     )
  __UpperCAmelCase  :     str		=    np.zeros_like(snake_case_     )

  # Prepare kernels for the calculation of the derivatives and the average velocity
  __UpperCAmelCase  :     Optional[Any]		=    np.array([[-1, 1], [-1, 1]]     ) * 0.25
  __UpperCAmelCase  :     int		=    np.array([[-1, -1], [1, 1]]     ) * 0.25
  __UpperCAmelCase  :     List[str]		=    np.array([[1, 1], [1, 1]]     ) * 0.25
  __UpperCAmelCase  :     Dict		=    np.array(
      [[1 / 12, 1 / 6, 1 / 12], [1 / 6, 0, 1 / 6], [1 / 12, 1 / 6, 1 / 12]]     )

  # Iteratively refine the flow
  for _ in range(snake_case_     ):
    __UpperCAmelCase  :     Optional[int]		=    warp(snake_case_						,					snake_case_						,					snake_case_     )
    __UpperCAmelCase  :     Optional[int]		=    convolve(snake_case_						,					snake_case_     ) + convolve(snake_case_						,					snake_case_     )
    __UpperCAmelCase  :     Any		=    convolve(snake_case_						,					snake_case_     ) + convolve(snake_case_						,					snake_case_     )
    __UpperCAmelCase  :     Dict		=    convolve(snake_case_						,					snake_case_     ) + convolve(snake_case_						,					-kernel_t     )

    __UpperCAmelCase  :     int		=    convolve(snake_case_						,					snake_case_     )
    __UpperCAmelCase  :     int		=    convolve(snake_case_						,					snake_case_     )

    # This updates the flow as proposed in the paper (Step 12)
    __UpperCAmelCase  :     Optional[Any]		=    (
        derivative_x * avg_horizontal_velocity
        + derivative_y * avg_vertical_velocity
        + derivative_t
    )
    __UpperCAmelCase  :     str		=    update / (alpha**2 + derivative_x**2 + derivative_y**2)

    __UpperCAmelCase  :     Optional[int]		=    avg_horizontal_velocity - derivative_x * update
    __UpperCAmelCase  :     List[Any]		=    avg_vertical_velocity - derivative_y * update

  return horizontal_flow, vertical_flow


if __name__ == "__main__":
      import doctest

      doctest.testmod()








import gc
import tempfile
import unittest

import numpy as np
import torch

from diffusers import VersatileDiffusionDualGuidedPipeline
from diffusers.utils.testing_utils import load_image, nightly, require_torch_gpu, torch_device


_SCREAMING_SNAKE_CASE          = False


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




 def 						a					( self       ):
   # clean up the VRAM after each test
   super().tearDown()
   gc.collect()
   torch.cuda.empty_cache()




 def 						a					( self       ):
   __UpperCAmelCase  :     Dict		=    VersatileDiffusionDualGuidedPipeline.from_pretrained('shi-labs/versatile-diffusion'       )
   # remove text_unet
   pipe.remove_unused_weights()
   pipe.to(_UpperCAmelCase       )
   pipe.set_progress_bar_config(disable=_UpperCAmelCase       )

   __UpperCAmelCase  :     Any		=    load_image(
       'https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/versatile_diffusion/benz.jpg'       )

   __UpperCAmelCase  :     Optional[Any]		=    torch.manual_seed(0       )
   __UpperCAmelCase  :     Union[str, Any]		=    pipe(
       prompt='first prompt'     ,     image=_UpperCAmelCase     ,     text_to_image_strength=0.7_5     ,     generator=_UpperCAmelCase     ,     guidance_scale=7.5     ,     num_inference_steps=2     ,     output_type='numpy'     ,     ).images

   with tempfile.TemporaryDirectory() as tmpdirname:
     pipe.save_pretrained(_UpperCAmelCase       )
     __UpperCAmelCase  :     List[str]		=    VersatileDiffusionDualGuidedPipeline.from_pretrained(_UpperCAmelCase       )

   pipe.to(_UpperCAmelCase       )
   pipe.set_progress_bar_config(disable=_UpperCAmelCase       )

   __UpperCAmelCase  :     List[str]		=    generator.manual_seed(0       )
   __UpperCAmelCase  :     Optional[int]		=    pipe(
       prompt='first prompt'     ,     image=_UpperCAmelCase     ,     text_to_image_strength=0.7_5     ,     generator=_UpperCAmelCase     ,     guidance_scale=7.5     ,     num_inference_steps=2     ,     output_type='numpy'     ,     ).images

   assert np.abs(image - new_image       ).sum() < 1E-5, "Models don't have the same forward pass"




 def 						a					( self       ):
   __UpperCAmelCase  :     Optional[Any]		=    VersatileDiffusionDualGuidedPipeline.from_pretrained('shi-labs/versatile-diffusion'       )
   pipe.remove_unused_weights()
   pipe.to(_UpperCAmelCase       )
   pipe.set_progress_bar_config(disable=_UpperCAmelCase       )

   __UpperCAmelCase  :     Tuple		=    'cyberpunk 2077'
   __UpperCAmelCase  :     Dict		=    load_image(
       'https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/versatile_diffusion/benz.jpg'       )
   __UpperCAmelCase  :     Optional[Any]		=    torch.manual_seed(0       )
   __UpperCAmelCase  :     List[Any]		=    pipe(
       prompt=_UpperCAmelCase     ,     image=_UpperCAmelCase     ,     text_to_image_strength=0.7_5     ,     generator=_UpperCAmelCase     ,     guidance_scale=7.5     ,     num_inference_steps=50     ,     output_type='numpy'     ,     ).images

   __UpperCAmelCase  :     Tuple		=    image[0, 253:256, 253:256, -1]

   assert image.shape == (1, 512, 512, 3)
   __UpperCAmelCase  :     Tuple		=    np.array([0.0_7_8_7, 0.0_8_4_9, 0.0_8_2_6, 0.0_8_1_2, 0.0_8_0_7, 0.0_7_9_5, 0.0_8_1_8, 0.0_7_9_8, 0.0_7_7_9]       )

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





