cosmos-framework/cosmos_framework/inference/metrics.py

# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
# SPDX-License-Identifier: OpenMDW-1.1

"""
Evaluation metrics for Action models.

This module provides standard metrics used for evaluating video prediction
and action prediction quality.

All metric functions follow a unified structure:
1. Check if gt and pred share the same shape
2. Check if shape is expected
3. Detach tensor while keeping the device (if tensor)
4. Compute metric
5. Return float
"""

from __future__ import annotations

import warnings

import numpy as np
import torch


def _get_dtype_info(
    gt: np.ndarray | torch.Tensor,
    pred: np.ndarray | torch.Tensor,
) -> tuple[float, float, float]:
    """
    Get dtype info and derive data range.

    Args:
        gt: Ground truth array/tensor.
        pred: Predicted array/tensor.

    Returns:
        Tuple of (data_range, expected_min, expected_max).

    Raises:
        ValueError: If dtypes don't match or are unsupported.
    """
    if isinstance(gt, torch.Tensor):
        gt_is_float = gt.dtype.is_floating_point
        gt_is_uint8 = gt.dtype == torch.uint8
        gt_dtype_str = str(gt.dtype)
    else:
        gt_is_float = np.issubdtype(gt.dtype, np.floating)
        gt_is_uint8 = gt.dtype == np.uint8
        gt_dtype_str = str(gt.dtype)

    if isinstance(pred, torch.Tensor):
        pred_is_float = pred.dtype.is_floating_point
        pred_is_uint8 = pred.dtype == torch.uint8
        pred_dtype_str = str(pred.dtype)
    else:
        pred_is_float = np.issubdtype(pred.dtype, np.floating)
        pred_is_uint8 = pred.dtype == np.uint8
        pred_dtype_str = str(pred.dtype)

    if gt_is_float != pred_is_float or gt_is_uint8 != pred_is_uint8:
        raise ValueError(f"Dtype mismatch: gt {gt_dtype_str} vs pred {pred_dtype_str}")

    if gt_is_float:
        return 2.0, -1.0, 1.0
    elif gt_is_uint8:
        return 255.0, 0.0, 255.0
    else:
        raise ValueError(f"Unsupported dtype: {gt_dtype_str}. Expected float or uint8.")


def _compute_motion_mask(
    video: np.ndarray | torch.Tensor,
    threshold_percentile: float,
) -> np.ndarray | torch.Tensor:
    """
    Compute per-frame motion mask based on frame differences.

    For each frame t, computes |video[t] - video[t-1]| and thresholds based on
    the given percentile. Frame 0 uses the same mask as frame 1.

    Args:
        video: Video array/tensor of shape (C, T, H, W) or (B, C, T, H, W).
        threshold_percentile: Percentile threshold (0-100). Pixels with motion
            magnitude above this percentile are marked as dynamic.
            E.g., 80.0 means top 20% of motion is considered "dynamic".

    Returns:
        Boolean mask of shape (T, H, W) or (B, T, H, W) where True = dynamic pixel.
    """
    is_tensor = isinstance(video, torch.Tensor)

    # Handle both (C, T, H, W) and (B, C, T, H, W) shapes
    if video.ndim == 4:
        # (C, T, H, W) -> time is dim 1
        time_dim = 1
    else:
        # (B, C, T, H, W) -> time is dim 2
        time_dim = 2

    if is_tensor:
        video = video.detach().float()
        # Compute frame differences: |video[t] - video[t-1]|
        if time_dim == 1:
            diff = torch.abs(video[:, 1:] - video[:, :-1])  # [C,T-1,H,W]
            # Average over channels to get motion magnitude
            motion_magnitude = diff.mean(dim=0)  # [T-1,H,W]
        else:
            diff = torch.abs(video[:, :, 1:] - video[:, :, :-1])  # [B,C,T-1,H,W]
            motion_magnitude = diff.mean(dim=1)  # [B,T-1,H,W]

        # Compute threshold value from percentile
        threshold_value = torch.quantile(motion_magnitude.flatten().float(), threshold_percentile / 100.0)

        # Create mask for frames 1..T-1
        mask_after_first = motion_magnitude > threshold_value

        # For frame 0, use the same mask as frame 1
        if time_dim == 1:
            first_frame_mask = mask_after_first[0:1]  # [1,H,W]
            mask = torch.cat([first_frame_mask, mask_after_first], dim=0)  # [T,H,W]
        else:
            first_frame_mask = mask_after_first[:, 0:1]  # [B,1,H,W]
            mask = torch.cat([first_frame_mask, mask_after_first], dim=1)  # [B,T,H,W]
    else:
        video = np.asarray(video, dtype=np.float32)
        # Compute frame differences
        if time_dim == 1:
            diff = np.abs(video[:, 1:] - video[:, :-1])  # [C,T-1,H,W]
            motion_magnitude = diff.mean(axis=0)  # [T-1,H,W]
        else:
            diff = np.abs(video[:, :, 1:] - video[:, :, :-1])  # [B,C,T-1,H,W]
            motion_magnitude = diff.mean(axis=1)  # [B,T-1,H,W]

        # Compute threshold value from percentile
        threshold_value = np.percentile(motion_magnitude.flatten(), threshold_percentile)

        # Create mask for frames 1..T-1
        mask_after_first = motion_magnitude > threshold_value

        # For frame 0, use the same mask as frame 1
        if time_dim == 1:
            first_frame_mask = mask_after_first[0:1]  # [1,H,W]
            mask = np.concatenate([first_frame_mask, mask_after_first], axis=0)  # [T,H,W]
        else:
            first_frame_mask = mask_after_first[:, 0:1]  # [B,1,H,W]
            mask = np.concatenate([first_frame_mask, mask_after_first], axis=1)  # [B,T,H,W]

    return mask


def compute_psnr(
    gt: np.ndarray | torch.Tensor,
    pred: np.ndarray | torch.Tensor,
) -> float:
    """
    Compute Peak Signal-to-Noise Ratio (PSNR) between ground truth and prediction.

    PSNR is defined as: 20 * log10(MAX / sqrt(MSE))

    The data range is automatically derived from dtype:
    - float dtype: expects values in [-1, 1], data_range = 2.0
    - uint8 dtype: expects values in [0, 255], data_range = 255.0

    Args:
        gt: Ground truth array/tensor of shape (C, T, H, W) or (B, C, T, H, W).
            - If float dtype: values must be in [-1, 1] range.
            - If uint8 dtype: values must be in [0, 255] range.
        pred: Predicted array/tensor of same shape and dtype as gt.

    Returns:
        PSNR value in decibels (dB). Higher is better.

    Raises:
        ValueError: If shapes don't match, dtypes don't match, dtype is unsupported,
            or values are out of expected range.

    Example:
        >>> gt_video = torch.randn(3, 16, 96, 96).clamp(-1, 1)  # (C, T, H, W), float in [-1, 1]
        >>> pred_video = torch.randn(3, 16, 96, 96).clamp(-1, 1)
        >>> psnr = compute_psnr(gt_video, pred_video)
    """
    # 1. Check if gt and pred share the same shape
    if gt.shape != pred.shape:
        raise ValueError(f"Shape mismatch: gt {gt.shape} vs pred {pred.shape}")

    # 2. Check if shape is expected: (C, T, H, W) or (B, C, T, H, W)
    if gt.ndim not in (4, 5):
        raise ValueError(
            f"Expected gt to have 4 dims (C, T, H, W) or 5 dims (B, C, T, H, W), got {gt.ndim} dims: {gt.shape}"
        )

    # 3. Check dtype and derive data range
    data_range, expected_min, expected_max = _get_dtype_info(gt, pred)

    # 4. Validate value range
    if isinstance(gt, torch.Tensor):
        gt_min, gt_max = gt.min().item(), gt.max().item()
    else:
        gt_min, gt_max = float(gt.min()), float(gt.max())

    if isinstance(pred, torch.Tensor):
        pred_min, pred_max = pred.min().item(), pred.max().item()
    else:
        pred_min, pred_max = float(pred.min()), float(pred.max())

    if gt_min < expected_min or gt_max > expected_max:
        raise ValueError(f"gt values out of range: got [{gt_min}, {gt_max}], expected [{expected_min}, {expected_max}]")
    if pred_min < expected_min or pred_max > expected_max:
        raise ValueError(
            f"pred values out of range: got [{pred_min}, {pred_max}], expected [{expected_min}, {expected_max}]"
        )

    # 5. Detach tensor while keeping the device (if tensor)
    # 6. Compute metric
    if isinstance(gt, torch.Tensor) and isinstance(pred, torch.Tensor):
        gt_t = gt.detach().float()
        pred_t = pred.detach().float()
        mse = torch.mean((gt_t - pred_t) ** 2).item()
    else:
        gt_np = np.asarray(gt, dtype=np.float32)
        pred_np = np.asarray(pred, dtype=np.float32)
        mse = float(np.mean((gt_np - pred_np) ** 2))

    if mse == 0:
        psnr = float("inf")
    else:
        psnr = 20.0 * np.log10(data_range / np.sqrt(mse))

    # 7. Return float
    return float(psnr)


def compute_dynamic_psnr(
    gt: np.ndarray | torch.Tensor,
    pred: np.ndarray | torch.Tensor,
    motion_threshold_percentile: float = 80.0,
) -> tuple[float, np.ndarray | torch.Tensor]:
    """
    Compute PSNR only on dynamic (moving) pixels, ignoring static background.

    This metric is more meaningful for videos with large static backgrounds,
    as it focuses on the quality of dynamic content rather than being dominated
    by easy-to-predict static regions.

    Dynamic pixels are detected using frame differences: for each frame t,
    pixels where |gt[t] - gt[t-1]| exceeds a percentile threshold are considered
    "dynamic". PSNR is computed only on these pixels.

    Args:
        gt: Ground truth array/tensor of shape (C, T, H, W) or (B, C, T, H, W).
            - If float dtype: values must be in [-1, 1] range.
            - If uint8 dtype: values must be in [0, 255] range.
        pred: Predicted array/tensor of same shape and dtype as gt.
        motion_threshold_percentile: Percentile threshold for motion detection.
            Default 80.0 means top 20% of motion magnitude is considered "dynamic".
            Higher values = stricter threshold = fewer dynamic pixels.

    Returns:
        Tuple of (psnr, motion_mask):
            - psnr: PSNR value in decibels (dB) computed only on dynamic regions.
              Higher is better. Returns regular PSNR if no dynamic pixels are found.
            - motion_mask: Boolean mask of shape (T, H, W) or (B, T, H, W) where
              True indicates a dynamic pixel. Same type as input (tensor or ndarray).

    Raises:
        ValueError: If shapes don't match, dtypes don't match, dtype is unsupported,
            or values are out of expected range.

    Example:
        >>> gt_video = torch.randn(3, 16, 96, 96).clamp(-1, 1)  # (C, T, H, W)
        >>> pred_video = torch.randn(3, 16, 96, 96).clamp(-1, 1)
        >>> dynamic_psnr, motion_mask = compute_dynamic_psnr(gt_video, pred_video)
        >>> print(f"Dynamic PSNR: {dynamic_psnr:.2f}, Mask shape: {motion_mask.shape}")
    """

    # 1. Check if gt and pred share the same shape
    if gt.shape != pred.shape:
        raise ValueError(f"Shape mismatch: gt {gt.shape} vs pred {pred.shape}")

    # 2. Check if shape is expected: (C, T, H, W) or (B, C, T, H, W)
    if gt.ndim not in (4, 5):
        raise ValueError(
            f"Expected gt to have 4 dims (C, T, H, W) or 5 dims (B, C, T, H, W), got {gt.ndim} dims: {gt.shape}"
        )

    # 3. Check dtype and derive data range
    data_range, expected_min, expected_max = _get_dtype_info(gt, pred)

    # 4. Validate value range
    if isinstance(gt, torch.Tensor):
        gt_min, gt_max = gt.min().item(), gt.max().item()
    else:
        gt_min, gt_max = float(gt.min()), float(gt.max())

    if isinstance(pred, torch.Tensor):
        pred_min, pred_max = pred.min().item(), pred.max().item()
    else:
        pred_min, pred_max = float(pred.min()), float(pred.max())

    if gt_min < expected_min or gt_max > expected_max:
        raise ValueError(f"gt values out of range: got [{gt_min}, {gt_max}], expected [{expected_min}, {expected_max}]")
    if pred_min < expected_min or pred_max > expected_max:
        raise ValueError(
            f"pred values out of range: got [{pred_min}, {pred_max}], expected [{expected_min}, {expected_max}]"
        )

    # 5. Compute motion mask from GT
    motion_mask = _compute_motion_mask(gt, motion_threshold_percentile)

    # 6. Check if there are any dynamic pixels
    if isinstance(motion_mask, torch.Tensor):
        num_dynamic = motion_mask.sum().item()
    else:
        num_dynamic = int(motion_mask.sum())

    if num_dynamic == 0:
        warnings.warn(
            "No dynamic pixels found in video. Returning regular PSNR. Consider lowering motion_threshold_percentile.",
            stacklevel=2,
        )
        return compute_psnr(gt, pred), motion_mask

    # 7. Compute MSE only on dynamic pixels
    if isinstance(gt, torch.Tensor) and isinstance(pred, torch.Tensor):
        gt_t = gt.detach().float()
        pred_t = pred.detach().float()
        # motion_mask is a tensor when gt is a tensor
        mask_t = motion_mask if isinstance(motion_mask, torch.Tensor) else torch.from_numpy(motion_mask)

        # Expand mask to match video shape (add channel dimension)
        if gt.ndim == 4:
            # (C, T, H, W) - mask is (T, H, W)
            expanded_mask = mask_t.unsqueeze(0).expand_as(gt_t)  # [C,T,H,W]
        else:
            # (B, C, T, H, W) - mask is (B, T, H, W)
            expanded_mask = mask_t.unsqueeze(1).expand_as(gt_t)  # [B,C,T,H,W]

        # Compute squared error only on dynamic pixels
        squared_error = (gt_t - pred_t) ** 2
        masked_squared_error = squared_error[expanded_mask]
        mse = masked_squared_error.mean().item()
    else:
        gt_np = np.asarray(gt, dtype=np.float32)
        pred_np = np.asarray(pred, dtype=np.float32)
        # motion_mask is an ndarray when gt is an ndarray
        mask_np = motion_mask if isinstance(motion_mask, np.ndarray) else motion_mask.numpy()

        # Expand mask to match video shape
        if gt_np.ndim == 4:
            # (C, T, H, W) - mask is (T, H, W)
            expanded_mask = np.broadcast_to(mask_np[np.newaxis, :, :, :], gt_np.shape)
        else:
            # (B, C, T, H, W) - mask is (B, T, H, W)
            expanded_mask = np.broadcast_to(mask_np[:, np.newaxis, :, :, :], gt_np.shape)

        # Compute squared error only on dynamic pixels
        squared_error = (gt_np - pred_np) ** 2
        masked_squared_error = squared_error[expanded_mask]
        mse = float(masked_squared_error.mean())

    # 8. Convert MSE to PSNR
    if mse == 0:
        psnr = float("inf")
    else:
        psnr = 20.0 * np.log10(data_range / np.sqrt(mse))

    return float(psnr), motion_mask


def compute_ssim(
    gt: np.ndarray | torch.Tensor,
    pred: np.ndarray | torch.Tensor,
) -> float:
    """
    Compute Structural Similarity Index (SSIM) between ground truth and prediction.

    SSIM measures the structural similarity between two images, considering
    luminance, contrast, and structure. It is computed per-frame and averaged.

    Uses skimage.metrics.structural_similarity for computation, following the
    implementation in projects/cosmos/tokenizer/evaluation/metric.py.

    The data range is automatically derived from dtype:
    - float dtype: expects values in [-1, 1], data_range = 2.0
    - uint8 dtype: expects values in [0, 255], data_range = 255.0

    Args:
        gt: Ground truth array/tensor of shape (C, T, H, W) or (B, C, T, H, W).
            If float dtype: values must be in [-1, 1] range.
            If uint8 dtype: values must be in [0, 255] range.
        pred: Predicted array/tensor of same shape and dtype as gt.

    Returns:
        SSIM value in range [-1, 1]. Higher is better (1.0 = identical).

    Raises:
        ValueError: If shapes don't match, dtypes don't match, dtype is unsupported,
            or values are out of expected range.

    Example:
        >>> gt_video = torch.randn(3, 16, 96, 96).clamp(-1, 1)  # (C, T, H, W), float in [-1, 1]
        >>> pred_video = torch.randn(3, 16, 96, 96).clamp(-1, 1)
        >>> ssim_val = compute_ssim(gt_video, pred_video)
    """
    from skimage.metrics import structural_similarity as ssim

    # 1. Check if gt and pred share the same shape
    if gt.shape != pred.shape:
        raise ValueError(f"Shape mismatch: gt {gt.shape} vs pred {pred.shape}")

    # 2. Check if shape is expected: (C, T, H, W) or (B, C, T, H, W)
    if gt.ndim not in (4, 5):
        raise ValueError(
            f"Expected gt to have 4 dims (C, T, H, W) or 5 dims (B, C, T, H, W), got {gt.ndim} dims: {gt.shape}"
        )

    # 3. Check dtype and derive data range
    data_range, expected_min, expected_max = _get_dtype_info(gt, pred)

    # 4. Validate value range
    if isinstance(gt, torch.Tensor):
        gt_min, gt_max = gt.min().item(), gt.max().item()
    else:
        gt_min, gt_max = float(gt.min()), float(gt.max())

    if isinstance(pred, torch.Tensor):
        pred_min, pred_max = pred.min().item(), pred.max().item()
    else:
        pred_min, pred_max = float(pred.min()), float(pred.max())

    if isinstance(gt, torch.Tensor) and gt.dtype == torch.bfloat16:
        gt = gt.float()
    if isinstance(pred, torch.Tensor) and pred.dtype == torch.bfloat16:
        pred = pred.float()

    if gt_min < expected_min or gt_max > expected_max:
        raise ValueError(f"gt values out of range: got [{gt_min}, {gt_max}], expected [{expected_min}, {expected_max}]")
    if pred_min < expected_min or pred_max > expected_max:
        raise ValueError(
            f"pred values out of range: got [{pred_min}, {pred_max}], expected [{expected_min}, {expected_max}]"
        )

    # 5. Convert to numpy arrays
    if isinstance(gt, torch.Tensor):
        gt_np = gt.detach().cpu().numpy()
    else:
        gt_np = np.asarray(gt)

    if isinstance(pred, torch.Tensor):
        pred_np = pred.detach().cpu().numpy()
    else:
        pred_np = np.asarray(pred)

    # 6. Reshape to (N, C, H, W) where N = number of frames
    if gt_np.ndim == 4:
        # (C, T, H, W) -> (T, C, H, W)
        gt_frames = np.transpose(gt_np, (1, 0, 2, 3))
        pred_frames = np.transpose(pred_np, (1, 0, 2, 3))
    else:
        # (B, C, T, H, W) -> (B*T, C, H, W)
        b, c, t, h, w = gt_np.shape
        gt_frames = np.transpose(gt_np, (0, 2, 1, 3, 4)).reshape(b * t, c, h, w)
        pred_frames = np.transpose(pred_np, (0, 2, 1, 3, 4)).reshape(b * t, c, h, w)

    # 7. Compute SSIM per frame using skimage
    ssim_values = []
    for gt_frame, pred_frame in zip(gt_frames, pred_frames):
        # gt_frame and pred_frame are (C, H, W)
        frame_ssim = ssim(gt_frame, pred_frame, channel_axis=0, data_range=data_range)
        ssim_values.append(frame_ssim)

    # 8. Return average SSIM
    return float(np.mean(ssim_values))


def compute_action_mse(
    gt_action: np.ndarray | torch.Tensor,
    pred_action: np.ndarray | torch.Tensor,
) -> float:
    """
    Compute Mean Squared Error (MSE) between ground truth and predicted actions.

    Args:
        gt_action: Ground truth array/tensor of shape (T, D) or (B, T, D),
            where T is the number of timesteps and D is the action dimension.
        pred_action: Predicted array/tensor of same shape as gt_action.

    Returns:
        MSE value. Lower is better.

    Example:
        >>> gt = np.random.randn(16, 2)  # (T, D) - 16 timesteps, 2D actions
        >>> pred = np.random.randn(16, 2)
        >>> mse = compute_action_mse(gt, pred)
    """
    # 1. Check if gt and pred share the same shape
    if gt_action.shape != pred_action.shape:
        raise ValueError(f"Shape mismatch: gt_action {gt_action.shape} vs pred_action {pred_action.shape}")

    # 2. Check if shape is expected: (T, D) or (B, T, D)
    if gt_action.ndim not in (2, 3):
        raise ValueError(
            f"Expected gt_action to have 2 dims (T, D) or 3 dims (B, T, D), got {gt_action.ndim} dims: {gt_action.shape}"
        )

    # 3. Detach tensor while keeping the device (if tensor)
    # 4. Compute metric
    if isinstance(gt_action, torch.Tensor) and isinstance(pred_action, torch.Tensor):
        gt_t = gt_action.detach()
        pred_t = pred_action.detach()
        mse = torch.mean((gt_t - pred_t) ** 2).item()
    else:
        gt_np = np.asarray(gt_action)
        pred_np = np.asarray(pred_action)
        mse = float(np.mean((gt_np - pred_np) ** 2))

    # 5. Return float
    return float(mse)


def compute_grouped_action_mse(
    gt_action: np.ndarray | torch.Tensor,
    pred_action: np.ndarray | torch.Tensor,
) -> dict[str, float]:
    """
    Compute grouped MSE for translation, rotation, and gripper action components.

    This metric is useful for robotics tasks where actions are structured as
    [translation(3), rotation(9), gripper(1)] using 9D rotation matrix representation.

    NOTE: All actions must be converted to 9D rotation matrix format (13D total)
    before calling this function. Use conversion utilities in the inference stage
    to convert from other rotation representations (axis-angle, 6D) to 9D.

    Args:
        gt_action: Ground truth array/tensor of shape (T, 13) or (B, T, 13),
            where T is the number of timesteps.
            Expected format: [translation(3), rotation_matrix(9), gripper(1)]
        pred_action: Predicted array/tensor of same shape as gt_action.

    Returns:
        Dictionary with MSE values for each component:
            - "translation": MSE for x, y, z (dims 0-2)
            - "rotation": MSE for flattened rotation matrix (dims 3-11)
            - "gripper": MSE for gripper (dim 12)
        Values are 0.0 if the corresponding dimensions are not present.
    """
    if gt_action.shape != pred_action.shape:
        raise ValueError(f"Shape mismatch: gt_action {gt_action.shape} vs pred_action {pred_action.shape}")

    if gt_action.ndim not in (2, 3):
        raise ValueError(
            f"Expected gt_action to have 2 dims (T, D) or 3 dims (B, T, D), got {gt_action.ndim} dims: {gt_action.shape}"
        )

    if isinstance(gt_action, torch.Tensor):
        gt_np = gt_action.detach().cpu().numpy()
    else:
        gt_np = np.asarray(gt_action)

    if isinstance(pred_action, torch.Tensor):
        pred_np = pred_action.detach().cpu().numpy()
    else:
        pred_np = np.asarray(pred_action)

    result: dict[str, float] = {"translation": 0.0, "rotation": 0.0, "gripper": 0.0}
    action_dim = gt_np.shape[-1]
    gripper_idx = 12

    if action_dim >= 3:
        result["translation"] = float(np.mean((gt_np[..., :3] - pred_np[..., :3]) ** 2))

    # Rotation: dimensions 3-11 (flattened 3x3 rotation matrix)
    if action_dim >= gripper_idx:
        result["rotation"] = float(np.mean((gt_np[..., 3:gripper_idx] - pred_np[..., 3:gripper_idx]) ** 2))

    # Gripper: dimension 12
    if action_dim > gripper_idx:
        result["gripper"] = float(np.mean((gt_np[..., gripper_idx] - pred_np[..., gripper_idx]) ** 2))

    return result


def compute_action_mae(
    gt_action: np.ndarray | torch.Tensor,
    pred_action: np.ndarray | torch.Tensor,
) -> float:
    """
    Compute Mean Absolute Error (MAE) between ground truth and predicted actions.

    Args:
        gt_action: Ground truth array/tensor of shape (T, D) or (B, T, D),
            where T is the number of timesteps and D is the action dimension.
        pred_action: Predicted array/tensor of same shape as gt_action.

    Returns:
        MAE value. Lower is better.

    Example:
        >>> gt = np.random.randn(16, 2)  # (T, D) - 16 timesteps, 2D actions
        >>> pred = np.random.randn(16, 2)
        >>> mae = compute_action_mae(gt, pred)
    """
    # 1. Check if gt and pred share the same shape
    if gt_action.shape != pred_action.shape:
        raise ValueError(f"Shape mismatch: gt_action {gt_action.shape} vs pred_action {pred_action.shape}")

    # 2. Check if shape is expected: (T, D) or (B, T, D)
    if gt_action.ndim not in (2, 3):
        raise ValueError(
            f"Expected gt_action to have 2 dims (T, D) or 3 dims (B, T, D), got {gt_action.ndim} dims: {gt_action.shape}"
        )

    # 3. Detach tensor while keeping the device (if tensor)
    # 4. Compute metric
    if isinstance(gt_action, torch.Tensor) and isinstance(pred_action, torch.Tensor):
        gt_t = gt_action.detach()
        pred_t = pred_action.detach()
        mae = torch.mean(torch.abs(gt_t - pred_t)).item()
    else:
        gt_np = np.asarray(gt_action)
        pred_np = np.asarray(pred_action)
        mae = float(np.mean(np.abs(gt_np - pred_np)))

    # 5. Return float
    return float(mae)


def compute_geodesic_rotation_error(
    gt_rot: np.ndarray,
    pred_rot: np.ndarray,
) -> np.ndarray:
    """Geodesic angular error between rotation matrices on SO(3).

    Computes ``arccos((tr(R_gt^T @ R_pred) - 1) / 2)`` for each pair,
    returning the angular distance in **degrees**.

    Args:
        gt_rot: Ground-truth rotation matrices of shape ``(N, 3, 3)``.
        pred_rot: Predicted rotation matrices of shape ``(N, 3, 3)``.

    Returns:
        Per-element angular errors in degrees, shape ``(N,)``.
    """
    if gt_rot.shape != pred_rot.shape or gt_rot.shape[-2:] != (3, 3):
        raise ValueError(f"Expected (N,3,3) arrays, got gt={gt_rot.shape}, pred={pred_rot.shape}")

    R_err = np.matmul(np.transpose(gt_rot, (0, 2, 1)), pred_rot)  # [N,3,3]
    trace = np.trace(R_err, axis1=1, axis2=2)  # [N]
    cos_angle = np.clip((trace - 1.0) / 2.0, -1.0, 1.0)  # [N]
    return np.degrees(np.arccos(cos_angle))  # [N]