# ----------------------------------------------------------------------------
# Copyright (c) 2021-2026 DexForce Technology Co., Ltd.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
# ----------------------------------------------------------------------------
"""Typed records shared by the motion-generation benchmark modules."""
from __future__ import annotations
import math
from dataclasses import asdict, dataclass, field, replace
from enum import Enum
from typing import cast
import torch
from .contracts import EvaluationDomain, PhysicalEvaluation, StageOutcome
__all__ = [
"AlgorithmRole",
"BenchmarkCase",
"CaseOutcome",
"PlannerMetadata",
"TrialPhase",
"TrialRecord",
"with_physical_evaluation",
]
def _json_value(value: object) -> object:
"""Convert nested benchmark values into deterministic JSON values."""
if isinstance(value, float) and not math.isfinite(value):
return None
if isinstance(value, Enum):
return value.value
if isinstance(value, frozenset):
return sorted(_json_value(item) for item in value)
if isinstance(value, tuple):
return [_json_value(item) for item in value]
if isinstance(value, list):
return [_json_value(item) for item in value]
if isinstance(value, dict):
return {str(key): _json_value(item) for key, item in value.items()}
return value
[docs]
class AlgorithmRole(str, Enum):
"""Role an algorithm plays in benchmark comparisons."""
CANDIDATE = "candidate"
PRIMARY_BASELINE = "primary_baseline"
DIAGNOSTIC_BASELINE = "diagnostic_baseline"
[docs]
class TrialPhase(str, Enum):
"""Lifecycle phase represented by a raw trial record."""
AVAILABILITY = "availability"
CONSTRUCT = "construct"
PREPARE = "prepare"
COLD = "cold"
WARMUP = "warmup"
MEASURED = "measured"
[docs]
@dataclass(frozen=True)
class BenchmarkCase:
"""One env-batched planner input frozen before execution.
The first fields retain the free-space case contract. The trailing fields
describe execution/task tracks without forcing planner adapters to know
about a particular robot, atomic skill, or object implementation.
"""
suite_version: str
track: str
scenario_id: str
case_id: str
seed: int
batch_size: int
num_waypoints: int
path_shape: str
start_state_bin: str
start_qpos: torch.Tensor
target_waypoints: torch.Tensor
reference_qpos: torch.Tensor
robot_id: str = "franka_panda"
skill_id: str = "N/A"
object_id: str | None = None
task_difficulty: str | None = None
primary_success: str = "motion_valid"
full_start_qpos: torch.Tensor | None = None
case_parameters: dict[str, object] = field(default_factory=dict)
domain: EvaluationDomain | None = None
track_group: str | None = None
embodiment_id: str | None = None
required_capabilities: frozenset[str] = frozenset()
[docs]
@dataclass(frozen=True)
class CaseOutcome:
"""External validation result for one environment row in a batch."""
env_index: int
planning_success: bool
finite: bool
ordered_waypoints_reached: bool
motion_valid: bool
completed_waypoint_ratio: float
final_translation_err_mm: float | None
final_rotation_err_deg: float | None
waypoint_translation_err_mm_mean: float | None
waypoint_translation_err_mm_p95: float | None
waypoint_translation_err_mm_max: float | None
waypoint_rotation_err_deg_mean: float | None
waypoint_rotation_err_deg_p95: float | None
waypoint_rotation_err_deg_max: float | None
joint_limit_violation: bool
max_normalized_joint_violation: float | None
joint_path_length_rad: float | None
cartesian_path_length_m: float | None
path_efficiency: float | None
failure_code: str | None = None
planner_failure_code: str | None = None
execution_success: bool | None = None
task_success: bool | None = None
task_completion_time_s: float | None = None
joint_tracking_rmse_rad: float | None = None
object_lift_delta_m: float | None = None
articulation_joint_initial: float | None = None
articulation_joint_final: float | None = None
articulation_joint_delta: float | None = None
articulation_joint_peak_signed_delta: float | None = None
articulation_joint_effect_end: float | None = None
object_rotation_peak_signed_rad: float | None = None
replan_count: int | None = None
min_translation_err_mm: float | None = None
min_rotation_err_deg: float | None = None
trajectory_moved: bool | None = None
waypoint_min_translation_err_mm: tuple[float, ...] = ()
waypoint_min_rotation_err_deg: tuple[float, ...] = ()
waypoint_min_rotation_err_deg_at_position: tuple[float | None, ...] = ()
waypoint_min_translation_err_mm_at_orientation: tuple[float | None, ...] = ()
executed_final_translation_err_mm: float | None = None
executed_final_rotation_err_deg: float | None = None
max_joint_velocity_rad_s: float | None = None
velocity_utilization: float | None = None
velocity_limit_violation: bool | None = None
max_joint_acceleration_rad_s2: float | None = None
acceleration_utilization: float | None = None
acceleration_limit_violation: bool | None = None
max_joint_jerk_rad_s3: float | None = None
jerk_utilization: float | None = None
jerk_limit_violation: bool | None = None
dynamic_measurement_status: str | None = None
"""Whether trajectory dynamics were ``finite``, ``unbounded``, or unavailable."""
dynamic_limits_satisfied: bool | None = None
"""Whether every applicable dynamic limit holds, or ``None`` when none apply.
Acceleration and jerk limits have no asset source, so a suite that leaves
them unset yields ``None`` rather than a satisfied result.
"""
stages: tuple[StageOutcome, ...] = ()
failure_stage: str | None = None
physical_skill_success: bool | None = None
physical_failure_code: str | None = None
evaluation_protocol: str | None = None
timed_motion_valid: bool | None = None
"""Validity under solved timing; nominal or unknown clocks remain unavailable."""
constraint_metrics: dict[str, object] = field(default_factory=dict)
"""Constrained-component arrival errors; unused components are absent."""
geometry_valid: bool = False
"""Derived path validity, also accepted when reading serialized outcomes."""
start_state_valid: bool = True
"""Whether the first native sample matches the requested joint state."""
start_joint_error_rad: float | None = None
reference_path_deviation_mm_mean: float | None = None
reference_path_deviation_mm_max: float | None = None
"""Sampled translation cross-track error to the requested waypoint polyline.
Diagnostic only; ordered arrival remains the validity contract. Joint goals
and orientation-only requests have no translation reference-path metric.
"""
def __post_init__(self) -> None:
"""Recompute geometry validity from its authoritative component checks."""
object.__setattr__(
self,
"geometry_valid",
self.finite
and self.ordered_waypoints_reached
and not self.joint_limit_violation
and self.start_state_valid,
)
@property
def end_to_end_success(self) -> bool | None:
"""Return overall success; unavailable physical evaluation stays unknown."""
if self.physical_skill_success is None:
return None
return (
self.planning_success
and self.motion_valid
and self.execution_success is True
and self.physical_skill_success
)
[docs]
def with_physical_evaluation(
outcome: CaseOutcome,
evaluation: PhysicalEvaluation,
*,
skill_id: str,
) -> CaseOutcome:
"""Attach physical results without gating them on motion diagnostics.
Args:
outcome: Motion and controller diagnostics for the same environment.
evaluation: Measured physical stages for that environment.
skill_id: Canonical skill id used to validate a versioned result.
Returns:
Outcome retaining independent physical success and the legacy overall alias.
Raises:
ValueError: Environment rows or claimed standard results disagree.
"""
from .evaluation import validate_standard_evaluation
if outcome.env_index != evaluation.env_index:
raise ValueError("Physical evaluation belongs to a different environment row.")
validate_standard_evaluation(skill_id, evaluation)
updated = replace(
outcome,
physical_skill_success=evaluation.task_success,
physical_failure_code=evaluation.failure_code,
evaluation_protocol=evaluation.protocol_version,
stages=evaluation.stages,
failure_stage=evaluation.failure_stage,
failure_code=outcome.failure_code or evaluation.failure_code,
)
return replace(updated, task_success=updated.end_to_end_success)
[docs]
@dataclass(frozen=True)
class TrialRecord:
"""Raw lifecycle or planning record written to ``trials.jsonl``."""
suite_version: str
track: str
scenario_id: str
case_id: str
algorithm_id: str
algorithm_role: AlgorithmRole
model_revision: str
planner_config_hash: str
seed: int
repeat: int
batch_size: int
waypoint_count: int
path_shape: str
start_state_bin: str
phase: TrialPhase
status: str = "ok"
failure_code: str | None = None
failure_message: str | None = None
cost_time_ms: float | None = None
cpu_delta_mb: float | None = None
gpu_delta_mb: float | None = None
peak_gpu_mb: float | None = None
robot_id: str = "franka_panda"
skill_id: str = "N/A"
object_id: str | None = None
task_difficulty: str | None = None
primary_success: str = "motion_valid"
execution_time_ms: float | None = None
end_to_end_time_ms: float | None = None
trajectory_duration_s: float | None = None
trajectory_waypoints: int | None = None
metadata: dict[str, object] = field(default_factory=dict)
outcomes: tuple[CaseOutcome, ...] = ()
domain: EvaluationDomain | None = None
track_group: str | None = None
embodiment_id: str | None = None
@property
def uses_selected_planner_backend(self) -> bool:
"""Return whether this trial exercised its attributed planner backend.
Older artifacts predate explicit motion attribution, so a missing flag
retains their original planner-attributed behavior.
"""
return self.metadata.get("selected_planner_backend_used") is not False
@property
def effective_motion_algorithm(self) -> str:
"""Return the algorithm that actually produced this trial's motion.
Planner-attributed trials retain their configured planner identity.
Scenario-defined motion uses its explicit metadata identity, with a
stable fallback for malformed or older non-planner artifacts.
"""
if self.uses_selected_planner_backend:
return self.algorithm_id
value = self.metadata.get("effective_motion_algorithm")
if isinstance(value, str) and value.strip():
return value
return "unattributed_motion"
[docs]
def to_dict(self) -> dict[str, object]:
"""Return a JSON-serializable mapping while retaining numeric values."""
payload = cast(dict[str, object], _json_value(asdict(self)))
for row, outcome in zip(payload["outcomes"], self.outcomes):
row["end_to_end_success"] = outcome.end_to_end_success
return payload