"""Stable, reward-weight-independent evaluation scoring.""" from __future__ import annotations import math from collections.abc import Mapping from typing import Any DEFAULT_OBJECTIVE_WEIGHTS = { "velocity_tracking": 0.35, "action_smoothness": 0.20, "posture_stability": 0.15, "fall_avoidance": 0.15, "foot_slip": 0.10, "energy": 0.05, } REQUIRED_METRICS = { "linear_velocity_rmse", "angular_velocity_rmse", "mean_action_acc", "orientation_error", "fall_rate", "slip_velocity", "mechanical_power", } PHYSICAL_FLOORS = { "linear_velocity_rmse": 0.10, "angular_velocity_rmse": 0.10, "mean_action_acc": 0.01, "orientation_error": 0.05, "fall_rate": 0.02, "slip_velocity": 0.05, "mechanical_power": 10.0, } class EvaluationError(ValueError): pass def validate_objective_weights(value: Any) -> dict[str, float]: if not isinstance(value, Mapping) or set(value) != set(DEFAULT_OBJECTIVE_WEIGHTS): raise EvaluationError("objectiveWeights 必须完整包含六个目标") result: dict[str, float] = {} for key in DEFAULT_OBJECTIVE_WEIGHTS: raw = value[key] if isinstance(raw, bool) or not isinstance(raw, (int, float)): raise EvaluationError(f"objectiveWeights.{key} 必须是数值") number = float(raw) if not math.isfinite(number) or number < 0.0 or number > 1.0: raise EvaluationError(f"objectiveWeights.{key} 必须在 0–1 之间") result[key] = number if not math.isclose(sum(result.values()), 1.0, abs_tol=1.0e-6): raise EvaluationError("objectiveWeights 总和必须为 1") return result def validate_metrics(value: Any) -> dict[str, float]: if not isinstance(value, Mapping): raise EvaluationError("metrics 必须是对象") missing = REQUIRED_METRICS - set(value) if missing: raise EvaluationError(f"metrics 缺少:{', '.join(sorted(missing))}") result: dict[str, float] = {} for key in REQUIRED_METRICS: raw = value[key] if isinstance(raw, bool) or not isinstance(raw, (int, float)): raise EvaluationError(f"metrics.{key} 必须是数值") number = float(raw) if not math.isfinite(number) or number < 0.0: raise EvaluationError(f"metrics.{key} 必须是非负有限数值") result[key] = number if result["fall_rate"] > 1.0: raise EvaluationError("metrics.fall_rate 必须在 0–1 之间") return result def _improvement(baseline: Mapping[str, float], current: Mapping[str, float], key: str) -> float: scale = max(abs(baseline[key]), PHYSICAL_FLOORS[key]) return max(-1.0, min(1.0, (baseline[key] - current[key]) / scale)) def score_evaluation( baseline_value: Any, current_value: Any, objective_weights: Any = DEFAULT_OBJECTIVE_WEIGHTS, ) -> dict[str, Any]: baseline = validate_metrics(baseline_value) current = validate_metrics(current_value) weights = validate_objective_weights(objective_weights) components = { "velocity_tracking": 0.8 * _improvement(baseline, current, "linear_velocity_rmse") + 0.2 * _improvement(baseline, current, "angular_velocity_rmse"), "action_smoothness": _improvement(baseline, current, "mean_action_acc"), "posture_stability": _improvement(baseline, current, "orientation_error"), "fall_avoidance": _improvement(baseline, current, "fall_rate"), "foot_slip": _improvement(baseline, current, "slip_velocity"), "energy": _improvement(baseline, current, "mechanical_power"), } tracking_limit = max( baseline["linear_velocity_rmse"] * 1.05, baseline["linear_velocity_rmse"] + 1.0e-6 ) eligible = ( current["fall_rate"] <= baseline["fall_rate"] + 0.02 and current["linear_velocity_rmse"] <= tracking_limit ) total = sum(weights[key] * components[key] for key in weights) if not eligible: total = min(total, -1.0) return { "eligible": eligible, "score": total, "components": components, "metrics": current, "baselineMetrics": baseline, }