mlog_tool/tests/test_parser.py

from pathlib import Path

import pandas as pd

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 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"

    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