mlog_tool/tests/test_parser.py

import struct
import warnings
from pathlib import Path

import pandas as pd
import pytest

from mlog_tool.constants import MLOG_BEGIN_MSG1, MLOG_BEGIN_MSG2, MLOG_END_MSG
from mlog_tool.models import BusFrame, LogHeader, ParsedLog, PlotTraceSpec
from mlog_tool.parser import MLogParser
from mlog_tool.plotting import LogPlotter
from mlog_tool.utils import normalize_field_name


def _pack_fixed_string(value: str, length: int) -> bytes:
    raw = value.encode("utf-8")[:length]
    return raw + (b"\x00" * (length - len(raw)))


def _write_synthetic_log(path: Path, message_count: int = 2048) -> None:
    max_name_len = 16
    max_desc_len = 24
    max_model_info_len = 24

    payload = bytearray()
    payload.extend(struct.pack("<H I H H H", 1, 123456789, max_name_len, max_desc_len, max_model_info_len))
    payload.extend(_pack_fixed_string("synthetic", max_desc_len))
    payload.extend(_pack_fixed_string("unit-test", max_model_info_len))

    payload.extend(struct.pack("<B", 1))
    payload.extend(_pack_fixed_string("IMU0", max_name_len))
    payload.extend(struct.pack("<B B", 7, 3))

    payload.extend(_pack_fixed_string("timestamp_ms", max_name_len))
    payload.extend(struct.pack("<H H", 5, 1))

    payload.extend(_pack_fixed_string("ax", max_name_len))
    payload.extend(struct.pack("<H H", 6, 1))

    payload.extend(_pack_fixed_string("gyro", max_name_len))
    payload.extend(struct.pack("<H H", 6, 3))

    payload.extend(struct.pack("<B", 0))

    for index in range(message_count):
        payload.extend(bytes([MLOG_BEGIN_MSG1, MLOG_BEGIN_MSG2, 7]))
        payload.extend(struct.pack("<I", index * 10))
        payload.extend(struct.pack("<f", index * 0.5))
        payload.extend(struct.pack("<fff", index + 0.1, index + 0.2, index + 0.3))
        payload.extend(bytes([MLOG_END_MSG]))

    path.write_bytes(payload)


def _write_wide_synthetic_log(path: Path, field_count: int = 128, message_count: int = 32) -> None:
    max_name_len = 16
    max_desc_len = 24
    max_model_info_len = 24

    payload = bytearray()
    payload.extend(struct.pack("<H I H H H", 1, 123456789, max_name_len, max_desc_len, max_model_info_len))
    payload.extend(_pack_fixed_string("wide", max_desc_len))
    payload.extend(_pack_fixed_string("unit-test", max_model_info_len))

    payload.extend(struct.pack("<B", 1))
    payload.extend(_pack_fixed_string("WIDE", max_name_len))
    payload.extend(struct.pack("<B B", 9, field_count + 1))

    payload.extend(_pack_fixed_string("timestamp_ms", max_name_len))
    payload.extend(struct.pack("<H H", 5, 1))

    for index in range(field_count):
        payload.extend(_pack_fixed_string(f"f{index}", max_name_len))
        payload.extend(struct.pack("<H H", 6, 1))

    payload.extend(struct.pack("<B", 0))

    for row_index in range(message_count):
        payload.extend(bytes([MLOG_BEGIN_MSG1, MLOG_BEGIN_MSG2, 9]))
        payload.extend(struct.pack("<I", row_index * 20))
        for field_index in range(field_count):
            payload.extend(struct.pack("<f", row_index + field_index / 10))
        payload.extend(bytes([MLOG_END_MSG]))

    path.write_bytes(payload)


def _write_offset_bus_log(path: Path) -> None:
    max_name_len = 16
    max_desc_len = 24
    max_model_info_len = 24

    payload = bytearray()
    payload.extend(struct.pack("<H I H H H", 1, 123456789, max_name_len, max_desc_len, max_model_info_len))
    payload.extend(_pack_fixed_string("offset", max_desc_len))
    payload.extend(_pack_fixed_string("unit-test", max_model_info_len))

    payload.extend(struct.pack("<B", 2))

    payload.extend(_pack_fixed_string("IMU0", max_name_len))
    payload.extend(struct.pack("<B B", 7, 2))
    payload.extend(_pack_fixed_string("timestamp_ms", max_name_len))
    payload.extend(struct.pack("<H H", 5, 1))
    payload.extend(_pack_fixed_string("ax", max_name_len))
    payload.extend(struct.pack("<H H", 6, 1))

    payload.extend(_pack_fixed_string("GPS", max_name_len))
    payload.extend(struct.pack("<B B", 8, 2))
    payload.extend(_pack_fixed_string("timestamp_ms", max_name_len))
    payload.extend(struct.pack("<H H", 5, 1))
    payload.extend(_pack_fixed_string("ve", max_name_len))
    payload.extend(struct.pack("<H H", 6, 1))

    payload.extend(struct.pack("<B", 0))

    for timestamp_ms, value in [(1000, 1.0), (1010, 2.0)]:
        payload.extend(bytes([MLOG_BEGIN_MSG1, MLOG_BEGIN_MSG2, 7]))
        payload.extend(struct.pack("<I", timestamp_ms))
        payload.extend(struct.pack("<f", value))
        payload.extend(bytes([MLOG_END_MSG]))

    for timestamp_ms, value in [(1300, 3.0), (1310, 4.0)]:
        payload.extend(bytes([MLOG_BEGIN_MSG1, MLOG_BEGIN_MSG2, 8]))
        payload.extend(struct.pack("<I", timestamp_ms))
        payload.extend(struct.pack("<f", value))
        payload.extend(bytes([MLOG_END_MSG]))

    path.write_bytes(payload)


def test_parser_type_exists() -> None:
    parser = MLogParser()
    assert parser is not None


def test_message_markers_match_matlab_definition() -> None:
    assert MLOG_BEGIN_MSG1 == 0x92
    assert MLOG_BEGIN_MSG2 == 0x05
    assert MLOG_END_MSG == 0x26


def test_normalize_field_name() -> None:
    assert normalize_field_name("timestamp_ms") == "timestamp_ms"
    assert normalize_field_name("gyro[0]") == "gyro_0"
    assert normalize_field_name("  att roll ") == "att_roll"


def test_parsed_log_parameter_lookup_returns_none_when_missing() -> None:
    header = LogHeader(
        version=1,
        timestamp=0,
        max_name_len=16,
        max_desc_len=16,
        max_model_info_len=16,
        description="demo",
        model_info="demo",
    )
    parsed_log = ParsedLog(source_path=Path("demo.bin"), header=header)
    assert parsed_log.get_parameter("INS", "acc_bias") is None


def test_parser_can_parse_sample_log() -> None:
    parser = MLogParser()
    sample_log = Path(__file__).resolve().parents[2] / "mlog16.bin"
    if not sample_log.exists():
        pytest.skip("sample log mlog16.bin is not present in this workspace")

    parsed_log = parser.parse(sample_log)

    assert parsed_log.header.version >= 1
    assert len(parsed_log.header.buses) == 48
    assert len(parsed_log.header.parameter_groups) == 9
    assert len(parsed_log.buses) == 19
    assert "IMU0" in parsed_log.buses
    assert "GPS" in parsed_log.buses
    assert "time_s" in parsed_log.buses["IMU0"].frame.columns
    assert parsed_log.get_parameter_group("SYSTEM") is not None
    assert parsed_log.get_parameter("SYSTEM", "MAV_SYS_ID") == 2


def test_plotter_can_compare_traces_across_subplots() -> None:
    header = LogHeader(
        version=1,
        timestamp=0,
        max_name_len=16,
        max_desc_len=16,
        max_model_info_len=16,
        description="demo",
        model_info="demo",
    )
    frame = pd.DataFrame(
        {
            "time_s": [0.0, 0.1, 0.2],
            "ax": [1.0, 2.0, 3.0],
            "ay": [4.0, 5.0, 6.0],
        }
    )
    parsed_log = ParsedLog(
        source_path=Path("demo.bin"),
        header=header,
        buses={
            "IMU0": BusFrame(name="IMU0", frame=frame, timestamp_field="time_s"),
            "GPS": BusFrame(name="GPS", frame=frame.rename(columns={"ax": "ve", "ay": "vn"}), timestamp_field="time_s"),
        },
    )

    figure = LogPlotter().build_comparison_figure(
        parsed_log,
        [
            [
                PlotTraceSpec(bus_name="IMU0", field_name="ax", subplot_index=0, label="IMU0.ax"),
                PlotTraceSpec(bus_name="GPS", field_name="ve", subplot_index=0, label="GPS.ve"),
            ],
            [PlotTraceSpec(bus_name="IMU0", field_name="ay", subplot_index=1, label="IMU0.ay")],
        ],
        subplot_count=2,
    )

    assert len(figure.axes) == 2
    assert figure.axes[0].get_title() == "Subplot 1"
    assert figure.axes[1].get_title() == "Subplot 2"
    assert figure.axes[0].get_shared_x_axes().joined(figure.axes[0], figure.axes[1])
    assert len(figure._mlog_cursor_payload) == 2
    assert len(figure._mlog_cursor_payload[0]["traces"]) == 2
    assert len(figure._mlog_cursor_payload[1]["traces"]) == 1


def test_plotter_can_build_four_subplot_grid() -> None:
    header = LogHeader(
        version=1,
        timestamp=0,
        max_name_len=16,
        max_desc_len=16,
        max_model_info_len=16,
        description="demo",
        model_info="demo",
    )
    frame = pd.DataFrame(
        {
            "time_s": [0.0, 0.1, 0.2],
            "ax": [1.0, 2.0, 3.0],
            "ay": [4.0, 5.0, 6.0],
            "az": [7.0, 8.0, 9.0],
            "gx": [0.1, 0.2, 0.3],
        }
    )
    parsed_log = ParsedLog(
        source_path=Path("demo.bin"),
        header=header,
        buses={"IMU0": BusFrame(name="IMU0", frame=frame, timestamp_field="time_s")},
    )

    figure = LogPlotter().build_comparison_figure(
        parsed_log,
        [
            [PlotTraceSpec(bus_name="IMU0", field_name="ax", subplot_index=0, label="IMU0.ax")],
            [PlotTraceSpec(bus_name="IMU0", field_name="ay", subplot_index=1, label="IMU0.ay")],
            [PlotTraceSpec(bus_name="IMU0", field_name="az", subplot_index=2, label="IMU0.az")],
            [PlotTraceSpec(bus_name="IMU0", field_name="gx", subplot_index=3, label="IMU0.gx")],
        ],
        subplot_count=4,
    )

    assert len(figure.axes) == 4
    assert [axis.get_title() for axis in figure.axes] == [
        "Subplot 1",
        "Subplot 2",
        "Subplot 3",
        "Subplot 4",
    ]
    assert figure.axes[0].get_subplotspec().rowspan.start == 0
    assert figure.axes[1].get_subplotspec().rowspan.start == 0
    assert figure.axes[2].get_subplotspec().rowspan.start == 1
    assert figure.axes[3].get_subplotspec().rowspan.start == 1
    assert figure.axes[0].get_subplotspec().colspan.start == 0
    assert figure.axes[1].get_subplotspec().colspan.start == 1
    assert figure.axes[2].get_subplotspec().colspan.start == 0
    assert figure.axes[3].get_subplotspec().colspan.start == 1
    assert figure.axes[0].get_shared_x_axes().joined(figure.axes[0], figure.axes[3])
    assert len(figure._mlog_cursor_payload) == 4


def test_parser_handles_many_messages_without_losing_column_shape(tmp_path: Path) -> None:
    parser = MLogParser()
    log_path = tmp_path / "synthetic_mlog.bin"
    _write_synthetic_log(log_path, message_count=4096)

    parsed_log = parser.parse(log_path)

    imu = parsed_log.get_bus("IMU0")
    assert imu is not None
    assert imu.timestamp_field == "time_s"
    assert list(imu.frame.columns[:6]) == ["timestamp_ms", "time_s", "ax", "gyro_0", "gyro_1", "gyro_2"]
    assert len(imu.frame) == 4096
    assert imu.frame.iloc[0]["time_s"] == 0.0
    assert imu.frame.iloc[-1]["timestamp_ms"] == 40950
    assert imu.frame.iloc[-1]["gyro_2"] == pytest.approx(4095.3)


def test_parser_aligns_relative_time_to_global_first_timestamp(tmp_path: Path) -> None:
    parser = MLogParser()
    log_path = tmp_path / "offset_bus_mlog.bin"
    _write_offset_bus_log(log_path)

    parsed_log = parser.parse(log_path)

    imu = parsed_log.get_bus("IMU0")
    gps = parsed_log.get_bus("GPS")
    assert imu is not None
    assert gps is not None
    assert imu.timestamp_field == "time_s"
    assert gps.timestamp_field == "time_s"
    assert list(imu.frame["time_s"]) == pytest.approx([0.0, 0.01])
    assert list(gps.frame["time_s"]) == pytest.approx([0.3, 0.31])


def test_parser_avoids_fragmentation_warning_when_adding_relative_time(tmp_path: Path) -> None:
    parser = MLogParser()
    log_path = tmp_path / "wide_synthetic_mlog.bin"
    _write_wide_synthetic_log(log_path, field_count=128, message_count=64)

    with warnings.catch_warnings():
        warnings.simplefilter("error", pd.errors.PerformanceWarning)
        parsed_log = parser.parse(log_path)

    wide_bus = parsed_log.get_bus("WIDE")
    assert wide_bus is not None
    assert wide_bus.timestamp_field == "time_s"
    assert list(wide_bus.frame.columns[:3]) == ["timestamp_ms", "time_s", "f0"]
    assert len(wide_bus.frame.columns) == 130
    assert wide_bus.frame.iloc[-1]["time_s"] == pytest.approx(1.26)