
"""simple docstring"""

import os
import random
from pathlib import Path
from typing import List

import numpy as np
import torch
from torch.cuda.amp import GradScaler

from .utils import (
    MODEL_NAME,
    OPTIMIZER_NAME,
    RNG_STATE_NAME,
    SCALER_NAME,
    SCHEDULER_NAME,
    get_pretty_name,
    is_tpu_available,
    is_xpu_available,
    save,
)


if is_tpu_available(check_device=False):
   import torch_xla.core.xla_model as xm

from .logging import get_logger
from .state import PartialState


SCREAMING_SNAKE_CASE_			    =  get_logger(__name__)




def 			lowercase__  (    lowerCAmelCase :    str       , lowerCAmelCase :    List[dict]       , lowerCAmelCase :    list       , lowerCAmelCase :    list       , lowerCAmelCase :    int       , lowerCAmelCase :    GradScaler = None       , )       ->					Any:






    """simple docstring"""




    for i, state in enumerate(lowerCAmelCase       ):
        UpperCAmelCase  	=    F"{MODEL_NAME}.bin" if i == 0 else F"{MODEL_NAME}_{i}.bin"
        UpperCAmelCase  	=    os.path.join(lowerCAmelCase       , lowerCAmelCase       )
        save(lowerCAmelCase       , lowerCAmelCase       )
        logger.info(F"Model weights saved in {output_model_file}"       )
    # Optimizer states
    for i, opt in enumerate(lowerCAmelCase       ):
        UpperCAmelCase  	=    opt.state_dict()
        UpperCAmelCase  	=    F"{OPTIMIZER_NAME}.bin" if i == 0 else F"{OPTIMIZER_NAME}_{i}.bin"
        UpperCAmelCase  	=    os.path.join(lowerCAmelCase       , lowerCAmelCase       )
        save(lowerCAmelCase       , lowerCAmelCase       )
        logger.info(F"Optimizer state saved in {output_optimizer_file}"       )
    # Scheduler states
    for i, scheduler in enumerate(lowerCAmelCase       ):
        UpperCAmelCase  	=    scheduler.state_dict()
        UpperCAmelCase  	=    F"{SCHEDULER_NAME}.bin" if i == 0 else F"{SCHEDULER_NAME}_{i}.bin"
        UpperCAmelCase  	=    os.path.join(lowerCAmelCase       , lowerCAmelCase       )
        save(lowerCAmelCase       , lowerCAmelCase       )
        logger.info(F"Scheduler state saved in {output_scheduler_file}"       )
    # GradScaler state
    if scaler is not None:
        UpperCAmelCase  	=    scaler.state_dict()
        UpperCAmelCase  	=    os.path.join(lowerCAmelCase       , lowerCAmelCase       )
        torch.save(lowerCAmelCase       , lowerCAmelCase       )
        logger.info(F"Gradient scaler state saved in {output_scaler_file}"       )
    # Random number generator states
    UpperCAmelCase  	=    {}
    UpperCAmelCase  	=    F"{RNG_STATE_NAME}_{process_index}.pkl"
    UpperCAmelCase  	=    random.getstate()
    UpperCAmelCase  	=    np.random.get_state()
    UpperCAmelCase  	=    torch.get_rng_state()
    if is_xpu_available():
        UpperCAmelCase  	=    torch.xpu.get_rng_state_all()
    else:
        UpperCAmelCase  	=    torch.cuda.get_rng_state_all()
    if is_tpu_available():
        UpperCAmelCase  	=    xm.get_rng_state()
    UpperCAmelCase  	=    os.path.join(lowerCAmelCase       , lowerCAmelCase       )
    torch.save(lowerCAmelCase       , lowerCAmelCase       )
    logger.info(F"Random states saved in {output_states_file}"       )
    return output_dir




def 			lowercase__  (    lowerCAmelCase :    Optional[int]       , lowerCAmelCase :    List[str]       , lowerCAmelCase :    str       , lowerCAmelCase :    str       , lowerCAmelCase :    List[str]       , lowerCAmelCase :    int=None       , lowerCAmelCase :    str=None       , **lowerCAmelCase :    Optional[int]       , )       ->					Union[str, Any]:






    """simple docstring"""




    if map_location not in [None, "cpu", "on_device"]:
        raise TypeError(
            'Unsupported optimizer map location passed, please choose one of `None`, `\'cpu\'`, or `\'on_device\'`'       )
    if map_location is None:
        UpperCAmelCase  	=    'cpu'
    elif map_location == "on_device":
        UpperCAmelCase  	=    PartialState().device
    # Model states
    for i, model in enumerate(lowerCAmelCase       ):
        UpperCAmelCase  	=    F"{MODEL_NAME}.bin" if i == 0 else F"{MODEL_NAME}_{i}.bin"
        UpperCAmelCase  	=    os.path.join(lowerCAmelCase       , lowerCAmelCase       )
        models[i].load_state_dict(torch.load(lowerCAmelCase       , map_location=lowerCAmelCase       )       , **lowerCAmelCase       )
    logger.info('All model weights loaded successfully'       )

    # Optimizer states
    for i, opt in enumerate(lowerCAmelCase       ):
        UpperCAmelCase  	=    F"{OPTIMIZER_NAME}.bin" if i == 0 else F"{OPTIMIZER_NAME}_{i}.bin"
        UpperCAmelCase  	=    os.path.join(lowerCAmelCase       , lowerCAmelCase       )
        UpperCAmelCase  	=    torch.load(lowerCAmelCase       , map_location=lowerCAmelCase       )
        optimizers[i].load_state_dict(lowerCAmelCase       )
    logger.info('All optimizer states loaded successfully'       )

    # Scheduler states
    for i, scheduler in enumerate(lowerCAmelCase       ):
        UpperCAmelCase  	=    F"{SCHEDULER_NAME}.bin" if i == 0 else F"{SCHEDULER_NAME}_{i}.bin"
        UpperCAmelCase  	=    os.path.join(lowerCAmelCase       , lowerCAmelCase       )
        scheduler.load_state_dict(torch.load(lowerCAmelCase       )       )
    logger.info('All scheduler states loaded successfully'       )

    # GradScaler state
    if scaler is not None:
        UpperCAmelCase  	=    os.path.join(lowerCAmelCase       , lowerCAmelCase       )
        scaler.load_state_dict(torch.load(lowerCAmelCase       )       )
        logger.info('GradScaler state loaded successfully'       )

    # Random states
    try:
        UpperCAmelCase  	=    torch.load(os.path.join(lowerCAmelCase       , F"{RNG_STATE_NAME}_{process_index}.pkl"       )       )
        random.setstate(states['random_state']       )
        np.random.set_state(states['numpy_random_seed']       )
        torch.set_rng_state(states['torch_manual_seed']       )
        if is_xpu_available():
            torch.xpu.set_rng_state_all(states['torch_xpu_manual_seed']       )
        else:
            torch.cuda.set_rng_state_all(states['torch_cuda_manual_seed']       )
        if is_tpu_available():
            xm.set_rng_state(states['xm_seed']       )
        logger.info('All random states loaded successfully'       )
    except Exception:
        logger.info('Could not load random states'       )




def 			lowercase__  (    lowerCAmelCase :    Dict       , lowerCAmelCase :    Optional[Any]       , lowerCAmelCase :    int = 0       )       ->					Optional[int]:






    """simple docstring"""




    UpperCAmelCase  	=    Path(lowerCAmelCase       ) / F"custom_checkpoint_{index}.pkl"
    logger.info(F"Saving the state of {get_pretty_name(lowerCAmelCase       )} to {save_location}"       )
    torch.save(obj.state_dict()       , lowerCAmelCase       )







def 			lowercase__  (    lowerCAmelCase :    List[str]       , lowerCAmelCase :    str       , lowerCAmelCase :    int = 0       )       ->					Union[str, Any]:






    """simple docstring"""




    UpperCAmelCase  	=    F"{path}/custom_checkpoint_{index}.pkl"
    logger.info(F"Loading the state of {get_pretty_name(lowerCAmelCase       )} from {load_location}"       )
    obj.load_state_dict(torch.load(lowerCAmelCase       , map_location='cpu'       )       )





"""simple docstring"""

from typing import SupportsIndex

import numpy as np
from scipy.ndimage import convolve




def 			lowercase__  (    lowerCAmelCase :    np.ndarray       , lowerCAmelCase :    np.ndarray       , lowerCAmelCase :    np.ndarray       )       ->					np.ndarray:






    """simple docstring"""




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

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

    UpperCAmelCase  	=    image[grid[:, :, 1], grid[:, :, 0]]

    # Set pixels at invalid locations to 0
    UpperCAmelCase  	=    0

    return warped




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






    """simple docstring"""




    if alpha is None:
        UpperCAmelCase  	=    0.1

    # Initialize flow
    UpperCAmelCase  	=    np.zeros_like(lowerCAmelCase       )
    UpperCAmelCase  	=    np.zeros_like(lowerCAmelCase       )

    # Prepare kernels for the calculation of the derivatives and the average velocity
    UpperCAmelCase  	=    np.array([[-1, 1], [-1, 1]]       ) * 0.25
    UpperCAmelCase  	=    np.array([[-1, -1], [1, 1]]       ) * 0.25
    UpperCAmelCase  	=    np.array([[1, 1], [1, 1]]       ) * 0.25
    UpperCAmelCase  	=    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(lowerCAmelCase       ):
        UpperCAmelCase  	=    warp(lowerCAmelCase       , lowerCAmelCase       , lowerCAmelCase       )
        UpperCAmelCase  	=    convolve(lowerCAmelCase       , lowerCAmelCase       ) + convolve(lowerCAmelCase       , lowerCAmelCase       )
        UpperCAmelCase  	=    convolve(lowerCAmelCase       , lowerCAmelCase       ) + convolve(lowerCAmelCase       , lowerCAmelCase       )
        UpperCAmelCase  	=    convolve(lowerCAmelCase       , lowerCAmelCase       ) + convolve(lowerCAmelCase       , -kernel_t       )

        UpperCAmelCase  	=    convolve(lowerCAmelCase       , lowerCAmelCase       )
        UpperCAmelCase  	=    convolve(lowerCAmelCase       , lowerCAmelCase       )

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

        UpperCAmelCase  	=    avg_horizontal_velocity - derivative_x * update
        UpperCAmelCase  	=    avg_vertical_velocity - derivative_y * update

    return horizontal_flow, vertical_flow


if __name__ == "__main__":
   import doctest

   doctest.testmod()



