robometric_frame.trajectory_quality

Trajectory quality metrics for robotics policy evaluation.

This module provides metrics for evaluating the quality of robot trajectories, including path length, smoothness, curvature change, trajectory errors, and dynamic time warping metrics for temporally misaligned trajectories.

class robometric_frame.trajectory_quality.AbsoluteTrajectoryError(**kwargs)[source]

Compute Absolute Trajectory Error (ATE) for robotics policy trajectory evaluation.

ATE is calculated as:

\[ATE = \frac{1}{L} \sum_{i=1}^{L} \|\mathbf{p}_i - \mathbf{p}_i^*\|_2\]

where \(\mathbf{p}_i\) are predicted trajectory points, \(\mathbf{p}_i^*\) are reference (ground truth) trajectory points, and \(L\) is the trajectory length. ATE evaluates global consistency by measuring the average Euclidean distance between corresponding points in predicted and reference trajectories.

This metric is critical for navigation and manipulation tasks requiring precise positioning. Lower ATE values indicate better trajectory tracking performance.

This metric accumulates errors across multiple trajectory pairs and returns the average ATE when compute() is called.

Parameters:: **kwargs (Any) – Additional keyword arguments passed to the base Metric class.

Example

>>> from robometric_frame.trajectory_quality import AbsoluteTrajectoryError
>>> import torch
>>> metric = AbsoluteTrajectoryError()
>>> # Perfect prediction (zero error)
>>> predicted = torch.tensor([[0.0, 0.0], [1.0, 0.0], [2.0, 0.0]])
>>> reference = torch.tensor([[0.0, 0.0], [1.0, 0.0], [2.0, 0.0]])
>>> metric.update(predicted, reference)
>>> metric.compute()
tensor(0.0000)

Example (with error):

>>> # Prediction with constant offset
>>> metric = AbsoluteTrajectoryError()
>>> predicted = torch.tensor([[0.0, 0.0], [1.0, 0.0], [2.0, 0.0]])
>>> reference = torch.tensor([[0.0, 1.0], [1.0, 1.0], [2.0, 1.0]])
>>> metric.update(predicted, reference)
>>> metric.compute()
tensor(1.0000)

Example (batched):

>>> # Batch of trajectory pairs - shape (B, L, D)
>>> metric = AbsoluteTrajectoryError()
>>> predicted_batch = torch.tensor([
...     [[0.0, 0.0], [1.0, 0.0], [2.0, 0.0]],
...     [[0.0, 0.0], [0.0, 1.0], [0.0, 2.0]]
... ])
>>> reference_batch = torch.tensor([
...     [[0.0, 0.5], [1.0, 0.5], [2.0, 0.5]],
...     [[0.0, 0.0], [0.0, 1.0], [0.0, 2.0]]
... ])
>>> metric.update(predicted_batch, reference_batch)
>>> result = metric.compute()

Example (3D trajectories):

>>> # 3D trajectory comparison
>>> metric = AbsoluteTrajectoryError()
>>> predicted = torch.tensor([
...     [0.0, 0.0, 0.0],
...     [1.0, 0.0, 0.0],
...     [1.0, 1.0, 0.0]
... ])
>>> reference = torch.tensor([
...     [0.0, 0.0, 0.0],
...     [1.0, 0.0, 0.0],
...     [1.0, 1.0, 1.0]
... ])
>>> metric.update(predicted, reference)
>>> result = metric.compute()

Example (distributed):

>>> # In distributed training, metrics are automatically synced
>>> metric = AbsoluteTrajectoryError()
>>> # On GPU 0
>>> pred_gpu0 = torch.tensor([[0.0, 0.0], [1.0, 0.0]])
>>> ref_gpu0 = torch.tensor([[0.0, 0.0], [1.0, 0.0]])
>>> metric.update(pred_gpu0, ref_gpu0)
>>> # On GPU 1
>>> pred_gpu1 = torch.tensor([[0.0, 0.0], [1.0, 1.0]])
>>> ref_gpu1 = torch.tensor([[0.0, 0.0], [1.0, 0.0]])
>>> metric.update(pred_gpu1, ref_gpu1)
>>> # Final result aggregates across all GPUs
>>> result = metric.compute()

full_state_update: bool = False

total_error: Tensor

num_trajectories: Tensor

__init__(**kwargs)[source]

Initialize the AbsoluteTrajectoryError metric.

update(predicted, reference)[source]

Update metric state with new predicted and reference trajectory pair(s).

Parameters:

predicted (Tensor) –
Predicted trajectory tensor of shape (…, L, D) where: - … represents any number of batch dimensions (can be empty) - L is the number of points (must be >= 1) - D is the spatial dimensionality (e.g., 2 for 2D, 3 for 3D)

Examples of valid shapes: - (L, D): Single trajectory - (B, L, D): Batch of B trajectories - (B, T, L, D): Batch of B sequences with T slices each

Points should be ordered chronologically along the L dimension.
reference (Tensor) – Reference (ground truth) trajectory tensor with the same shape as predicted.

Raises:

ValueError – If trajectories have invalid shape, mismatched shapes, or insufficient points.

Return type:

None

compute()[source]

Compute the average Absolute Trajectory Error across all trajectory pairs.

Return type:: Tensor
Returns:: Average ATE as a scalar tensor. Lower values indicate better trajectory tracking performance.
Raises:: RuntimeError – If no trajectories have been recorded.

class robometric_frame.trajectory_quality.CurvatureChange(**kwargs)[source]

Compute Curvature Change for robotics policy trajectory evaluation.

Curvature Change is calculated as:

\[CC = \frac{1}{L-2} \sum_{i=1}^{L-2} |\kappa_{i+1} - \kappa_i|, \quad \kappa_i = \frac{\theta_{i+1} - \theta_i}{\|\mathbf{p}_{i+1} - \mathbf{p}_i\|_2}\]

where \(\mathbf{p}_i\) are trajectory positions, \(\theta_i\) are orientations (heading angles), and \(\kappa_i\) is the curvature at segment \(i\). Unlike path smoothness, this metric incorporates angular velocity and is particularly useful for evaluating car-like mobile robots where curvature relates to turning radius constraints.

This metric accumulates curvature change values across multiple trajectories and returns the average when compute() is called.

Parameters:: **kwargs (Any) – Additional keyword arguments passed to the base Metric class.

Example

>>> from robometric_frame.trajectory_quality import CurvatureChange
>>> import torch
>>> metric = CurvatureChange()
>>> # Straight line motion (constant orientation)
>>> positions = torch.tensor([[0.0, 0.0], [1.0, 0.0], [2.0, 0.0], [3.0, 0.0]])
>>> orientations = torch.tensor([0.0, 0.0, 0.0, 0.0])
>>> metric.update(positions, orientations)
>>> metric.compute()
tensor(0.0000)

Example (with turn):

>>> # Path with a turn
>>> metric = CurvatureChange()
>>> positions = torch.tensor([
...     [0.0, 0.0],
...     [1.0, 0.0],
...     [2.0, 0.0],
...     [3.0, 1.0]
... ])
>>> # Orientations change from 0 to π/4 radians
>>> orientations = torch.tensor([0.0, 0.0, 0.0, 0.785])
>>> metric.update(positions, orientations)
>>> result = metric.compute()

Example (batched):

>>> # Batch of trajectory pairs - shape (B, L, D)
>>> metric = CurvatureChange()
>>> positions_batch = torch.tensor([
...     [[0.0, 0.0], [1.0, 0.0], [2.0, 0.0], [3.0, 0.0]],
...     [[0.0, 0.0], [0.0, 1.0], [0.0, 2.0], [0.0, 3.0]]
... ])
>>> orientations_batch = torch.tensor([
...     [0.0, 0.0, 0.0, 0.0],
...     [1.57, 1.57, 1.57, 1.57]
... ])
>>> metric.update(positions_batch, orientations_batch)
>>> result = metric.compute()

Example (circular path):

>>> # Circular motion with constant curvature
>>> metric = CurvatureChange()
>>> import math
>>> angles = torch.linspace(0, math.pi/2, 10)
>>> positions = torch.stack([torch.cos(angles), torch.sin(angles)], dim=1)
>>> orientations = angles + math.pi/2  # Tangent direction
>>> metric.update(positions, orientations)
>>> result = metric.compute()  # Should be small for smooth circular motion

Example (distributed):

>>> # In distributed training, metrics are automatically synced
>>> metric = CurvatureChange()
>>> # On GPU 0
>>> pos_gpu0 = torch.tensor([[0.0, 0.0], [1.0, 0.0], [2.0, 0.0]])
>>> ori_gpu0 = torch.tensor([0.0, 0.0, 0.0])
>>> metric.update(pos_gpu0, ori_gpu0)
>>> # On GPU 1
>>> pos_gpu1 = torch.tensor([[0.0, 0.0], [0.0, 1.0], [0.0, 2.0]])
>>> ori_gpu1 = torch.tensor([1.57, 1.57, 1.57])
>>> metric.update(pos_gpu1, ori_gpu1)
>>> # Final result aggregates across all GPUs
>>> result = metric.compute()

full_state_update: bool = False

total_curvature_change: Tensor

num_trajectories: Tensor

__init__(**kwargs)[source]

Initialize the CurvatureChange metric.

update(positions, orientations)[source]

Update metric state with new trajectory or batch of trajectories.

Parameters:

positions (Tensor) –
Position trajectory tensor of shape (…, L, D) where: - … represents any number of batch dimensions (can be empty) - L is the number of points (must be >= 3) - D is the spatial dimensionality (typically 2 for mobile robots)

Examples of valid shapes: - (L, D): Single trajectory - (B, L, D): Batch of B trajectories - (B, T, L, D): Batch of B sequences with T slices each

Points should be ordered chronologically along the L dimension.
orientations (Tensor) –
Orientation (heading angle) tensor of shape (…, L) where: - … represents the same batch dimensions as positions - L is the number of points (must match positions) - Values are heading angles in radians

Must have the same batch dimensions and L as positions, but without the spatial dimension D.

Raises:

ValueError – If trajectories have invalid shape, mismatched shapes, or insufficient points.

Return type:

None

compute()[source]

Compute the average Curvature Change across all trajectories.

Return type:: Tensor
Returns:: Average curvature change as a scalar tensor. Lower values indicate smoother trajectories with more consistent turning behavior.
Raises:: RuntimeError – If no trajectories have been recorded.

class robometric_frame.trajectory_quality.DTWBase(**kwargs)[source]

Base class for DTW-based trajectory metrics.

Provides shared validation, DTW computation, path length computation, and normalization factor logic used by DTWDistance, NormalizedDTW, and SuccessWeightedDTW.

Subclasses must define their own metric states and implement update()/compute().

is_differentiable: bool = False

full_state_update: bool = False

num_trajectory_pairs: Tensor

__init__(**kwargs)[source]

Initialize internal Module state, shared by both nn.Module and ScriptModule.

class robometric_frame.trajectory_quality.DTWDistance(**kwargs)[source]

Compute Dynamic Time Warping (DTW) distance for trajectory evaluation.

DTW distance measures the minimum-cost temporal alignment between predicted and reference trajectories. Unlike MSE which compares trajectories timestep-by- timestep, DTW finds the optimal warping to align sequences that may differ in length or timing.

DTW is calculated by building an accumulated cost matrix D where:: D[0,0] = C[0,0] D[i,0] = D[i-1,0] + C[i,0] for i > 0 D[0,j] = D[0,j-1] + C[0,j] for j > 0 D[i,j] = C[i,j] + min(D[i-1,j], D[i,j-1], D[i-1,j-1]) for i,j > 0

where C[i,j] is the Euclidean distance between predicted[i] and reference[j]. The final DTW distance is D[n-1, m-1].

This metric is particularly useful for evaluating VLA models and policies using action chunking (e.g., ACT, Diffusion Policy) where predicted trajectories may be temporally misaligned with demonstrations.

This metric accumulates DTW distances across multiple trajectory pairs and returns the average DTW distance when compute() is called.

Parameters:: **kwargs (Any) – Additional keyword arguments passed to the base Metric class.

higher_is_better: False - lower DTW distance indicates better similarity.

is_differentiable: False - DTW computation is not differentiable.

full_state_update: False - incremental state updates.

Note

Memory complexity is O(T_pred * T_ref) for the cost matrices. For very long trajectories, this may require significant memory.

Example

>>> from robometric_frame.trajectory_quality import DTWDistance
>>> import torch
>>> metric = DTWDistance()
>>> # Identical trajectories (zero distance)
>>> predicted = torch.tensor([[0.0, 0.0], [1.0, 0.0], [2.0, 0.0]])
>>> reference = torch.tensor([[0.0, 0.0], [1.0, 0.0], [2.0, 0.0]])
>>> metric.update(predicted, reference)
>>> metric.compute()
tensor(0.0000)

Example (different lengths):

>>> # Trajectories of different lengths (the core use case)
>>> metric = DTWDistance()
>>> reference = torch.tensor([[0.0, 0.0], [1.0, 0.0], [2.0, 0.0], [3.0, 0.0]])
>>> predicted = torch.tensor([[0.0, 0.0], [0.5, 0.0], [1.0, 0.0], [1.5, 0.0],
...                           [2.0, 0.0], [2.5, 0.0], [3.0, 0.0]])
>>> metric.update(predicted, reference)
>>> result = metric.compute()  # Small value (same path, different density)

Example (temporal shift):

>>> # Hesitation at start (same actions, different timing)
>>> metric = DTWDistance()
>>> reference = torch.tensor([[0.0, 0.0], [1.0, 0.0], [2.0, 0.0]])
>>> predicted = torch.tensor([[0.0, 0.0], [0.0, 0.0], [0.0, 0.0],
...                           [1.0, 0.0], [2.0, 0.0]])
>>> metric.update(predicted, reference)
>>> result = metric.compute()  # Small value (DTW tolerates hesitation)

higher_is_better: bool = False

total_dtw_distance: Tensor

__init__(**kwargs)[source]

Initialize the DTWDistance metric.

update(predicted, reference)[source]

Update metric state with new predicted and reference trajectory pair.

Parameters:

predicted (Tensor) –
Predicted trajectory tensor of shape (T_pred, D) where: - T_pred is the number of timesteps (can differ from T_ref) - D is the spatial dimensionality (e.g., 2 for 2D, 3 for 3D, 7 for 7-DoF)

Points should be ordered chronologically.
reference (Tensor) – Reference (ground truth) trajectory tensor of shape (T_ref, D). T_ref can differ from T_pred - this is the core advantage of DTW. D must match predicted trajectory dimensionality.

Raises:

ValueError – If trajectories have invalid shape (< 2 dimensions), mismatched dimensionality, or insufficient points.

Return type:

None

compute()[source]

Compute the average DTW distance across all trajectory pairs.

Return type:: Tensor
Returns:: Average DTW distance as a scalar tensor. Lower values indicate better trajectory similarity.
Raises:: RuntimeError – If no trajectory pairs have been recorded.

class robometric_frame.trajectory_quality.NormalizedDTW(normalization_factor=None, **kwargs)[source]

Compute Normalized DTW (nDTW) for trajectory evaluation.

nDTW normalizes the raw DTW distance and maps it to a [0, 1] score:

nDTW = exp(-DTW / (|R| * d))

where:

DTW is the raw DTW distance
|R| is the length of the reference trajectory (number of points)
d is a normalization constant (average step distance of the reference, or a user-specified value)

Higher nDTW scores indicate better trajectory similarity (1.0 = perfect match, approaches 0.0 for very dissimilar trajectories).

This metric is particularly useful for evaluating VLA models and policies using action chunking where predicted trajectories may be temporally misaligned.

Parameters:

normalization_factor (Optional[float]) – Optional user-specified normalization constant d. If None (default), automatically computed as the mean step distance of the reference trajectory: PathLength(reference) / (len(reference) - 1).
**kwargs (Any) – Additional keyword arguments passed to the base Metric class.

higher_is_better: True - higher nDTW indicates better similarity.

is_differentiable: False - DTW computation is not differentiable.

full_state_update: False - incremental state updates.

Note

Memory complexity is O(T_pred * T_ref) for the cost matrices. For very long trajectories, this may require significant memory.

Example

>>> from robometric_frame.trajectory_quality import NormalizedDTW
>>> import torch
>>> metric = NormalizedDTW()
>>> # Identical trajectories (perfect score)
>>> predicted = torch.tensor([[0.0, 0.0], [1.0, 0.0], [2.0, 0.0]])
>>> reference = torch.tensor([[0.0, 0.0], [1.0, 0.0], [2.0, 0.0]])
>>> metric.update(predicted, reference)
>>> metric.compute()
tensor(1.0000)

Example (different lengths):

>>> # Trajectories of different lengths
>>> metric = NormalizedDTW()
>>> reference = torch.tensor([[0.0, 0.0], [1.0, 0.0], [2.0, 0.0], [3.0, 0.0]])
>>> predicted = torch.tensor([[0.0, 0.0], [0.5, 0.0], [1.0, 0.0], [1.5, 0.0],
...                           [2.0, 0.0], [2.5, 0.0], [3.0, 0.0]])
>>> metric.update(predicted, reference)
>>> result = metric.compute()  # High value (same path)

Example (custom normalization):

>>> # Use custom normalization factor
>>> metric = NormalizedDTW(normalization_factor=0.5)
>>> reference = torch.tensor([[0.0, 0.0], [1.0, 0.0], [2.0, 0.0]])
>>> predicted = torch.tensor([[0.0, 0.1], [1.0, 0.1], [2.0, 0.1]])
>>> metric.update(predicted, reference)
>>> result = metric.compute()

higher_is_better: bool = True

total_ndtw: Tensor

__init__(normalization_factor=None, **kwargs)[source]

Initialize the NormalizedDTW metric.

Parameters:

normalization_factor (Optional[float]) – Optional user-specified normalization constant d. If None, automatically computed as the mean step distance of the reference trajectory.
**kwargs (Any) – Additional keyword arguments passed to the base Metric class.

update(predicted, reference)[source]

Update metric state with new predicted and reference trajectory pair.

Parameters:

predicted (Tensor) –
Predicted trajectory tensor of shape (T_pred, D) where: - T_pred is the number of timesteps (can differ from T_ref) - D is the spatial dimensionality

Points should be ordered chronologically.
reference (Tensor) – Reference (ground truth) trajectory tensor of shape (T_ref, D). T_ref can differ from T_pred. D must match predicted dimensionality.

Raises:

ValueError – If trajectories have invalid shape, mismatched dimensionality, or insufficient points.

Return type:

None

compute()[source]

Compute the average nDTW score across all trajectory pairs.

Return type:: Tensor
Returns:: Average nDTW score as a scalar tensor in range [0, 1]. Higher values indicate better trajectory similarity.
Raises:: RuntimeError – If no trajectory pairs have been recorded.

class robometric_frame.trajectory_quality.PathLength(**kwargs)[source]

Compute Path Length for robotics policy trajectory evaluation.

Path Length is calculated as:

\[PL = \sum_{i=1}^{L-1} \|\mathbf{p}_{i+1} - \mathbf{p}_i\|_2\]

where \(\mathbf{p}_i\) are trajectory points in D-dimensional space and \(L\) is the length of the trajectory. Shorter paths generally indicate more efficient task execution.

This metric accumulates path lengths across multiple trajectories and returns the average path length when compute() is called.

Parameters:: **kwargs (Any) – Additional keyword arguments passed to the base Metric class.

Example

>>> from robometric_frame.trajectory_quality import PathLength
>>> import torch
>>> metric = PathLength()
>>> # 2D trajectory with 5 points
>>> trajectory = torch.tensor([
...     [0.0, 0.0],
...     [1.0, 0.0],
...     [1.0, 1.0],
...     [2.0, 1.0],
...     [2.0, 2.0]
... ])
>>> metric.update(trajectory)
>>> metric.compute()
tensor(4.0000)

Example (batched):

>>> # Batch of trajectories - shape (B, L, D)
>>> metric = PathLength()
>>> batch = torch.tensor([
...     [[0.0, 0.0], [1.0, 0.0], [2.0, 0.0]],  # trajectory 1
...     [[0.0, 0.0], [0.0, 1.0], [0.0, 2.0]]   # trajectory 2
... ])
>>> metric.update(batch)
>>> metric.compute()  # Average of 2.0 and 2.0
tensor(2.0000)

Example (3D trajectories):

>>> # 3D trajectory
>>> metric = PathLength()
>>> trajectory_3d = torch.tensor([
...     [0.0, 0.0, 0.0],
...     [1.0, 0.0, 0.0],
...     [1.0, 1.0, 0.0],
...     [1.0, 1.0, 1.0]
... ])
>>> metric.update(trajectory_3d)
>>> metric.compute()
tensor(3.0000)

Example (distributed):

>>> # In distributed training, metrics are automatically synced
>>> metric = PathLength()
>>> # On GPU 0
>>> traj_gpu0 = torch.tensor([[0.0, 0.0], [1.0, 0.0]])
>>> metric.update(traj_gpu0)
>>> # On GPU 1
>>> traj_gpu1 = torch.tensor([[0.0, 0.0], [0.0, 1.0]])
>>> metric.update(traj_gpu1)
>>> # Final result aggregates across all GPUs
>>> result = metric.compute()  # Returns aggregated average path length

full_state_update: bool = False

total_path_length: Tensor

num_trajectories: Tensor

__init__(**kwargs)[source]

Initialize the PathLength metric.

update(trajectory)[source]

Update metric state with new trajectory or batch of trajectories.

Parameters:

trajectory (Tensor) –

Tensor of shape (…, L, D) where: - … represents any number of batch dimensions (can be empty) - L is the number of points (must be >= 2) - D is the spatial dimensionality (e.g., 2 for 2D, 3 for 3D)

Examples of valid shapes: - (L, D): Single trajectory - (B, L, D): Batch of B trajectories - (B, T, L, D): Batch of B sequences with T slices each

Points should be ordered chronologically along the L dimension.

Raises:

ValueError – If trajectory has invalid shape or insufficient points.

Return type:

None

compute()[source]

Compute the average Path Length across all trajectories.

Return type:: Tensor
Returns:: Average path length as a scalar tensor.
Raises:: RuntimeError – If no trajectories have been recorded.

class robometric_frame.trajectory_quality.PathSmoothness(**kwargs)[source]

Compute Path Smoothness for robotics policy trajectory evaluation.

Path Smoothness is calculated as:

\[PS = \frac{1}{L-2} \sum_{i=1}^{L-2} \|(\mathbf{p}_{i+2} - \mathbf{p}_{i+1}) - (\mathbf{p}_{i+1} - \mathbf{p}_i)\|_2\]

where \(\mathbf{p}_i\) are trajectory points in D-dimensional space, \(L\) is the length of the trajectory, and \(L-2\) is the number of direction-change samples. This metric measures the rate of change in trajectory direction, with lower values indicating smoother paths.

The metric calculates the difference between consecutive displacement vectors, effectively measuring the second derivative (acceleration) of the path. It is normalized by the number of direction-change samples (L-2) to decouple smoothness from path length / trajectory scale.

This metric accumulates smoothness values across multiple trajectories and returns the average path smoothness when compute() is called.

Parameters:: **kwargs (Any) – Additional keyword arguments passed to the base Metric class.

Example

>>> from robometric_frame.trajectory_quality import PathSmoothness
>>> import torch
>>> metric = PathSmoothness()
>>> # Smooth straight line (perfect smoothness = 0)
>>> trajectory = torch.tensor([
...     [0.0, 0.0],
...     [1.0, 0.0],
...     [2.0, 0.0],
...     [3.0, 0.0]
... ])
>>> metric.update(trajectory)
>>> metric.compute()
tensor(0.0000)

Example (with direction change):

>>> # Path with a turn (higher smoothness value)
>>> metric = PathSmoothness()
>>> trajectory = torch.tensor([
...     [0.0, 0.0],
...     [1.0, 0.0],
...     [2.0, 0.0],
...     [2.0, 1.0]
... ])
>>> metric.update(trajectory)
>>> result = metric.compute()
>>> result > 0  # Non-zero smoothness due to direction change
tensor(True)

Example (batched):

>>> # Batch of trajectories - shape (B, L, D)
>>> metric = PathSmoothness()
>>> batch = torch.tensor([
...     [[0.0, 0.0], [1.0, 0.0], [2.0, 0.0], [3.0, 0.0]],  # smooth
...     [[0.0, 0.0], [1.0, 0.0], [1.0, 1.0], [2.0, 1.0]]   # has turn
... ])
>>> metric.update(batch)
>>> result = metric.compute()  # Average smoothness

Example (3D trajectories):

>>> # 3D trajectory
>>> metric = PathSmoothness()
>>> trajectory_3d = torch.tensor([
...     [0.0, 0.0, 0.0],
...     [1.0, 0.0, 0.0],
...     [2.0, 0.0, 0.0],
...     [3.0, 0.0, 0.0]
... ])
>>> metric.update(trajectory_3d)
>>> metric.compute()  # Perfect smoothness for straight line
tensor(0.0000)

Example (distributed):

>>> # In distributed training, metrics are automatically synced
>>> metric = PathSmoothness()
>>> # On GPU 0
>>> traj_gpu0 = torch.tensor([[0.0, 0.0], [1.0, 0.0], [2.0, 0.0]])
>>> metric.update(traj_gpu0)
>>> # On GPU 1
>>> traj_gpu1 = torch.tensor([[0.0, 0.0], [0.0, 1.0], [0.0, 2.0]])
>>> metric.update(traj_gpu1)
>>> # Final result aggregates across all GPUs
>>> result = metric.compute()  # Returns aggregated average smoothness

full_state_update: bool = False

total_smoothness: Tensor

num_trajectories: Tensor

__init__(**kwargs)[source]

Initialize the PathSmoothness metric.

update(trajectory)[source]

Update metric state with new trajectory or batch of trajectories.

Parameters:

trajectory (Tensor) –

Tensor of shape (…, L, D) where: - … represents any number of batch dimensions (can be empty) - L is the number of points (must be >= 3) - D is the spatial dimensionality (e.g., 2 for 2D, 3 for 3D)

Examples of valid shapes: - (L, D): Single trajectory - (B, L, D): Batch of B trajectories - (B, T, L, D): Batch of B sequences with T slices each

Points should be ordered chronologically along the L dimension.

Raises:

ValueError – If trajectory has invalid shape or insufficient points.

Return type:

None

compute()[source]

Compute the average Path Smoothness across all trajectories.

Return type:: Tensor
Returns:: Average path smoothness as a scalar tensor. Lower values indicate smoother trajectories with less direction changes.
Raises:: RuntimeError – If no trajectories have been recorded.

class robometric_frame.trajectory_quality.RelativeTrajectoryError(delta=1, **kwargs)[source]

Compute Relative Trajectory Error (RTE) for robotics policy trajectory evaluation.

RTE is calculated as:

\[RTE = \frac{1}{L-\Delta} \sum_{i=1}^{L-\Delta} \|(\mathbf{p}_{i+\Delta} - \mathbf{p}_i) - (\mathbf{p}_{i+\Delta}^* - \mathbf{p}_i^*)\|_2\]

where \(\mathbf{p}_i\) are predicted trajectory points, \(\mathbf{p}_i^*\) are reference (ground truth) trajectory points, \(L\) is the trajectory length, and \(\Delta\) (delta) is the step size for computing relative motion.

RTE assesses local accuracy by comparing displacement vectors between the predicted and reference trajectories. Unlike ATE which measures global consistency, RTE focuses on the correctness of relative motion, making it particularly useful for evaluating drift and local tracking performance.

This metric accumulates errors across multiple trajectory pairs and returns the average RTE when compute() is called.

Parameters:

delta (int) – Step size for computing relative motion. Must be >= 1. Larger values assess consistency over longer time windows. Default: 1 (consecutive points).
**kwargs (Any) – Additional keyword arguments passed to the base Metric class.

Example

>>> from robometric_frame.trajectory_quality import RelativeTrajectoryError
>>> import torch
>>> metric = RelativeTrajectoryError(delta=1)
>>> # Perfect prediction (zero error)
>>> predicted = torch.tensor([[0.0, 0.0], [1.0, 0.0], [2.0, 0.0]])
>>> reference = torch.tensor([[0.0, 0.0], [1.0, 0.0], [2.0, 0.0]])
>>> metric.update(predicted, reference)
>>> metric.compute()
tensor(0.0000)

Example (with drift):

>>> # Prediction with constant drift in motion
>>> metric = RelativeTrajectoryError(delta=1)
>>> predicted = torch.tensor([[0.0, 0.0], [1.0, 0.0], [2.0, 0.5]])
>>> reference = torch.tensor([[0.0, 0.0], [1.0, 0.0], [2.0, 0.0]])
>>> metric.update(predicted, reference)
>>> result = metric.compute()

Example (larger delta):

>>> # Using delta=2 to check motion over 2-step windows
>>> metric = RelativeTrajectoryError(delta=2)
>>> predicted = torch.tensor([
...     [0.0, 0.0], [1.0, 0.0], [2.0, 0.0], [3.0, 0.0]
... ])
>>> reference = torch.tensor([
...     [0.0, 0.0], [1.0, 0.0], [2.0, 0.0], [3.0, 0.0]
... ])
>>> metric.update(predicted, reference)
>>> metric.compute()
tensor(0.0000)

Example (batched):

>>> # Batch of trajectory pairs - shape (B, L, D)
>>> metric = RelativeTrajectoryError(delta=1)
>>> predicted_batch = torch.tensor([
...     [[0.0, 0.0], [1.0, 0.0], [2.0, 0.0]],
...     [[0.0, 0.0], [0.0, 1.0], [0.0, 2.0]]
... ])
>>> reference_batch = torch.tensor([
...     [[0.0, 0.0], [1.0, 0.0], [2.0, 0.0]],
...     [[0.0, 0.0], [0.0, 1.0], [0.0, 2.0]]
... ])
>>> metric.update(predicted_batch, reference_batch)
>>> metric.compute()
tensor(0.0000)

Example (3D trajectories):

>>> # 3D trajectory comparison
>>> metric = RelativeTrajectoryError(delta=1)
>>> predicted = torch.tensor([
...     [0.0, 0.0, 0.0],
...     [1.0, 0.0, 0.0],
...     [1.0, 1.0, 0.0]
... ])
>>> reference = torch.tensor([
...     [0.0, 0.0, 0.0],
...     [1.0, 0.0, 0.0],
...     [1.0, 1.0, 1.0]
... ])
>>> metric.update(predicted, reference)
>>> result = metric.compute()

Example (distributed):

>>> # In distributed training, metrics are automatically synced
>>> metric = RelativeTrajectoryError(delta=1)
>>> # On GPU 0
>>> pred_gpu0 = torch.tensor([[0.0, 0.0], [1.0, 0.0], [2.0, 0.0]])
>>> ref_gpu0 = torch.tensor([[0.0, 0.0], [1.0, 0.0], [2.0, 0.0]])
>>> metric.update(pred_gpu0, ref_gpu0)
>>> # On GPU 1
>>> pred_gpu1 = torch.tensor([[0.0, 0.0], [1.0, 1.0], [2.0, 2.0]])
>>> ref_gpu1 = torch.tensor([[0.0, 0.0], [1.0, 0.0], [2.0, 0.0]])
>>> metric.update(pred_gpu1, ref_gpu1)
>>> # Final result aggregates across all GPUs
>>> result = metric.compute()

full_state_update: bool = False

total_error: Tensor

num_trajectories: Tensor

__init__(delta=1, **kwargs)[source]

Initialize the RelativeTrajectoryError metric.

Parameters:

delta (int) – Step size for computing relative motion. Must be >= 1.
**kwargs (Any) – Additional keyword arguments passed to the base Metric class.

Raises:

ValueError – If delta is less than 1.

delta: int

update(predicted, reference)[source]

Update metric state with new predicted and reference trajectory pair(s).

Parameters:

predicted (Tensor) –
Predicted trajectory tensor of shape (…, L, D) where: - … represents any number of batch dimensions (can be empty) - L is the number of points (must be > delta) - D is the spatial dimensionality (e.g., 2 for 2D, 3 for 3D)

Examples of valid shapes: - (L, D): Single trajectory - (B, L, D): Batch of B trajectories - (B, T, L, D): Batch of B sequences with T slices each

Points should be ordered chronologically along the L dimension.
reference (Tensor) – Reference (ground truth) trajectory tensor with the same shape as predicted.

Raises:

ValueError – If trajectories have invalid shape, mismatched shapes, or insufficient points.

Return type:

None

compute()[source]

Compute the average Relative Trajectory Error across all trajectory pairs.

Return type:: Tensor
Returns:: Average RTE as a scalar tensor. Lower values indicate better local tracking performance and less drift.
Raises:: RuntimeError – If no trajectories have been recorded.

class robometric_frame.trajectory_quality.SuccessWeightedDTW(normalization_factor=None, **kwargs)[source]

Compute Success-weighted DTW (SDTW) for trajectory evaluation.

SDTW combines trajectory fidelity with task success:

SDTW = nDTW * Success

where:

nDTW is the normalized DTW score (see NormalizedDTW)
Success is a binary indicator (1 if task succeeded, 0 if not)

If the task failed, SDTW = 0 regardless of trajectory similarity. This captures both “did you succeed?” and “did you follow the right path?”

This metric is particularly useful for benchmarking policies where both task completion and trajectory quality matter.

Parameters:

normalization_factor (Optional[float]) – Optional user-specified normalization constant d. If None (default), automatically computed as the mean step distance of the reference trajectory.
**kwargs (Any) – Additional keyword arguments passed to the base Metric class.

higher_is_better: True - higher SDTW indicates better performance.

is_differentiable: False - DTW computation is not differentiable.

full_state_update: False - incremental state updates.

Note

Memory complexity is O(T_pred * T_ref) for the cost matrices. For very long trajectories, this may require significant memory.

Example

>>> from robometric_frame.trajectory_quality import SuccessWeightedDTW
>>> import torch
>>> metric = SuccessWeightedDTW()
>>> # Successful task with good trajectory
>>> predicted = torch.tensor([[0.0, 0.0], [1.0, 0.0], [2.0, 0.0]])
>>> reference = torch.tensor([[0.0, 0.0], [1.0, 0.0], [2.0, 0.0]])
>>> metric.update(predicted, reference, success=torch.tensor(True))
>>> metric.compute()
tensor(1.0000)

Example (failed task):

>>> # Failed task (SDTW = 0 regardless of trajectory quality)
>>> metric = SuccessWeightedDTW()
>>> predicted = torch.tensor([[0.0, 0.0], [1.0, 0.0], [2.0, 0.0]])
>>> reference = torch.tensor([[0.0, 0.0], [1.0, 0.0], [2.0, 0.0]])
>>> metric.update(predicted, reference, success=torch.tensor(False))
>>> metric.compute()
tensor(0.0000)

Example (multiple updates):

>>> # Mix of successes and failures
>>> metric = SuccessWeightedDTW()
>>> ref = torch.tensor([[0.0, 0.0], [1.0, 0.0], [2.0, 0.0]])
>>> pred = torch.tensor([[0.0, 0.0], [1.0, 0.0], [2.0, 0.0]])
>>> metric.update(pred, ref, success=torch.tensor(True))   # SDTW = 1.0
>>> metric.update(pred, ref, success=torch.tensor(False))  # SDTW = 0.0
>>> metric.compute()  # Average: 0.5
tensor(0.5000)

higher_is_better: bool = True

total_sdtw: Tensor

__init__(normalization_factor=None, **kwargs)[source]

Initialize the SuccessWeightedDTW metric.

Parameters:

normalization_factor (Optional[float]) – Optional user-specified normalization constant d. If None, automatically computed as the mean step distance of the reference trajectory.
**kwargs (Any) – Additional keyword arguments passed to the base Metric class.

update(predicted, reference, success)[source]

Update metric state with new trajectory pair and success indicator.

Parameters:

predicted (Tensor) –
Predicted trajectory tensor of shape (T_pred, D) where: - T_pred is the number of timesteps (can differ from T_ref) - D is the spatial dimensionality

Points should be ordered chronologically.
reference (Tensor) – Reference (ground truth) trajectory tensor of shape (T_ref, D). T_ref can differ from T_pred. D must match predicted dimensionality.
success (Tensor) – Boolean or 0/1 tensor indicating task success. If False/0, SDTW will be 0 regardless of trajectory similarity.

Raises:

ValueError – If trajectories have invalid shape, mismatched dimensionality, or insufficient points.

Return type:

None

compute()[source]

Compute the average SDTW score across all trajectory pairs.

Return type:: Tensor
Returns:: Average SDTW score as a scalar tensor in range [0, 1]. Higher values indicate better overall performance (trajectory quality weighted by task success).
Raises:: RuntimeError – If no trajectory pairs have been recorded.

Modules

`absolute_trajectory_error`	Absolute Trajectory Error (ATE) metric for robotics policy trajectory evaluation.
`curvature_change`	Curvature Change metric for robotics policy trajectory evaluation.
`dtw`	Dynamic Time Warping (DTW) metrics for robotics policy trajectory evaluation.
`path_length`	Path Length metric for robotics policy trajectory evaluation.
`path_smoothness`	Path Smoothness metric for robotics policy trajectory evaluation.
`relative_trajectory_error`	Relative Trajectory Error (RTE) metric for robotics policy trajectory evaluation.