mlog_tool/tests/test_parser.py

import struct
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 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_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)