test(firmware): add LOS-31 golden replay scenarios

projects/rc_vehicle/firmware/sim/README.md

@@ -23,10 +23,22 @@
 
 - `simlib/replay.py` — `load_telemetry_csv()` (формат прошивки → кадры) +
   `find_invariant_violations()` (нет NaN, throttle/steering ∈ [-1,1], EKF не
-  расходится).
+  расходится). Необязательные колонки `rc_present`/`wifi_present` позволяют
+  воспроизводить потерю источника управления без изменения сенсорных данных.
 - `simlib/sim_host_runner.py` — `find_sim_host()` + `run_batch()`: прогон через
-  `sim_host` (FW-S2.1) в batch-режиме, фиделити «со средней точки» (`--identity-calib`).
-- `tests/fixtures/rides/` — golden-фикстуры (+ `PROVENANCE.md`, генератор синтетики).
+  `sim_host` (FW-S2.1) в batch-режиме, фиделити «со средней точки»
+  (`--identity-calib`); реальные LOS-31 вырезки используют калибровку
+  перевёрнутого монтажа (`--inverted-z-calib`, gravity=(0,0,-1),
+  forward=(1,0,0)) и при необходимости saved hard-iron/PCA mag calibration
+  (`--mag-calib`). Для сценариев также можно восстановить drive mode,
+  стабилизацию и записанные пороги oversteer.
+- `tests/fixtures/rides/` — golden-фикстуры, provenance и воспроизводимая
+  нарезка реальных эпизодов.
+- `tests/test_scenario_replay.py` — LOS-31: наклон, failsafe, oversteer,
+  прямая и задний ход с допусками вместо точного сравнения float.
+  Телеметрийные срезы ≈100 Гц линейно интерполируются на тики
+  2 мс, чтобы `HostStep()` шёл с производственной частотой 500 Гц;
+  сценарные ассерты выбирают выходы, ближайшие к исходным строкам.
 
 Для replay-теста нужен собранный `sim_host`:
 ```bash

projects/rc_vehicle/firmware/sim/simlib/__init__.py

@@ -29,7 +29,13 @@
     synth_gyro,
     synth_mag,
 )
-from .sim_host_runner import find_sim_host, run_batch
+from .sim_host_runner import (
+    MagReplayCalibration,
+    OversteerReplayConfig,
+    find_sim_host,
+    resample_frames,
+    run_batch,
+)
 from .sim_params import (
     SimParams,
     fitted_dynamic_params_2026_07_18,
@@ -56,6 +62,8 @@
     "Provenance",
     "RoadExcitation",
     "RoadNoiseModel",
+    "MagReplayCalibration",
+    "OversteerReplayConfig",
     "SLIP_CHANNELS",
     "SensorFrame",
     "SimParams",
@@ -80,6 +88,7 @@
     "load_profile",
     "load_profile_info",
     "resolve_profile",
+    "resample_frames",
     "save_profile",
     "simulate",
     "find_sim_host",

projects/rc_vehicle/firmware/sim/simlib/replay.py

@@ -22,20 +22,38 @@ def _f(row: dict, key: str, default: float = 0.0) -> float:
         return default
 
 
+def _b(row: dict, key: str, default: bool) -> bool:
+    """Прочитать необязательный bool из CSV без привязки к формату записи."""
+    value = row.get(key, "")
+    if value is None or value == "":
+        return default
+    normalized = str(value).strip().lower()
+    if normalized in {"1", "true", "yes", "on"}:
+        return True
+    if normalized in {"0", "false", "no", "off"}:
+        return False
+    return default
+
+
 def load_telemetry_csv(path: str) -> list[SensorFrame]:
     """Прочитать CSV прошивки → список кадров для sim_host.
 
     Используются: `ts_ms` (→ dt), калиброванный IMU `ax..gz`, сырой mag
-    `mx/my/mz`, команды `rc_throttle/rc_steering`. Разбор по именам колонок
-    (csv.DictReader) — устойчив к порядку/доп. полям.
+    `mx/my/mz`, команды `rc_throttle/rc_steering`. Golden-вырезки могут явно
+    задавать `dt_ms`, `mag_present`, `rc_present`, `wifi_present` и
+    Wi-Fi-команды. Явный `dt_ms` имеет приоритет над разницей `ts_ms`.
+    Старые логи без этих колонок сохраняют прежний контракт: mag/RC доступны,
+    Wi-Fi отсутствует. Разбор по именам колонок устойчив к порядку/доп. полям.
     """
     frames: list[SensorFrame] = []
     with open(path, newline="") as fh:
         reader = csv.DictReader(fh)
         prev_ts: float | None = None
         for row in reader:
             ts = _f(row, "ts_ms")
-            if prev_ts is None:
+            if row.get("dt_ms", "") != "":
+                dt_ms = max(1, int(round(_f(row, "dt_ms"))))
+            elif prev_ts is None:
                 dt_ms = 2
             else:
                 dt_ms = max(1, int(round(ts - prev_ts)))
@@ -44,11 +62,14 @@ def load_telemetry_csv(path: str) -> list[SensorFrame]:
                 dt_ms=dt_ms,
                 ax=_f(row, "ax"), ay=_f(row, "ay"), az=_f(row, "az"),
                 gx=_f(row, "gx"), gy=_f(row, "gy"), gz=_f(row, "gz"),
-                mag_present=True,
+                mag_present=_b(row, "mag_present", True),
                 mx=_f(row, "mx"), my=_f(row, "my"), mz=_f(row, "mz"),
-                rc_present=True,
+                rc_present=_b(row, "rc_present", True),
                 rc_throttle=_f(row, "rc_throttle"),
                 rc_steering=_f(row, "rc_steering"),
+                wifi_present=_b(row, "wifi_present", False),
+                wifi_throttle=_f(row, "wifi_throttle"),
+                wifi_steering=_f(row, "wifi_steering"),
             ))
     return frames
 

projects/rc_vehicle/firmware/sim/simlib/sim_host_runner.py

@@ -6,8 +6,119 @@
 
 import os
 import subprocess
+from bisect import bisect_left
+from dataclasses import dataclass
 from pathlib import Path
 
+from .frame import SensorFrame
+
+
+@dataclass(frozen=True)
+class OversteerReplayConfig:
+    """Параметры oversteer guard для воспроизведения записанного конфига."""
+
+    slip_thresh_deg: float = 10.0
+    rate_thresh_deg_s: float = 30.0
+    throttle_reduction: float = 0.7
+
+
+@dataclass(frozen=True)
+class MagReplayCalibration:
+    """Saved hard-iron/PCA calibration supplied to a real-data replay."""
+
+    offset: tuple[float, float, float]
+    field_strength_mgauss: float
+    normal: tuple[float, float, float]
+    basis1: tuple[float, float, float]
+    basis2: tuple[float, float, float]
+
+
+def resample_frames(
+    frames: list[SensorFrame], period_ms: int = 2
+) -> tuple[list[SensorFrame], list[int]]:
+    """Interpolate sparse telemetry onto the firmware control cadence.
+
+    Returns the 500 Hz replay frames and, for every source row, the index of
+    the nearest replay frame. The final short tick preserves exact duration.
+    """
+    if period_ms <= 0:
+        raise ValueError("period_ms must be positive")
+    if not frames:
+        return [], []
+
+    source_times: list[int] = []
+    elapsed_ms = 0
+    for frame in frames:
+        if frame.dt_ms <= 0:
+            raise ValueError("frame dt_ms must be positive")
+        elapsed_ms += frame.dt_ms
+        source_times.append(elapsed_ms)
+
+    tick_times = list(range(period_ms, elapsed_ms + 1, period_ms))
+    if not tick_times or tick_times[-1] != elapsed_ms:
+        tick_times.append(elapsed_ms)
+
+    def lerp(left: float, right: float, alpha: float) -> float:
+        return left + (right - left) * alpha
+
+    replay: list[SensorFrame] = []
+    right_index = 0
+    previous_tick = 0
+    for tick in tick_times:
+        while (
+            right_index < len(source_times)
+            and source_times[right_index] < tick
+        ):
+            right_index += 1
+
+        if right_index == 0:
+            left = right = frames[0]
+            alpha = 0.0
+        elif right_index >= len(frames):
+            left = right = frames[-1]
+            alpha = 0.0
+        else:
+            left = frames[right_index - 1]
+            right = frames[right_index]
+            interval = source_times[right_index] - source_times[right_index - 1]
+            alpha = (tick - source_times[right_index - 1]) / interval
+
+        discrete = left if alpha < 0.5 else right
+        replay.append(SensorFrame(
+            dt_ms=tick - previous_tick,
+            ax=lerp(left.ax, right.ax, alpha),
+            ay=lerp(left.ay, right.ay, alpha),
+            az=lerp(left.az, right.az, alpha),
+            gx=lerp(left.gx, right.gx, alpha),
+            gy=lerp(left.gy, right.gy, alpha),
+            gz=lerp(left.gz, right.gz, alpha),
+            mag_present=discrete.mag_present,
+            mx=lerp(left.mx, right.mx, alpha),
+            my=lerp(left.my, right.my, alpha),
+            mz=lerp(left.mz, right.mz, alpha),
+            rc_present=discrete.rc_present,
+            rc_throttle=lerp(left.rc_throttle, right.rc_throttle, alpha),
+            rc_steering=lerp(left.rc_steering, right.rc_steering, alpha),
+            wifi_present=discrete.wifi_present,
+            wifi_throttle=lerp(left.wifi_throttle, right.wifi_throttle, alpha),
+            wifi_steering=lerp(left.wifi_steering, right.wifi_steering, alpha),
+        ))
+        previous_tick = tick
+
+    source_indices: list[int] = []
+    for source_time in source_times:
+        index = bisect_left(tick_times, source_time)
+        if index == len(tick_times):
+            index -= 1
+        elif index > 0 and (
+            source_time - tick_times[index - 1]
+            <= tick_times[index] - source_time
+        ):
+            index -= 1
+        source_indices.append(index)
+
+    return replay, source_indices
+
 
 def find_sim_host() -> str | None:
     """Найти бинарь sim_host.
@@ -25,14 +136,53 @@ def find_sim_host() -> str | None:
 
 
 def run_batch(frames, sim_host_bin: str, identity_calib: bool = True,
-              timeout: float = 120.0) -> list[dict]:
+              timeout: float = 120.0, *, drive_mode: str | None = None,
+              stabilize: bool = False,
+              oversteer: OversteerReplayConfig | None = None,
+              inverted_z_calib: bool = False,
+              mag_calib: MagReplayCalibration | None = None,
+              start_test: str | None = None,
+              target_accel: float | None = None,
+              test_duration: float | None = None,
+              test_steering: float | None = None) -> list[dict]:
     """Прогнать кадры через sim_host в batch-режиме → список выходных строк.
 
     Каждая выходная строка — dict {имя_колонки: float} по заголовку sim_host.
     """
     args = [sim_host_bin, "--batch"]
-    if identity_calib:
+    if inverted_z_calib:
+        args.append("--inverted-z-calib")
+    elif identity_calib:
         args.append("--identity-calib")
+    if mag_calib is not None:
+        values = (
+            *mag_calib.offset,
+            mag_calib.field_strength_mgauss,
+            *mag_calib.normal,
+            *mag_calib.basis1,
+            *mag_calib.basis2,
+        )
+        args += ["--mag-calib", *(repr(float(value)) for value in values)]
+    if drive_mode:
+        args += ["--drive-mode", drive_mode]
+    if stabilize:
+        args.append("--stabilize")
+    if oversteer is not None:
+        args += [
+            "--oversteer",
+            "--oversteer-slip-thresh", repr(float(oversteer.slip_thresh_deg)),
+            "--oversteer-rate-thresh", repr(float(oversteer.rate_thresh_deg_s)),
+            "--oversteer-throttle-reduction",
+            repr(float(oversteer.throttle_reduction)),
+        ]
+    if start_test:
+        args += ["--start-test", start_test]
+        if target_accel is not None:
+            args += ["--target-accel", repr(float(target_accel))]
+        if test_duration is not None:
+            args += ["--test-duration", repr(float(test_duration))]
+        if test_steering is not None:
+            args += ["--test-steering", repr(float(test_steering))]
     payload = "".join(f.to_csv() + "\n" for f in frames)
     proc = subprocess.run(args, input=payload, capture_output=True, text=True,
                           timeout=timeout)

projects/rc_vehicle/firmware/sim/tests/fixtures/rides/PROVENANCE.md

@@ -3,12 +3,52 @@
 | Файл | Происхождение | Сценарий | Что проверяем |
 |------|---------------|----------|---------------|
 | `golden_synth_drive.csv` | **синтетика** (`gen_golden.py`, clean-signal профиль `fitted_2026_07_18`) | разгон + синусоида руля, 400 кадров @100 Гц | инварианты replay (нет NaN, throttle/steering ∈ [-1,1], EKF не расходится) |
+| `golden_real_static_tilt.csv` | `test_runs/telemetry_log_night_test_2.csv`, строки 9300:10400; извлечено 2026-08-18 | ровный pre-roll для startup-калибровки, затем статический наклон около 20°, 12.3 с | наклон в replay-выходе и circular σ PCA `heading_deg` |
+| `golden_real_failsafe_reconstructed.csv` | `test_runs/telemetry_log_auto_forward_2.csv`, строки 0:300; извлечено 2026-08-18 | неподвижная машина, источники управления отсутствуют | failsafe и нейтраль PWM |
+| `golden_real_oversteer_2026_08_02.csv` | `test_runs/telemetry_log_day_02_08_26.csv`, строки 19300:19650, запись 2026-08-02, `drive_mode=2` | реальный Drift-манёвр, исходное срабатывание 209346–209721 мс | `oversteer_active` внутри размеченной фазы |
+| `golden_real_straight_marker.csv` | `test_runs/telemetry_log_auto_forward_2.csv`, строки 2535:2763, `test_marker=1`; извлечено 2026-08-18 | движущаяся автоматическая прямая, 2.5 с | команда газа авто-теста и ограниченный дрейф курса |
+| `golden_real_reverse_2026_07_18.csv` | `test_runs/telemetry_log_forward_back.csv`, строки 1800:2200, запись 2026-07-18 | задний ход с почти нулевым рулевым вводом | отсутствие внесённого стабилизатором руля |
 
 Синтетическая фикстура использует динамику, откалиброванную по реальным заездам
 LOS-32, но не содержит сами логи или дорожный шум. Поэтому она детерминирована,
 самодостаточна и сохраняет clean-signal контракт replay из LOS-287.
 Формат CSV совпадает с прошивкой (`telemetry_log_*.csv`), поэтому тот же
 загрузчик (`simlib.load_telemetry_csv`) работает и на реальных поездках.
 
-**Реальные golden-вырезки** (наклон, занос, failsafe, задний ход) добавляются в
-FW-S2.3 (сценарные ассерты) и FW-S2.7 (решение по хранению/git LFS).
+Вырезки воспроизводятся скриптом `extract_real_golden.py`; полные исходные логи
+остаются локальными и исключены из Git. Колонка `replay_phase` отделяет контекст
+от участка сценарного ассерта.
+Перед `sim_host` срезы ≈100 Гц интерполируются на равномерные тики
+2 мс: прошивка получает примерно пять `HostStep()` на каждую строку
+телеметрии, как в production control loop 500 Гц. Для сверки с
+`replay_phase` берётся ближайший 2-мс output к времени исходной строки.
+
+Во всех реальных источниках IMU установлен перевёрнуто: покой даёт `az≈−1 g`,
+а ось X совпадает с направлением движения. Replay поэтому загружает сохранённый
+vehicle frame `gravity=(0,0,-1)`, `forward=(1,0,0)` с нулевыми bias (логи уже
+bias-корректированы). `RotateToVehicleFrame()` приводит accel и gyro к
+каноническому кадру с `az≈+1 g`; identity-калибровка используется только старой
+синтетической fixture.
+
+Исходные логи не содержат `rc_present`/`wifi_present`, поэтому failsafe-кейс
+использует реальные сенсоры неподвижного эпизода, но состояние отсутствующих
+источников управления реконструировано колонками `rc_present=0` и
+`wifi_present=0`. Для oversteer из записанного config snapshot воспроизводятся
+`warn_enabled=true`, пороги 10°/30°·с⁻¹ и снижение газа 0.7.
+Для straight replay по записанному `test_marker=1` восстанавливается auto-test
+`straight` с target acceleration 0.1 g и duration 3 с, а также WebSocket keepalive
+без ручной Wi-Fi команды. Эти дополнения явно относятся к replay и не
+выдаются за поля исходной телеметрии.
+
+Для tilt-эпизода saved magnetometer calibration восстановлена из
+планарного поворота в том же `telemetry_log_night_test_2.csv`, строки
+9800:11800, тем же hard-iron min/max + PCA-алгоритмом, что и
+`MagCalibration::Finish()`. Планарность выборки λmin/λmid=0.044;
+offset=(335.907, 56.915, 447.906) мГс, field strength=208.505 мГс. Replay
+передаёт offset, normal и оба basis-вектора в `sim_host` до `Init()`,
+поэтому `heading_deg` идёт через `ComputePcaHeadingDeg()`, а не fallback
+`atan2(my, mx)`.
+Фикстура также хранит исходный `dt_ms` первого кадра, чтобы
+холодный replay не терял фактический интервал на границе вырезки.
+
+Решение о хранении полных логов и Git LFS остаётся в FW-S2.7.