Optimize large log parsing and GUI responsiveness

src/mlog_tool/parser.py

@@ -3,6 +3,7 @@
 from __future__ import annotations
 
 import struct
+from dataclasses import dataclass
 from pathlib import Path
 from typing import Any, BinaryIO
 
@@ -21,6 +22,20 @@ from .models import (
 from .utils import clean_c_string, normalize_field_name
 
 
+@dataclass(slots=True)
+class _ElementLayout:
+    reader: struct.Struct
+    value_count: int
+
+
+@dataclass(slots=True)
+class _BusLayout:
+    column_names: tuple[str, ...]
+    element_layouts: tuple[_ElementLayout, ...]
+    column_data: list[list[Any]]
+    timestamp_field: str | None
+
+
 class MLogParser:
     """
     Parse an MLog file into structured Python objects.
@@ -112,29 +127,25 @@ class MLogParser:
     def _read_messages(self, stream: BinaryIO, header: LogHeader) -> dict[str, BusFrame]:
         """Scan the payload message area and aggregate valid messages by bus."""
 
-        buses_by_id = {bus.msg_id: bus for bus in header.buses}
-        records_by_bus: dict[str, list[dict[str, Any]]] = {bus.name: [] for bus in header.buses}
+        layouts_by_id = self._prepare_bus_layouts(header.buses)
 
         while self._seek_next_message(stream):
             msg_id_raw = stream.read(1)
             if not msg_id_raw:
                 break
 
-            bus = buses_by_id.get(msg_id_raw[0])
-            if bus is None:
+            layout = layouts_by_id.get(msg_id_raw[0])
+            if layout is None:
                 continue
 
-            row: dict[str, Any] = {}
+            row_values: list[Any] = []
             try:
-                for element_index, element in enumerate(bus.elements):
-                    values = self._read_typed_values(stream, element.type_id, element.count)
-                    column_name = normalize_field_name(element.name) or f"field_{element_index}"
-
-                    if element.count == 1:
-                        row[column_name] = values[0]
+                for element_layout in layout.element_layouts:
+                    values = element_layout.reader.unpack(self._read_exact(stream, element_layout.reader.size))
+                    if element_layout.value_count == 1:
+                        row_values.append(values[0])
                     else:
-                        for value_index, value in enumerate(values):
-                            row[f"{column_name}_{value_index}"] = value
+                        row_values.extend(values)
             except EOFError:
                 break
 
@@ -144,26 +155,30 @@ class MLogParser:
             if msg_end_raw[0] != MLOG_END_MSG:
                 continue
 
-            records_by_bus[bus.name].append(row)
+            for column_values, value in zip(layout.column_data, row_values):
+                column_values.append(value)
 
-        return self._build_bus_frames(header.buses, records_by_bus)
+        return self._build_bus_frames(header.buses, layouts_by_id)
 
     def _build_bus_frames(
         self,
         bus_definitions: list[BusDefinition],
-        records_by_bus: dict[str, list[dict[str, Any]]],
+        layouts_by_id: dict[int, _BusLayout],
     ) -> dict[str, BusFrame]:
-        """Convert aggregated message rows into DataFrames."""
+        """Convert aggregated column buffers into DataFrames."""
 
         buses: dict[str, BusFrame] = {}
 
         for bus in bus_definitions:
-            rows = records_by_bus.get(bus.name, [])
-            if not rows:
+            layout = layouts_by_id.get(bus.msg_id)
+            if layout is None or not layout.column_data or not layout.column_data[0]:
                 continue
 
-            frame = pd.DataFrame(rows)
-            timestamp_field = self._find_timestamp_field(bus)
+            frame = pd.DataFrame(
+                {column_name: values for column_name, values in zip(layout.column_names, layout.column_data)},
+                copy=False,
+            )
+            timestamp_field = layout.timestamp_field
 
             if timestamp_field and timestamp_field in frame.columns:
                 relative_time_name = "time_s"
@@ -181,6 +196,41 @@ class MLogParser:
 
         return buses
 
+    def _prepare_bus_layouts(self, bus_definitions: list[BusDefinition]) -> dict[int, _BusLayout]:
+        """Precompute column layouts to avoid per-message schema work."""
+
+        layouts_by_id: dict[int, _BusLayout] = {}
+
+        for bus in bus_definitions:
+            column_names: list[str] = []
+            element_layouts: list[_ElementLayout] = []
+
+            for element_index, element in enumerate(bus.elements):
+                if element.type_id not in TYPE_STRUCT_FORMATS:
+                    raise ValueError(f"Unsupported MLog type id: {element.type_id}")
+
+                column_name = normalize_field_name(element.name) or f"field_{element_index}"
+                if element.count == 1:
+                    column_names.append(column_name)
+                else:
+                    column_names.extend(f"{column_name}_{value_index}" for value_index in range(element.count))
+
+                element_layouts.append(
+                    _ElementLayout(
+                        reader=struct.Struct("<" + TYPE_STRUCT_FORMATS[element.type_id] * element.count),
+                        value_count=element.count,
+                    )
+                )
+
+            layouts_by_id[bus.msg_id] = _BusLayout(
+                column_names=tuple(column_names),
+                element_layouts=tuple(element_layouts),
+                column_data=[[] for _ in column_names],
+                timestamp_field=self._find_timestamp_field(bus),
+            )
+
+        return layouts_by_id
+
     def _find_timestamp_field(self, bus: BusDefinition) -> str | None:
         """Return the normalized timestamp field name if the bus exposes one."""
 

src/mlog_tool/qt_gui.py

@@ -15,7 +15,7 @@ from .services import MLogService
 QT_IMPORT_ERROR: Exception | None = None
 
 try:
-    from PySide6.QtCore import QAbstractTableModel, QModelIndex, Qt
+    from PySide6.QtCore import QAbstractTableModel, QModelIndex, Qt, QThread, Signal
     from PySide6.QtGui import QAction, QColor, QBrush, QFont
     from PySide6.QtWidgets import (
         QApplication,
@@ -46,6 +46,25 @@ if QT_IMPORT_ERROR is None:
     ITEM_VALUE_ROLE = Qt.ItemDataRole.UserRole + 21
     ITEM_PARENT_ROLE = Qt.ItemDataRole.UserRole + 22
 
+    class ParseThread(QThread):
+        """Parse a log file off the UI thread."""
+
+        parse_succeeded = Signal(object)
+        parse_failed = Signal(str)
+
+        def __init__(self, log_path: Path) -> None:
+            super().__init__()
+            self.log_path = Path(log_path)
+
+        def run(self) -> None:
+            try:
+                parsed_log = MLogService().parse(self.log_path)
+            except Exception as exc:
+                self.parse_failed.emit(str(exc))
+                return
+
+            self.parse_succeeded.emit(parsed_log)
+
     class DataFrameTableModel(QAbstractTableModel):
         """Lightweight table model for large pandas DataFrames."""
 
@@ -121,26 +140,28 @@ if QT_IMPORT_ERROR is None:
             self.subplot_count = 2
             self.subplot_traces: list[list[PlotTraceSpec]] = [[], []]
             self._updating_tree = False
+            self._parse_thread: ParseThread | None = None
             self.preview_model = DataFrameTableModel()
 
             self._build_ui()
+            self._update_action_states()
 
         def _build_ui(self) -> None:
             toolbar = QToolBar("Main")
             toolbar.setMovable(False)
             self.addToolBar(toolbar)
 
-            open_action = QAction("Open Log", self)
-            open_action.triggered.connect(self.open_log_file)
-            toolbar.addAction(open_action)
+            self.open_action = QAction("Open Log", self)
+            self.open_action.triggered.connect(self.open_log_file)
+            toolbar.addAction(self.open_action)
 
-            export_action = QAction("Export CSV", self)
-            export_action.triggered.connect(self.export_csv)
-            toolbar.addAction(export_action)
+            self.export_action = QAction("Export CSV", self)
+            self.export_action.triggered.connect(self.export_csv)
+            toolbar.addAction(self.export_action)
 
-            refresh_action = QAction("Refresh Plot", self)
-            refresh_action.triggered.connect(self.refresh_plot)
-            toolbar.addAction(refresh_action)
+            self.refresh_action = QAction("Refresh Plot", self)
+            self.refresh_action.triggered.connect(self.refresh_plot)
+            toolbar.addAction(self.refresh_action)
 
             cursor_action = QAction("Cursor", self)
             cursor_action.setCheckable(True)
@@ -259,12 +280,21 @@ if QT_IMPORT_ERROR is None:
                 QMessageBox.information(self, "No file", "Select a log file first.")
                 return
 
-            try:
-                self.parsed_log = self.service.parse(self.selected_log)
-            except Exception as exc:
-                QMessageBox.critical(self, "Parse failed", str(exc))
+            if self._parse_thread is not None and self._parse_thread.isRunning():
                 return
 
+            self._clear_loaded_log()
+            self.summary_label.setText("Parsing in background...")
+            self._parse_thread = ParseThread(self.selected_log)
+            self._parse_thread.parse_succeeded.connect(self._on_parse_succeeded)
+            self._parse_thread.parse_failed.connect(self._on_parse_failed)
+            self._parse_thread.finished.connect(self._on_parse_finished)
+            self._parse_thread.finished.connect(self._parse_thread.deleteLater)
+            self._parse_thread.start()
+            self._update_action_states()
+
+        def _on_parse_succeeded(self, parsed_log: ParsedLog) -> None:
+            self.parsed_log = parsed_log
             self.subplot_traces = [[], []]
             self._populate_tree()
             self.summary_label.setText(
@@ -285,6 +315,14 @@ if QT_IMPORT_ERROR is None:
                 self.signal_tree.setCurrentItem(first_item)
                 self.on_tree_selection_changed()
 
+        def _on_parse_failed(self, error_message: str) -> None:
+            self.summary_label.setText("Parse failed.")
+            QMessageBox.critical(self, "Parse failed", error_message)
+
+        def _on_parse_finished(self) -> None:
+            self._parse_thread = None
+            self._update_action_states()
+
         def export_csv(self) -> None:
             if self.parsed_log is None:
                 QMessageBox.information(self, "No data", "Parse a log file first.")
@@ -321,6 +359,24 @@ if QT_IMPORT_ERROR is None:
             self._refresh_tree_state()
             self.refresh_plot()
 
+        def _clear_loaded_log(self) -> None:
+            self.parsed_log = None
+            self.subplot_traces = [[], []]
+            self.signal_tree.clear()
+            self.preview.setPlainText("")
+            self._clear_preview_table()
+            self.refresh_plot()
+
+        def _update_action_states(self) -> None:
+            is_parsing = self._parse_thread is not None and self._parse_thread.isRunning()
+            has_data = self.parsed_log is not None
+            self.open_action.setEnabled(not is_parsing)
+            self.export_action.setEnabled(has_data and not is_parsing)
+            self.refresh_action.setEnabled(has_data and not is_parsing)
+            self.signal_tree.setEnabled(not is_parsing)
+            self.preview_table.setEnabled(not is_parsing)
+            self.preview.setEnabled(not is_parsing)
+
         def _toggle_cursor(self, enabled: bool) -> None:
             self.plot_widget.set_cursor_enabled(enabled)
 
@@ -669,6 +725,12 @@ if QT_IMPORT_ERROR is None:
                     self.preview_table.resizeColumnToContents(column_index)
                     header.setSectionResizeMode(column_index, QHeaderView.ResizeMode.Interactive)
 
+        def closeEvent(self, event: Any) -> None:  # noqa: N802
+            if self._parse_thread is not None and self._parse_thread.isRunning():
+                self.summary_label.setText("Waiting for parsing to finish before closing...")
+                self._parse_thread.wait()
+            super().closeEvent(event)
+
 
 def main() -> None:
     """Run the Qt GUI entry point."""

tests/test_parser.py

@@ -1,6 +1,8 @@
+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
@@ -9,6 +11,46 @@ 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
@@ -43,6 +85,8 @@ def test_parsed_log_parameter_lookup_returns_none_when_missing() -> 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)
 
@@ -102,3 +146,20 @@ def test_plotter_can_compare_traces_across_subplots() -> None:
     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)