

import json
import os
from collections.abc import Mapping
from typing import Dict, List, Optional, Union

import numpy as np
import torch

from .imports import is_safetensors_available







def _a					(			SCREAMING_SNAKE_CASE						:     Tuple       ,   SCREAMING_SNAKE_CASE						:     List[Any]       ,   SCREAMING_SNAKE_CASE						:     Optional[Any]       ,   SCREAMING_SNAKE_CASE						:     List[str]=None						)       ->       Union[str, Any]:

 """simple docstring"""






 __lowerCAmelCase:  str        = None
 # Check the string instead of the dtype to be compatible with versions of PyTorch that don't have bfloat16.
 if str(weight.dtype						) == "torch.bfloat16":
  # Need to reinterpret the underlined data as int16 since NumPy does not handle bfloat16s.
  __lowerCAmelCase:  List[Any]        = weight.view(torch.intaa						)
  __lowerCAmelCase:  Optional[Any]        = 'bfloat16'
 __lowerCAmelCase:  List[Any]        = weight.cpu().numpy()
 __lowerCAmelCase:  List[Any]        = os.path.join(SCREAMING_SNAKE_CASE       ,   f'''{weight_name}.dat'''						)
 if index is not None:
  if dtype is None:
   __lowerCAmelCase:  List[Any]        = str(array.dtype						)
  __lowerCAmelCase:  Optional[Any]        = {'dtype': dtype, 'shape': list(array.shape						)}
 if array.ndim == 0:
  __lowerCAmelCase:  int        = array[None]
 __lowerCAmelCase:  Any        = np.memmap(SCREAMING_SNAKE_CASE       ,   dtype=array.dtype       ,   mode='w+'       ,   shape=array.shape						)
 __lowerCAmelCase:  Optional[int]        = array[:]
 file_array.flush()
 return index







def _a					(			SCREAMING_SNAKE_CASE						:     Optional[Any]       ,   SCREAMING_SNAKE_CASE						:     Union[str, Any]						)       ->       Union[str, Any]:

 """simple docstring"""






 __lowerCAmelCase:  str        = tuple(weight_info['shape']						)
 if shape == ():
  # NumPy memory-mapped arrays can't have 0 dims so it was saved as 1d tensor
  __lowerCAmelCase:  Optional[Any]        = (1,)

 __lowerCAmelCase:  Any        = weight_info['dtype']
 if dtype == "bfloat16":
  # NumPy does not support bfloat16 so this was saved as a int16
  __lowerCAmelCase:  Optional[Any]        = 'int16'

 __lowerCAmelCase:  Tuple        = np.memmap(SCREAMING_SNAKE_CASE       ,   dtype=SCREAMING_SNAKE_CASE       ,   shape=SCREAMING_SNAKE_CASE       ,   mode='r'						)

 if len(weight_info['shape']						) == 0:
  __lowerCAmelCase:  Any        = weight[0]
 __lowerCAmelCase:  Union[str, Any]        = torch.tensor(SCREAMING_SNAKE_CASE						)
 if weight_info["dtype"] == "bfloat16":
  __lowerCAmelCase:  Optional[int]        = weight.view(torch.bfloataa						)

 return weight







def _a					(			SCREAMING_SNAKE_CASE						:     Union[str, Any]       ,   SCREAMING_SNAKE_CASE						:     List[str]						)       ->       Optional[int]:

 """simple docstring"""






 if index is None or len(SCREAMING_SNAKE_CASE						) == 0:
  # Nothing to save
  return

 __lowerCAmelCase:  Tuple        = os.path.join(SCREAMING_SNAKE_CASE       ,   'index.json'						)
 if os.path.isfile(SCREAMING_SNAKE_CASE						):
  with open(SCREAMING_SNAKE_CASE       ,   'r'       ,   encoding='utf-8'						) as f:
   __lowerCAmelCase:  Tuple        = json.load(SCREAMING_SNAKE_CASE						)
 else:
  __lowerCAmelCase:  Union[str, Any]        = {}
 current_index.update(SCREAMING_SNAKE_CASE						)

 with open(SCREAMING_SNAKE_CASE       ,   'w'       ,   encoding='utf-8'						) as f:
  json.dump(SCREAMING_SNAKE_CASE       ,   SCREAMING_SNAKE_CASE       ,   indent=2						)







def _a					(			SCREAMING_SNAKE_CASE						:     Union[str, os.PathLike]       ,   SCREAMING_SNAKE_CASE						:     Dict[str, torch.Tensor]						)       ->       int:

 """simple docstring"""






 os.makedirs(SCREAMING_SNAKE_CASE       ,   exist_ok=SCREAMING_SNAKE_CASE						)
 __lowerCAmelCase:  str        = {}
 for name, parameter in state_dict.items():
  __lowerCAmelCase:  Optional[Any]        = offload_weight(SCREAMING_SNAKE_CASE       ,   SCREAMING_SNAKE_CASE       ,   SCREAMING_SNAKE_CASE       ,   index=SCREAMING_SNAKE_CASE						)

 # Update index
 save_offload_index(SCREAMING_SNAKE_CASE       ,   SCREAMING_SNAKE_CASE						)






class 				A_					(						snake_case__   ):


     def __init__(				self     :						List[str]     ,    UpperCAmelCase     :						Mapping     ,    UpperCAmelCase     :						str   )				->      Dict:
      __lowerCAmelCase:  List[str]        = dataset
      __lowerCAmelCase:  List[Any]        = prefix


     def __getitem__(				self     :						Dict     ,    UpperCAmelCase     :						Union[str, Any]   )				->      int:
      return self.dataset[F'''{self.prefix}{key}''']


     def __iter__(				self     :						Optional[Any]   )				->      Optional[int]:
      return iter([key for key in self.dataset if key.startswith(self.prefix   )]   )


     def __len__(				self     :						List[str]   )				->      Union[str, Any]:
      return len(self.dataset   )







class 				A_					(						snake_case__   ):


     def __init__(				self     :						str     ,    UpperCAmelCase     :						Dict[str, torch.Tensor] = None     ,    UpperCAmelCase     :						Optional[Union[str, os.PathLike]] = None     ,    UpperCAmelCase     :						Mapping = None     ,    UpperCAmelCase     :						List[str]=None     ,    )				->      List[str]:
      if state_dict is None and save_folder is None:
       raise ValueError('Need either a `state_dict` or a `save_folder` containing offloaded weights.'   )

      __lowerCAmelCase:  Tuple        = {} if state_dict is None else state_dict
      __lowerCAmelCase:  Optional[Any]        = save_folder
      if index is None and save_folder is not None:
       with open(os.path.join(UpperCAmelCase     ,    'index.json'   )   ) as f:
        __lowerCAmelCase:  Tuple        = json.load(UpperCAmelCase   )
      __lowerCAmelCase:  Dict        = {} if index is None else index
      __lowerCAmelCase:  Optional[Any]        = list(self.state_dict.keys()   )
      self.all_keys.extend([key for key in self.index if key not in self.all_keys]   )
      __lowerCAmelCase:  Optional[Any]        = device


     def __getitem__(				self     :						List[str]     ,    UpperCAmelCase     :						str   )				->      Optional[Any]:
      # State dict gets priority
      if key in self.state_dict:
       return self.state_dict[key]
      __lowerCAmelCase:  str        = self.index[key]
      if weight_info.get('safetensors_file'   ) is not None:
       if not is_safetensors_available():
        raise ImportError('These offloaded weights require the use of safetensors: `pip install safetensors`.'   )

       from safetensors import safe_open

       __lowerCAmelCase:  Tuple        = 'cpu' if self.device is None else self.device
       with safe_open(weight_info['safetensors_file']     ,    framework='pt'     ,    device=UpperCAmelCase   ) as f:
        __lowerCAmelCase:  Union[str, Any]        = f.get_tensor(weight_info.get('weight_name'     ,    UpperCAmelCase   )   )

       if "dtype" in weight_info:
        return tensor.to(getattr(UpperCAmelCase     ,    weight_info['dtype']   )   )
       else:
        return tensor

      __lowerCAmelCase:  int        = os.path.join(self.save_folder     ,    F'''{key}.dat'''   )
      return load_offloaded_weight(UpperCAmelCase     ,    UpperCAmelCase   )


     def __iter__(				self     :						List[str]   )				->      List[str]:
      return iter(self.all_keys   )


     def __len__(				self     :						Union[str, Any]   )				->      str:
      return len(self.all_keys   )







def _a					(			SCREAMING_SNAKE_CASE						:     Dict[str, torch.Tensor]       ,   SCREAMING_SNAKE_CASE						:     List[str]						)       ->       Optional[int]:

 """simple docstring"""






 __lowerCAmelCase:  Union[str, Any]        = {}
 for module_name in submodule_names:
  # We want to catch module_name parameter (module_name.xxx) or potentially module_name, but not any of the
  # submodules that could being like module_name (transformers.h.1 and transformers.h.10 for instance)
  result.update(
      {
          key: param
          for key, param in state_dict.items()
          if key == module_name or key.startswith(module_name + '.'						)
      }						)
 return result








from typing import SupportsIndex

import numpy as np
from scipy.ndimage import convolve







def _a					(			SCREAMING_SNAKE_CASE						:     np.ndarray       ,   SCREAMING_SNAKE_CASE						:     np.ndarray       ,   SCREAMING_SNAKE_CASE						:     np.ndarray						)       ->       np.ndarray:

 """simple docstring"""






 __lowerCAmelCase:  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
 __lowerCAmelCase:  List[str]        = np.stack(
     np.meshgrid(np.arange(0       ,   image.shape[1]						)       ,   np.arange(0       ,   image.shape[0]						)						)       ,   2						)
 __lowerCAmelCase:  int        = np.round(grid - flow						).astype(np.intaa						)

 # Find the locations outside of the original image
 __lowerCAmelCase:  Dict        = (grid < 0) | (grid >= np.array([image.shape[1], image.shape[0]]						))
 __lowerCAmelCase:  str        = 0

 __lowerCAmelCase:  List[str]        = image[grid[:, :, 1], grid[:, :, 0]]

 # Set pixels at invalid locations to 0
 __lowerCAmelCase:  List[str]        = 0

 return warped







def _a					(			SCREAMING_SNAKE_CASE						:     np.ndarray       ,   SCREAMING_SNAKE_CASE						:     np.ndarray       ,   SCREAMING_SNAKE_CASE						:     SupportsIndex       ,   SCREAMING_SNAKE_CASE						:     float | None = None       ,   )       ->       tuple[np.ndarray, np.ndarray]:

 """simple docstring"""






 if alpha is None:
  __lowerCAmelCase:  Tuple        = 0.1

 # Initialize flow
 __lowerCAmelCase:  Optional[int]        = np.zeros_like(SCREAMING_SNAKE_CASE						)
 __lowerCAmelCase:  Dict        = np.zeros_like(SCREAMING_SNAKE_CASE						)

 # Prepare kernels for the calculation of the derivatives and the average velocity
 __lowerCAmelCase:  List[Any]        = np.array([[-1, 1], [-1, 1]]						) * 0.2_5
 __lowerCAmelCase:  int        = np.array([[-1, -1], [1, 1]]						) * 0.2_5
 __lowerCAmelCase:  Tuple        = np.array([[1, 1], [1, 1]]						) * 0.2_5
 __lowerCAmelCase:  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(SCREAMING_SNAKE_CASE						):
  __lowerCAmelCase:  Optional[int]        = warp(SCREAMING_SNAKE_CASE       ,   SCREAMING_SNAKE_CASE       ,   SCREAMING_SNAKE_CASE						)
  __lowerCAmelCase:  List[str]        = convolve(SCREAMING_SNAKE_CASE       ,   SCREAMING_SNAKE_CASE						) + convolve(SCREAMING_SNAKE_CASE       ,   SCREAMING_SNAKE_CASE						)
  __lowerCAmelCase:  int        = convolve(SCREAMING_SNAKE_CASE       ,   SCREAMING_SNAKE_CASE						) + convolve(SCREAMING_SNAKE_CASE       ,   SCREAMING_SNAKE_CASE						)
  __lowerCAmelCase:  List[Any]        = convolve(SCREAMING_SNAKE_CASE       ,   SCREAMING_SNAKE_CASE						) + convolve(SCREAMING_SNAKE_CASE       ,   -kernel_t						)

  __lowerCAmelCase:  Any        = convolve(SCREAMING_SNAKE_CASE       ,   SCREAMING_SNAKE_CASE						)
  __lowerCAmelCase:  Dict        = convolve(SCREAMING_SNAKE_CASE       ,   SCREAMING_SNAKE_CASE						)

  # This updates the flow as proposed in the paper (Step 12)
  __lowerCAmelCase:  Tuple        = (
      derivative_x * avg_horizontal_velocity
      + derivative_y * avg_vertical_velocity
      + derivative_t
  )
  __lowerCAmelCase:  int        = update / (alpha**2 + derivative_x**2 + derivative_y**2)

  __lowerCAmelCase:  List[Any]        = avg_horizontal_velocity - derivative_x * update
  __lowerCAmelCase:  Union[str, Any]        = avg_vertical_velocity - derivative_y * update

 return horizontal_flow, vertical_flow


if __name__ == "__main__":
  import doctest

  doctest.testmod()





