refactor: separated parsing and extraction

Signed-off-by: Omkar Sarkar <omkarsarkar24@gmail.com>

Author: OmkarSarkar204

ardupilot_methodic_configurator/log_analysis/backend_log_extraction.py

@@ -11,7 +11,7 @@
 import re
 from typing import Any
 
-from pymavlink import mavutil
+from ardupilot_methodic_configurator.log_analysis.backend_log_parsing import parse_log
 
 PARAM_NAME_REGEX = r"^[A-Z][A-Z_0-9]*$"
 PARAM_NAME_MAX_LEN = 16
@@ -27,37 +27,6 @@ def is_param_name_format_valid(pname: str) -> bool:
     return bool(re.match(PARAM_NAME_REGEX, pname))
 
 
-def open_log(logfile: str) -> mavutil.mavfile:
-    """
-    Open an ArduPilot log file.
-
-    Args:
-      logfile: The path to an Ardupilot .bin log file.
-
-    Returns:
-      A mavutil.mavfile connection object.
-
-    """
-    try:
-        mlog = mavutil.mavlink_connection(logfile)
-    except Exception as e:
-        msg = f"Error opening the {logfile} logfile: {e!s}"
-        raise OSError(msg) from e
-    return mlog  # pyright: ignore[reportReturnType]  # pymavlink stubs include CSVReader which doesn't extend mavfile
-
-
-def close_log(mlog: mavutil.mavfile) -> None:
-    """
-    Close an ArduPilot log file.
-
-    Args:
-      mlog: The mavutil.mavfile connection to close.
-
-    """
-    with contextlib.suppress(OSError):
-        mlog.close()
-
-
 class BatteryData:  # pylint: disable=too-many-instance-attributes,too-few-public-methods
     """Stores battery telemetry data extracted from BAT log messages."""
 
@@ -212,78 +181,70 @@ def extract_log(logfile: str) -> LogData:  # pylint: disable=too-many-branches
         logfile: The path to an ArduPilot .bin log file.
 
     Returns:
-        A populated LogData object containing parameters, firmware info, message counts, and frame type.
+        A populated LogData object containing parameters, firmware info,
+        message counts, and frame type.
 
     """
     log_data = LogData()
     message_counts: dict[str, int] = {}
     firmware_from_ver: tuple[str, int, int, int] | None = None
     firmware_from_msg: tuple[str, int, int, int] | None = None
 
-    mlog = open_log(logfile)
-
-    try:
-        while True:
-            msg = mlog.recv_match()
-            if msg is None:
-                break
-            msg_type = msg.get_type()
-            message_counts[msg_type] = message_counts.get(msg_type, 0) + 1
+    for msg in parse_log(logfile):
+        msg_type = msg.get_type()
+        message_counts[msg_type] = message_counts.get(msg_type, 0) + 1
 
-            # Extract PARM messages and store them, also is used in extract_param_defaults.py
-            if msg_type == "PARM":
-                process_param(msg, log_data)
+        # Extract PARM messages and store them, also is used in extract_param_defaults.py
+        if msg_type == "PARM":
+            process_param(msg, log_data)
 
-            # Extract the Version with Vehicle_type, Major, Minor and Patch.
-            elif msg_type == "VER":
-                firmware_from_ver = process_ver(msg)
+        # Extract the Version with Vehicle_type, Major, Minor and Patch.
+        elif msg_type == "VER":
+            firmware_from_ver = process_ver(msg)
 
-            # Fallback to MSG if version is not available.
-            elif msg_type == "MSG":
-                firmware_from_msg = process_msg_version_fallback(msg, firmware_from_msg)
+        # Fallback to MSG if version is not available.
+        elif msg_type == "MSG":
+            firmware_from_msg = process_msg_version_fallback(msg, firmware_from_msg)
 
-            # Extract the BAT messages and store them in BatteryData
-            elif msg_type == "BAT":
-                process_bat(msg, log_data)
+        # Extract the BAT messages and store them in BatteryData
+        elif msg_type == "BAT":
+            process_bat(msg, log_data)
 
-            # Extract the PM messages and store them in PMData
-            elif msg_type == "PM":
-                process_performance(msg, log_data)
+        # Extract the PM messages and store them in PMData
+        elif msg_type == "PM":
+            process_performance(msg, log_data)
 
-            # Extract IMU messages and store them in IMUData
-            elif msg_type == "IMU":
-                process_imu(msg, log_data)
+        # Extract IMU messages and store them in IMUData
+        elif msg_type == "IMU":
+            process_imu(msg, log_data)
 
-            # Extract VIBE messages and store them in VibeData
-            elif msg_type == "VIBE":
-                process_vibe(msg, log_data)
+        # Extract VIBE messages and store them in VibeData
+        elif msg_type == "VIBE":
+            process_vibe(msg, log_data)
 
-            # Extract GPS messages and store them in GPSData
-            elif msg_type == "GPS":
-                process_gps(msg, log_data)
+        # Extract GPS messages and store them in GPSData
+        elif msg_type == "GPS":
+            process_gps(msg, log_data)
 
-            # Extract ERR message and store them in ERRData
-            elif msg_type == "ERR":
-                process_err(msg, log_data)
+        # Extract ERR message and store them in ERRData
+        elif msg_type == "ERR":
+            process_err(msg, log_data)
 
-            # Extract ARM messages and store them in ARMStat
-            elif msg_type == "ARM":
-                process_arm_stat(msg, log_data)
+        # Extract ARM messages and store them in ARMStat
+        elif msg_type == "ARM":
+            process_arm_stat(msg, log_data)
 
-            # Extract MODE messages and store them in Mode
-            elif msg_type == "MODE":
-                process_mode(msg, log_data)
+        # Extract MODE messages and store them in Mode
+        elif msg_type == "MODE":
+            process_mode(msg, log_data)
 
-        if firmware_from_ver is not None:
-            log_data.firmware_info = firmware_from_ver
-        else:
-            log_data.firmware_info = firmware_from_msg
-
-        process_frame_type(log_data)
-        log_data.messages = message_counts
+    if firmware_from_ver is not None:
+        log_data.firmware_info = firmware_from_ver
+    else:
+        log_data.firmware_info = firmware_from_msg
 
-    finally:
-        close_log(mlog)
+    process_frame_type(log_data)
+    log_data.messages = message_counts
 
     return log_data
 

ardupilot_methodic_configurator/log_analysis/backend_log_parsing.py

@@ -0,0 +1,71 @@
+"""
+A parser for ArduPilot .bin log files.
+
+The ArduPilot .bin format is self contained: FMT messages define the schema of every message type.
+Pymavlink reads those FMT definitions and decodes each message accordingly.
+
+SPDX-FileCopyrightText: 2024-2026 Amilcar do Carmo Lucas <amilcar.lucas@iav.de>
+
+SPDX-License-Identifier: GPL-3.0-or-later
+"""
+
+import contextlib
+from typing import Any
+
+from pymavlink import mavutil
+
+
+def open_log(logfile: str) -> mavutil.mavfile:
+    """
+    Open an ArduPilot log file.
+
+    Args:
+        logfile: The path to an ArduPilot .bin log file.
+
+    Returns:
+        A mavutil.mavfile connection object.
+
+    """
+    try:
+        mlog = mavutil.mavlink_connection(logfile)
+    except Exception as e:
+        msg = f"Error opening the {logfile} logfile: {e!s}"
+        raise OSError(msg) from e
+    return mlog  # pyright: ignore[reportReturnType]
+
+
+def close_log(mlog: mavutil.mavfile) -> None:
+    """
+    Close an ArduPilot log file.
+
+    Args:
+        mlog: The mavutil.mavfile connection to close.
+
+    """
+    with contextlib.suppress(OSError):
+        mlog.close()
+
+
+def parse_log(logfile: str) -> Any:  # noqa: ANN401
+    """
+    Open a log and yield each DataFlash message.
+
+    Each message is yielded as decoded by pymavlink (using the file's own FMT
+    definitions) and discarded.
+
+    Args:
+        logfile: The path to an ArduPilot .bin log file.
+
+    Yields:
+        One DataFlash log message per iteration.
+
+    """
+    mlog = open_log(logfile)
+    try:
+        while True:
+            msg = mlog.recv_match()
+            if msg is None:
+                break
+            yield msg
+    finally:
+        close_log(mlog)