#!/usr/bin/env python3 """Offline evaluation of Ford's native four-field path polynomial. The experiment deliberately does not alter the live controller. It rebases the model path into the vehicle pose expected at actuation time, fits one cubic over the remaining short path, and converts the cubic into the LMC2 C0/C1/C2/C3 signals. A first-order C2 response envelope is included to expose commands that would look good only if the PSCM curvature channel were instantaneous. """ import argparse from collections import defaultdict from dataclasses import dataclass import glob import math from pathlib import Path import numpy as np from openpilot.tools.lib.logreader import LogReader DBC_OFFSET = (-5.12, 5.11) DBC_ANGLE = (-0.5, 0.5235) DBC_CURVATURE = (-0.02, 0.02) DBC_CURVATURE_RATE = (-0.001024, 0.001023) MAX_LATERAL_ACCEL = 3.0 + 9.81 * 0.06 MAX_LATERAL_JERK = 3.0 + 9.81 * 0.06 @dataclass(frozen=True) class ModelPath: x: np.ndarray y: np.ndarray heading: np.ndarray distance: np.ndarray @dataclass(frozen=True) class Sample: route: str time: float speed: float curvature: float steering_pressed: bool path: ModelPath sent_c0: float sent_c1: float sent_c2: float sent_c3: float @dataclass(frozen=True) class NativePath: c0: float c1: float c2: float c3: float fit_rmse: float path_rms: float def _model_path(model) -> ModelPath | None: try: x = np.asarray(model.position.x, dtype=float) y = np.asarray(model.position.y, dtype=float) heading = np.unwrap(np.asarray(model.orientation.z, dtype=float)) except (AttributeError, TypeError, ValueError): return None if len(x) < 4 or len(x) != len(y) or len(x) != len(heading): return None if not np.isfinite(np.concatenate((x, y, heading))).all(): return None distance = np.concatenate(([0.0], np.cumsum(np.hypot(np.diff(x), np.diff(y))))) unique_distance, unique = np.unique(distance, return_index=True) if len(unique_distance) < 4 or unique_distance[-1] <= 0.0: return None return ModelPath(x[unique], y[unique], heading[unique], unique_distance) def _arc_pose(distance: float, curvature: float) -> tuple[float, float, float]: heading = curvature * distance if abs(curvature) < 1e-9: return distance, 0.0, 0.0 return math.sin(heading) / curvature, (1.0 - math.cos(heading)) / curvature, heading def _relative_points(path: ModelPath, vehicle_pose: tuple[float, float, float], start: float, horizon: float, count: int = 25) -> tuple[np.ndarray, np.ndarray]: sample_distance = np.linspace(start, min(start + horizon, path.distance[-1]), count) desired_x = np.interp(sample_distance, path.distance, path.x) desired_y = np.interp(sample_distance, path.distance, path.y) vehicle_x, vehicle_y, vehicle_heading = vehicle_pose dx = desired_x - vehicle_x dy = desired_y - vehicle_y cosine = math.cos(vehicle_heading) sine = math.sin(vehicle_heading) return cosine * dx + sine * dy, -sine * dx + cosine * dy def _fit_points(path: ModelPath, speed: float, current_curvature: float, delay: float, horizon: float) -> tuple[np.ndarray, np.ndarray] | None: advance = min(max(speed, 0.0) * delay, path.distance[-1]) available = min(horizon, path.distance[-1] - advance) if available <= 0.25: return None x, y = _relative_points(path, _arc_pose(advance, current_curvature), advance, available) forward = (x >= -0.25) & (x <= horizon) x = x[forward] y = y[forward] if len(x) < 4 or np.ptp(x) <= 0.25: return None return x, y def _wire_coefficients(c0: float, c1: float, c2: float, c3: float) -> tuple[float, float, float, float]: slope = math.tan(c1) slope_norm = 1.0 + slope ** 2 a2 = 0.5 * c2 * slope_norm ** 1.5 a3 = (c3 + 12.0 * slope * a2 ** 2 / slope_norm ** 3) * slope_norm ** 2 / 6.0 return c0, slope, a2, a3 def _wire_rmse(command: tuple[float, float, float, float], x: np.ndarray, y: np.ndarray) -> float: a0, a1, a2, a3 = _wire_coefficients(*command) reconstructed = a0 + a1 * x + a2 * x ** 2 + a3 * x ** 3 return float(np.sqrt(np.mean((reconstructed - y) ** 2))) def fit_native_path(path: ModelPath, speed: float, current_curvature: float, *, delay: float, horizon: float) -> NativePath: """Fit the delay-aligned path and return physical LMC2 fields. C2 and C3 are curvature and curvature rate at the vehicle-frame origin, not the raw quadratic and cubic polynomial coefficients. """ points = _fit_points(path, speed, current_curvature, delay, horizon) if points is None: return NativePath(0.0, 0.0, 0.0, 0.0, 0.0, 0.0) x, y = points # Scaling x before the least-squares solve keeps tight-turn fits well # conditioned while preserving an ordinary cubic in vehicle coordinates. scale = max(float(np.max(np.abs(x))), 1.0) normalized_x = x / scale design = np.column_stack((np.ones(len(x)), normalized_x, normalized_x ** 2, normalized_x ** 3)) scaled, *_ = np.linalg.lstsq(design, y, rcond=None) a0, a1, a2, a3 = (float(scaled[index] / scale ** index) for index in range(4)) slope = a1 slope_norm = 1.0 + slope ** 2 curvature = 2.0 * a2 / slope_norm ** 1.5 curvature_rate = 6.0 * a3 / slope_norm ** 2 - 12.0 * slope * a2 ** 2 / slope_norm ** 3 command = (float(np.clip(a0, *DBC_OFFSET)), float(np.clip(math.atan(slope), *DBC_ANGLE)), float(np.clip(curvature, *DBC_CURVATURE)), float(np.clip(curvature_rate, *DBC_CURVATURE_RATE))) return NativePath( *command, _wire_rmse(command, x, y), float(np.sqrt(np.mean(y ** 2))), ) def fit_c2_aware_path(path: ModelPath, speed: float, current_curvature: float, *, delay: float, horizon: float, target_c2: float, effective_c2: float, use_c3: bool = True) -> NativePath: """Fit fast fields around the C2 curvature the PSCM is expected to realize.""" points = _fit_points(path, speed, current_curvature, delay, horizon) if points is None: return NativePath(0.0, 0.0, target_c2, 0.0, 0.0, 0.0) x, y = points slope = 0.0 a0 = a1 = a3 = 0.0 for _ in range(3): a2 = 0.5 * effective_c2 * (1.0 + slope ** 2) ** 1.5 design = np.column_stack((np.ones(len(x)), x, x ** 3)) (a0, a1, a3), *_ = np.linalg.lstsq(design, y - a2 * x ** 2, rcond=None) slope = float(a1) slope_norm = 1.0 + slope ** 2 c3 = 6.0 * float(a3) / slope_norm ** 2 - 12.0 * slope * a2 ** 2 / slope_norm ** 3 c3 = float(np.clip(c3, *DBC_CURVATURE_RATE)) if use_c3 else 0.0 # Once C2 and C3 are fixed to what the hardware can realize, refit C0/C1 so # their fast feedback preserves as much of the same path as possible. _, _, fixed_a2, fixed_a3 = _wire_coefficients(0.0, math.atan(slope), effective_c2, c3) (a0, a1), *_ = np.linalg.lstsq(np.column_stack((np.ones(len(x)), x)), y - fixed_a2 * x ** 2 - fixed_a3 * x ** 3, rcond=None) c0 = float(np.clip(a0, *DBC_OFFSET)) c1 = float(np.clip(math.atan(float(a1)), *DBC_ANGLE)) effective_command = (c0, c1, effective_c2, c3) return NativePath(c0, c1, target_c2, c3, _wire_rmse(effective_command, x, y), float(np.sqrt(np.mean(y ** 2)))) def _route(path: str) -> str: return Path(path).name.split("--", 1)[0] def load_samples(paths: list[str], stride: int = 2) -> list[Sample]: grouped: dict[str, list[str]] = defaultdict(list) for path in paths: grouped[_route(path)].append(path) samples = [] for route, route_paths in sorted(grouped.items()): events = [] for path in sorted(route_paths): events.extend(LogReader(path)) events.sort(key=lambda event: event.logMonoTime) if not events: continue start_time = events[0].logMonoTime model_path = None curvature = 0.0 lat_active = path_valid = False sent = (0.0, 0.0, 0.0, 0.0) car_state_count = 0 for event in events: which = event.which() if which == "modelV2": model_path = _model_path(event.modelV2) elif which == "controlsState": curvature = float(event.controlsState.curvature) elif which == "carControl": lat_active = bool(event.carControl.latActive) elif which == "carControlSP": command = event.carControlSP.fordLateralPath path_valid = bool(command.valid) sent = (float(command.pathOffset), float(command.pathAngle), float(command.curvature), float(command.curvatureRate)) elif which == "carState" and lat_active and path_valid and model_path is not None: car_state_count += 1 if car_state_count % stride: continue samples.append(Sample( route, (event.logMonoTime - start_time) * 1e-9, float(event.carState.vEgo), curvature, bool(event.carState.steeringPressed), model_path, *sent, )) return samples def _percentile(values: np.ndarray, percentile: float, mask: np.ndarray | None = None) -> float: selected = values if mask is None else values[mask] return float(np.percentile(np.abs(selected), percentile)) if len(selected) else math.nan def _route_rate(samples: list[Sample], values: np.ndarray) -> np.ndarray: rate = np.zeros(len(values)) for index in range(1, len(values)): dt = samples[index].time - samples[index - 1].time if samples[index].route == samples[index - 1].route and 0.005 <= dt <= 0.2: rate[index] = (values[index] - values[index - 1]) / dt return rate def _c2_response(samples: list[Sample], target: np.ndarray, tau_load: float, tau_unload: float) -> np.ndarray: effective = np.zeros(len(target)) previous_route = None previous_time = 0.0 state = 0.0 for index, sample in enumerate(samples): if sample.route != previous_route: state = 0.0 previous_time = sample.time dt = float(np.clip(sample.time - previous_time, 0.005, 0.2)) loading = target[index] * state >= 0.0 and abs(target[index]) > abs(state) tau = tau_load if loading else tau_unload state += (1.0 - math.exp(-dt / tau)) * (target[index] - state) effective[index] = state previous_route, previous_time = sample.route, sample.time return effective def _limit_c2_command(samples: list[Sample], target: np.ndarray) -> np.ndarray: """Mirror the CAN-FD Ford curvature acceleration/jerk limiter.""" limited = np.zeros(len(target)) previous_route = None previous_time = 0.0 previous = 0.0 for index, sample in enumerate(samples): if sample.route != previous_route: previous = 0.0 previous_time = sample.time dt = float(np.clip(sample.time - previous_time, 0.005, 0.2)) speed = max(sample.speed, 1.0) value = float(np.clip(target[index], -MAX_LATERAL_ACCEL / speed ** 2, MAX_LATERAL_ACCEL / speed ** 2)) step = MAX_LATERAL_JERK / speed ** 2 * dt value = float(np.clip(value, previous - step, previous + step)) limited[index] = float(np.clip(value, *DBC_CURVATURE)) previous = limited[index] previous_route, previous_time = sample.route, sample.time return limited def _limit_fast_fields(samples: list[Sample], c0_target: np.ndarray, c1_target: np.ndarray) -> tuple[np.ndarray, np.ndarray]: c0 = np.zeros(len(samples)) c1 = np.zeros(len(samples)) previous_route = None previous_time = 0.0 previous_c0 = previous_c1 = 0.0 for index, sample in enumerate(samples): if sample.route != previous_route: previous_c0 = previous_c1 = 0.0 previous_time = sample.time dt = float(np.clip(sample.time - previous_time, 0.005, 0.2)) c0[index] = np.clip(c0_target[index], previous_c0 - 4.0 * dt, previous_c0 + 4.0 * dt) c1[index] = np.clip(c1_target[index], previous_c1 - 1.0 * dt, previous_c1 + 1.0 * dt) previous_c0, previous_c1 = c0[index], c1[index] previous_route, previous_time = sample.route, sample.time return c0, c1 def evaluate(samples: list[Sample], *, delay: float, horizon: float, tau_load: float, tau_unload: float, horizon_time: float = 0.0, assumed_tau_load: float | None = None, assumed_tau_unload: float | None = None, use_c3: bool = True, c2_limit: float = DBC_CURVATURE[1]) -> dict[str, float]: horizons = np.asarray([float(np.clip(sample.speed * horizon_time, 1.0, horizon)) if horizon_time > 0.0 else horizon for sample in samples]) commands = [fit_native_path(sample.path, sample.speed, sample.curvature, delay=delay, horizon=sample_horizon) for sample, sample_horizon in zip(samples, horizons, strict=True)] c0 = np.asarray([command.c0 for command in commands]) c1 = np.asarray([command.c1 for command in commands]) raw_c2 = np.asarray([command.c2 for command in commands]) c2 = np.clip(raw_c2, -c2_limit, c2_limit) c3 = np.asarray([command.c3 for command in commands]) fit_rmse = np.asarray([command.fit_rmse for command in commands]) path_rms = np.asarray([command.path_rms for command in commands]) transmitted_c2 = _limit_c2_command(samples, c2) effective_c2 = _c2_response(samples, transmitted_c2, tau_load, tau_unload) estimated_c2 = _c2_response(samples, transmitted_c2, tau_load if assumed_tau_load is None else assumed_tau_load, tau_unload if assumed_tau_unload is None else assumed_tau_unload) compensated = [fit_c2_aware_path(sample.path, sample.speed, sample.curvature, delay=delay, horizon=sample_horizon, target_c2=target, effective_c2=estimated, use_c3=use_c3) for sample, sample_horizon, target, estimated in zip(samples, horizons, c2, estimated_c2, strict=True)] compensated_c0 = np.asarray([command.c0 for command in compensated]) compensated_c1 = np.asarray([command.c1 for command in compensated]) compensated_c3 = np.asarray([command.c3 for command in compensated]) limited_c0, limited_c1 = _limit_fast_fields(samples, compensated_c0, compensated_c1) estimated_compensated_rmse = np.asarray([command.fit_rmse for command in compensated]) compensated_rmse = [] for sample, sample_horizon, command, effective in zip(samples, horizons, compensated, effective_c2, strict=True): points = _fit_points(sample.path, sample.speed, sample.curvature, delay, sample_horizon) compensated_rmse.append(0.0 if points is None else _wire_rmse( (command.c0, command.c1, effective, command.c3), *points)) compensated_rmse = np.asarray(compensated_rmse) limited_compensated_rmse = [] for sample, sample_horizon, c0_value, c1_value, c3_value, effective in \ zip(samples, horizons, limited_c0, limited_c1, compensated_c3, effective_c2, strict=True): points = _fit_points(sample.path, sample.speed, sample.curvature, delay, sample_horizon) limited_compensated_rmse.append(0.0 if points is None else _wire_rmse( (c0_value, c1_value, effective, c3_value), *points)) limited_compensated_rmse = np.asarray(limited_compensated_rmse) missing_c2 = transmitted_c2 - effective_c2 # Compare channels by their lateral contribution at the fit horizon. This # includes C3: treating it as zero would incorrectly blame C0/C1 for a # curvature transition the native polynomial assigns to curvature rate. fast = (2.0 * compensated_c0 / horizons ** 2 + 2.0 * np.tan(compensated_c1) / horizons + compensated_c3 * horizons / 3.0) lagging = np.abs(missing_c2) > 0.0005 unloading = lagging & (np.abs(c2) < 0.75 * np.abs(effective_c2)) pressed = np.asarray([sample.steering_pressed for sample in samples]) speed = np.asarray([sample.speed for sample in samples]) sent_c2 = np.asarray([sample.sent_c2 for sample in samples]) sent_transmitted_c2 = _limit_c2_command(samples, sent_c2) sent_effective_c2 = _c2_response(samples, sent_transmitted_c2, tau_load, tau_unload) sent_lpf_rmse = [] for sample, sample_horizon, effective in zip(samples, horizons, sent_effective_c2, strict=True): points = _fit_points(sample.path, sample.speed, sample.curvature, delay, sample_horizon) sent_lpf_rmse.append(0.0 if points is None else _wire_rmse( (sample.sent_c0, sample.sent_c1, effective, sample.sent_c3), *points)) sent_lpf_rmse = np.asarray(sent_lpf_rmse) raw_c2_rate = _route_rate(samples, c2) c2_rate = _route_rate(samples, transmitted_c2) sent_c2_rate = _route_rate(samples, sent_c2) compensated_c0_rate = _route_rate(samples, compensated_c0) compensated_c1_rate = _route_rate(samples, compensated_c1) compensated_c3_rate = _route_rate(samples, compensated_c3) normalized_fit = np.divide(fit_rmse, path_rms, out=np.zeros_like(fit_rmse), where=path_rms > 1e-4) return { "samples": float(len(samples)), "delay": delay, "horizon": horizon, "horizon_time": horizon_time, "assumed_tau_load": tau_load if assumed_tau_load is None else assumed_tau_load, "assumed_tau_unload": tau_unload if assumed_tau_unload is None else assumed_tau_unload, "use_c3": float(use_c3), "c2_limit": c2_limit, "actual_horizon_p50": _percentile(horizons, 50), "actual_horizon_p95": _percentile(horizons, 95), "fit_rmse_p50": _percentile(fit_rmse, 50), "fit_rmse_p95": _percentile(fit_rmse, 95), "normalized_fit_p95": _percentile(normalized_fit, 95), "c2_aware_rmse_p50": _percentile(compensated_rmse, 50), "c2_aware_rmse_p95": _percentile(compensated_rmse, 95), "c2_aware_estimated_rmse_p95": _percentile(estimated_compensated_rmse, 95), "c2_aware_limited_rmse_p95": _percentile(limited_compensated_rmse, 95), "sent_lpf_rmse_p50": _percentile(sent_lpf_rmse, 50), "sent_lpf_rmse_p95": _percentile(sent_lpf_rmse, 95), "c0_p95": _percentile(c0, 95), "c1_p95": _percentile(c1, 95), "c2_p95": _percentile(c2, 95), "c3_p95": _percentile(c3, 95), "c0_clip_rate": float(np.mean((c0 <= DBC_OFFSET[0]) | (c0 >= DBC_OFFSET[1]))), "c1_clip_rate": float(np.mean((c1 <= DBC_ANGLE[0]) | (c1 >= DBC_ANGLE[1]))), "c2_clip_rate": float(np.mean((c2 <= DBC_CURVATURE[0]) | (c2 >= DBC_CURVATURE[1]))), "c3_clip_rate": float(np.mean((c3 <= DBC_CURVATURE_RATE[0]) | (c3 >= DBC_CURVATURE_RATE[1]))), "c2_aware_c0_p95": _percentile(compensated_c0, 95), "c2_aware_c1_p95": _percentile(compensated_c1, 95), "c2_aware_c3_p95": _percentile(compensated_c3, 95), "c2_aware_c0_rate_p95": _percentile(compensated_c0_rate, 95), "c2_aware_c1_rate_p95": _percentile(compensated_c1_rate, 95), "c2_aware_c0_rate_limit_rate": float(np.mean(np.abs(compensated_c0_rate) > 4.0)), "c2_aware_c1_rate_limit_rate": float(np.mean(np.abs(compensated_c1_rate) > 1.0)), "c2_aware_c3_rate_p95": _percentile(compensated_c3_rate, 95), "raw_c2_rate_p95": _percentile(raw_c2_rate, 95), "c2_rate_p95": _percentile(c2_rate, 95), "sent_c2_rate_p95": _percentile(sent_c2_rate, 95), "c2_lag_p95": _percentile(missing_c2, 95), "lag_samples": float(np.count_nonzero(lagging)), "lag_fast_support_rate": float(np.mean(fast[lagging] * missing_c2[lagging] > 0.0)) if np.any(lagging) else math.nan, "lag_fast_coverage_p50": _percentile(np.divide(fast, missing_c2, out=np.zeros_like(fast), where=np.abs(missing_c2) > 1e-6), 50, lagging), "unload_samples": float(np.count_nonzero(unloading)), "unload_fast_counter_rate": float(np.mean(fast[unloading] * effective_c2[unloading] < 0.0)) if np.any(unloading) else math.nan, "unload_residual_c2_p95": _percentile(effective_c2 - c2, 95, unloading), "pressed_c0_p95": _percentile(c0, 95, pressed), "pressed_c1_p95": _percentile(c1, 95, pressed), "low_speed_fit_p95": _percentile(fit_rmse, 95, speed < 5.0), "road_speed_fit_p95": _percentile(fit_rmse, 95, speed >= 15.0), } def _expand(patterns: list[str]) -> list[str]: return sorted({path for pattern in patterns for path in glob.glob(pattern)}) def _self_test() -> None: distance = np.linspace(0.0, 20.0, 81) coefficients = (0.2, 0.03, 0.004, -0.00005) y = sum(coefficient * distance ** power for power, coefficient in enumerate(coefficients)) slope = coefficients[1] + 2.0 * coefficients[2] * distance + 3.0 * coefficients[3] * distance ** 2 heading = np.arctan(slope) path = ModelPath(distance, y, heading, np.concatenate(([0.0], np.cumsum(np.hypot(np.diff(distance), np.diff(y)))))) command = fit_native_path(path, 0.0, 0.0, delay=0.1, horizon=7.0) assert abs(command.c0 - coefficients[0]) < 2e-3 assert abs(command.c1 - math.atan(coefficients[1])) < 2e-3 expected_c2 = 2.0 * coefficients[2] / (1.0 + coefficients[1] ** 2) ** 1.5 assert abs(command.c2 - expected_c2) < 2e-4 assert command.fit_rmse < 1e-4 def main() -> int: parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--logs", action="append", help="rlog glob", default=[]) parser.add_argument("--delay", type=float, default=0.1) parser.add_argument("--horizon", type=float, action="append") parser.add_argument("--time-horizon", type=float, default=0.0, help="if nonzero, use clamp(speed * seconds, 1 m, --horizon)") parser.add_argument("--tau-load", type=float, default=0.75) parser.add_argument("--tau-unload", type=float, default=1.3) parser.add_argument("--assumed-tau-load", type=float) parser.add_argument("--assumed-tau-unload", type=float) parser.add_argument("--zero-c3", action="store_true") parser.add_argument("--c2-limit", type=float, action="append", help="C2 cap to test; defaults to gentle 0.006 and full 0.02") parser.add_argument("--self-test", action="store_true") args = parser.parse_args() if args.self_test: _self_test() paths = _expand(args.logs) if not paths: if args.self_test: return 0 parser.error("at least one usable --logs glob is required") samples = load_samples(paths) if not samples: parser.error("logs contain no active Ford path samples") print(f"loaded_logs={len(paths)} samples={len(samples)} tau_load={args.tau_load} tau_unload={args.tau_unload}") for horizon in args.horizon or [3.5, 5.0, 7.0, 10.0]: for c2_limit in args.c2_limit or [0.006, DBC_CURVATURE[1]]: result = evaluate(samples, delay=args.delay, horizon=horizon, tau_load=args.tau_load, tau_unload=args.tau_unload, horizon_time=args.time_horizon, assumed_tau_load=args.assumed_tau_load, assumed_tau_unload=args.assumed_tau_unload, use_c3=not args.zero_c3, c2_limit=c2_limit) print(" ".join(f"{key}={value:.8g}" for key, value in result.items())) return 0 if __name__ == "__main__": raise SystemExit(main())