feat(log-analysis): add timestamped parameter history

ardupilot_methodic_configurator/log_analysis/backend_log_analysis.py

@@ -22,6 +22,7 @@
 )
 from ardupilot_methodic_configurator.log_analysis.data_model_log_analysis_context import LogAnalysisContext
 from ardupilot_methodic_configurator.log_analysis.data_model_log_data import LogData
+from ardupilot_methodic_configurator.log_analysis.data_model_parameter_history import ParameterHistory
 from ardupilot_methodic_configurator.log_analysis.utils import APMDoc
 
 
@@ -57,6 +58,7 @@ def analyze_log_data(  # pylint: disable=too-many-arguments
         configuration_steps=configuration_steps or {},
         vehicle_components=vehicle_components or {},
         apm_doc=apm_doc,
+        parameter_history=ParameterHistory.from_log_data(log_data),
     )
     return analyze_log(log_data, context)
 

ardupilot_methodic_configurator/log_analysis/data_model_log_analysis_context.py

@@ -15,6 +15,7 @@
     ConfigurationStepParameterDeriver,
     ParameterDeriver,
 )
+from ardupilot_methodic_configurator.log_analysis.data_model_parameter_history import ParameterHistory
 from ardupilot_methodic_configurator.log_analysis.utils import APMDoc
 
 
@@ -27,3 +28,4 @@ class LogAnalysisContext:
     vehicle_components: dict[str, Any] = field(default_factory=dict)
     apm_doc: APMDoc | None = None
     parameter_deriver: ParameterDeriver = field(default_factory=ConfigurationStepParameterDeriver)
+    parameter_history: ParameterHistory = field(default_factory=ParameterHistory)

ardupilot_methodic_configurator/log_analysis/data_model_parameter_history.py

@@ -0,0 +1,74 @@
+"""
+Timestamped ArduPilot parameter history for log analysis.
+
+SPDX-FileCopyrightText: 2026 Donald Smith
+
+SPDX-License-Identifier: GPL-3.0-or-later
+"""
+
+import math
+from bisect import bisect_right
+from dataclasses import dataclass, field
+
+from ardupilot_methodic_configurator.log_analysis.data_model_log_data import LogData
+
+
+@dataclass(frozen=True, slots=True)
+class _ParameterValue:
+    """One logged parameter value and its scaled timestamp in seconds."""
+
+    time_s: float
+    value: float
+
+
+@dataclass(frozen=True, slots=True)
+class ParameterHistory:
+    """
+    Resolve logged parameter values at analysis timestamps in seconds.
+
+    ArduPilot's initial parameter snapshot is emitted incrementally after log
+    startup. Therefore, the chronologically first record for a parameter is its
+    baseline from the beginning of the log, rather than becoming applicable
+    only at that record's timestamp. Later records take effect at their logged
+    timestamps. This is stepwise resolution, not interpolation.
+    """
+
+    _values_by_name: dict[str, tuple[_ParameterValue, ...]] = field(default_factory=dict)
+
+    @classmethod
+    def from_log_data(cls, log_data: LogData) -> "ParameterHistory":
+        """Build history from scaled PARM records already extracted into ``log_data``."""
+        values_by_name: dict[str, list[_ParameterValue]] = {}
+        for record in log_data.iter_message_records("PARM"):
+            name = record.get("Name")
+            value = record.get("Value")
+            time_s = record.get("TimeUS")
+            if not isinstance(name, str) or not name or value is None or time_s is None:
+                continue
+
+            timestamp = float(time_s)
+            if not math.isfinite(timestamp):
+                msg = f"PARM timestamp for {name} must be finite"
+                raise ValueError(msg)
+            values_by_name.setdefault(name, []).append(_ParameterValue(timestamp, float(value)))
+
+        return cls({name: tuple(sorted(values, key=lambda item: item.time_s)) for name, values in values_by_name.items()})
+
+    def value_at(self, parameter_name: str, time_s: float) -> float | None:
+        """
+        Return the value applicable at ``time_s``, or ``None`` when absent.
+
+        The first logged value is the log-start baseline, including for queries
+        before its timestamp. At duplicate timestamps, the last logged record
+        at that timestamp wins.
+        """
+        if not math.isfinite(time_s):
+            msg = "Parameter query time_s must be finite"
+            raise ValueError(msg)
+
+        values = self._values_by_name.get(parameter_name)
+        if not values:
+            return None
+
+        index = bisect_right(values, time_s, key=lambda item: item.time_s) - 1
+        return values[max(index, 0)].value

tests/test_backend_log_analysis.py

@@ -24,18 +24,21 @@ def test_analyze_log_file_loads_inputs_and_builds_context(monkeypatch: Any) -> N
     log_data.firmware_version = (4, 6, 3)
     log_data.add_message_columns(
         "PARM",
-        np.array([("LOG_BITMASK", 7.0)], dtype=[("Name", "U16"), ("Value", "f8")]),
+        np.array(
+            [(1.0, "LOG_BITMASK", 5.0), (2.0, "LOG_BITMASK", 7.0)],
+            dtype=[("TimeUS", "f8"), ("Name", "U16"), ("Value", "f8")],
+        ),
         MessageSchema(
             name="PARM",
             msg_type=1,
             length=1,
-            format="Nf",
-            fields=["Name", "Value"],
-            stored_units=["", ""],
-            scaled_units=["", ""],
-            multipliers=[None, None],
-            multipliers_applied_at_ingest=[False, False],
-            records=1,
+            format="QNf",
+            fields=["TimeUS", "Name", "Value"],
+            stored_units=["s", "", ""],
+            scaled_units=["s", "", ""],
+            multipliers=[None, None, None],
+            multipliers_applied_at_ingest=[False, False, False],
+            records=2,
         ),
     )
 
@@ -82,6 +85,8 @@ def fake_analyze(received_log_data: LogData, context: backend_log_analysis.LogAn
     assert seen["validate"] == ("ArduCopter", (4, 6, 3), "ArduCopter", "4.6.3")
     assert seen["log_data"] is log_data
     assert seen["context"].parameters == {"LOG_BITMASK": 7.0}
+    assert seen["context"].parameter_history.value_at("LOG_BITMASK", 1.5) == 5.0
+    assert seen["context"].parameter_history.value_at("LOG_BITMASK", 2.0) == 7.0
     assert seen["context"].configuration_steps == {"05_battery.param": {}}
     assert seen["context"].vehicle_components == {"Frame": {}}
     assert seen["context"].apm_doc == {"LOG_BITMASK": {}}

tests/test_data_model_parameter_history.py

@@ -0,0 +1,144 @@
+#!/usr/bin/env python3
+
+"""
+Tests for timestamped ArduPilot parameter history.
+
+SPDX-FileCopyrightText: 2026 Donald Smith
+
+SPDX-License-Identifier: GPL-3.0-or-later
+"""
+
+import math
+
+import numpy as np
+import pytest
+
+from ardupilot_methodic_configurator.log_analysis.data_model_log_data import LogData, MessageSchema
+from ardupilot_methodic_configurator.log_analysis.data_model_parameter_history import ParameterHistory
+
+# The PARM LogData fixture intentionally mirrors backend orchestration setup.
+# pylint: disable=duplicate-code
+
+
+def _log_data(records: list[tuple[float, str, float]]) -> LogData:
+    log_data = LogData()
+    rows = np.array(records, dtype=[("TimeUS", "f8"), ("Name", "U16"), ("Value", "f8")])
+    log_data.add_message_columns(
+        "PARM",
+        rows,
+        MessageSchema(
+            name="PARM",
+            msg_type=1,
+            length=1,
+            format="Qnf",
+            fields=["TimeUS", "Name", "Value"],
+            stored_units=["s", "", ""],
+            scaled_units=["s", "", ""],
+            multipliers=[None, None, None],
+            multipliers_applied_at_ingest=[False, False, False],
+            records=len(records),
+        ),
+    )
+    return log_data
+
+
+def test_absent_parameter_is_unavailable() -> None:
+    assert ParameterHistory.from_log_data(LogData()).value_at("MISSING", 10.0) is None
+
+
+@pytest.mark.parametrize("time_s", [0.0, 5.0, 10.0])
+def test_one_occurrence_applies_throughout_log(time_s: float) -> None:
+    history = ParameterHistory.from_log_data(_log_data([(5.0, "TEST", 1.0)]))
+
+    assert history.value_at("TEST", time_s) == 1.0
+
+
+@pytest.mark.parametrize(
+    ("time_s", "expected"),
+    [
+        (4.999999, 1.0),
+        (5.0, 1.0),
+        (5.000001, 1.0),
+        (9.999999, 1.0),
+        (10.0, 2.0),
+        (10.000001, 2.0),
+        (14.0, 2.0),
+        (20.0, 3.0),
+        (25.0, 3.0),
+    ],
+)
+def test_value_at_uses_baseline_and_stepwise_changes(time_s: float, expected: float) -> None:
+    history = ParameterHistory.from_log_data(_log_data([(5.0, "TEST", 1.0), (10.0, "TEST", 2.0), (20.0, "TEST", 3.0)]))
+
+    assert history.value_at("TEST", time_s) == expected
+
+
+def test_repeated_same_value_records_preserve_value() -> None:
+    history = ParameterHistory.from_log_data(_log_data([(5.0, "TEST", 1.0), (10.0, "TEST", 1.0)]))
+
+    assert history.value_at("TEST", 7.0) == 1.0
+    assert history.value_at("TEST", 10.0) == 1.0
+
+
+def test_multiple_parameter_names_are_independent() -> None:
+    history = ParameterHistory.from_log_data(_log_data([(5.0, "FIRST", 1.0), (6.0, "SECOND", 20.0), (10.0, "FIRST", 2.0)]))
+
+    assert history.value_at("FIRST", 12.0) == 2.0
+    assert history.value_at("SECOND", 12.0) == 20.0
+
+
+def test_records_are_stably_sorted_and_last_duplicate_timestamp_wins() -> None:
+    history = ParameterHistory.from_log_data(_log_data([(20.0, "TEST", 3.0), (10.0, "TEST", 1.0), (10.0, "TEST", 2.0)]))
+
+    assert history.value_at("TEST", 9.0) == 1.0
+    assert history.value_at("TEST", 10.0) == 2.0
+    assert history.value_at("TEST", 15.0) == 2.0
+    assert history.value_at("TEST", 20.0) == 3.0
+
+
+def test_scaled_timeus_is_queried_in_seconds() -> None:
+    log_data = _log_data([(5_000_000.0, "TEST", 1.0), (10_000_000.0, "TEST", 2.0)])
+    schema = log_data.schemas["PARM"]
+    schema.stored_units[0] = "µs"
+    schema.multipliers[0] = 1e-6
+
+    history = ParameterHistory.from_log_data(log_data)
+
+    assert history.value_at("TEST", 9.0) == 1.0
+    assert history.value_at("TEST", 10.0) == 2.0
+
+
+def test_incomplete_records_are_ignored() -> None:
+    log_data = _log_data([(5.0, "TEST", 1.0)])
+    log_data.add_message_columns(
+        "PARM",
+        np.array([("TEST", 2.0)], dtype=[("Name", "U16"), ("Value", "f8")]),
+        MessageSchema(
+            name="PARM",
+            msg_type=1,
+            length=1,
+            format="nf",
+            fields=["Name", "Value"],
+            stored_units=["", ""],
+            scaled_units=["", ""],
+            multipliers=[None, None],
+            multipliers_applied_at_ingest=[False, False],
+            records=1,
+        ),
+    )
+
+    assert ParameterHistory.from_log_data(log_data).value_at("TEST", 10.0) is None
+
+
+@pytest.mark.parametrize("time_s", [math.nan, math.inf, -math.inf])
+def test_non_finite_record_timestamp_is_rejected(time_s: float) -> None:
+    with pytest.raises(ValueError, match="PARM timestamp for TEST must be finite"):
+        ParameterHistory.from_log_data(_log_data([(time_s, "TEST", 1.0)]))
+
+
+@pytest.mark.parametrize("time_s", [math.nan, math.inf, -math.inf])
+def test_non_finite_query_timestamp_is_rejected(time_s: float) -> None:
+    history = ParameterHistory.from_log_data(_log_data([(5.0, "TEST", 1.0)]))
+
+    with pytest.raises(ValueError, match="Parameter query time_s must be finite"):
+        history.value_at("TEST", time_s)