feat: added PM, IMU, VIBE, GPS, ERR, MODE, ARM messages

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

Author: OmkarSarkar204

ardupilot_methodic_configurator/log_analysis/backend_log_extraction.py

@@ -77,7 +77,111 @@ def __init__(self) -> None:
         self.state_health: list[int] = []
 
 
-class LogData:  # pylint: disable=too-few-public-methods
+class PMData:  # pylint: disable=too-few-public-methods
+    """Stores Flight Controller's CPU performance telemetry data extracted from PM messages."""
+
+    def __init__(self) -> None:
+        self.time_us: list[int] = []
+        self.load: list[int] = []
+        self.mem: list[int] = []
+        self.loop_rate: list[int] = []
+        self.int_err_bitmask: list[int] = []
+        self.long_loops: list[int] = []
+
+
+class IMUData:  # pylint: disable=too-many-instance-attributes,too-few-public-methods
+    """
+    Stores Inertial Measurement Unit data extracted from IMU messages.
+
+    Note:
+        GyrX,Y,Z are in rad/s.
+
+    """
+
+    def __init__(self) -> None:
+        self.time_us: list[int] = []
+        self.gyr_x: list[float] = []
+        self.gyr_y: list[float] = []
+        self.gyr_z: list[float] = []
+        self.acc_x: list[float] = []
+        self.acc_y: list[float] = []
+        self.acc_z: list[float] = []
+        self.err_gyro: list[int] = []
+        self.err_acc: list[int] = []
+        self.temp: list[float] = []
+        self.gyro_hlt: list[int] = []
+        self.acc_hlt: list[int] = []
+        self.gyro_rate: list[int] = []
+        self.acc_rate: list[int] = []
+
+
+class VibeData:  # pylint: disable=too-few-public-methods
+    """Stores Processed Vibration Information data extracted from VIBE messages."""
+
+    def __init__(self) -> None:
+        self.time_us: list[int] = []
+        self.vibe_x: list[float] = []
+        self.vibe_y: list[float] = []
+        self.vibe_z: list[float] = []
+        self.clip: list[int] = []
+
+
+class GPSData:  # pylint: disable=too-many-instance-attributes,too-few-public-methods
+    """
+    Stores GPS/GNSS data extracted from GPS messages.
+
+    Note:
+        GCrs and Yaw are in degrees, unlike GyrX,Y,Z. Have to convert before any operation.
+
+    """
+
+    def __init__(self) -> None:
+        self.time_us: list[int] = []
+        self.status: list[int] = []
+        self.num_sats: list[int] = []
+        self.hor_dop: list[float] = []
+        self.lat: list[float] = []
+        self.lng: list[float] = []
+        self.alt: list[float] = []
+        self.spd: list[float] = []
+        self.gcrs: list[float] = []
+        self.vz: list[float] = []
+        self.yaw: list[float] = []
+        self.in_use: list[int] = []  # Boolean value
+        # GMS/GWk not captured; TimeUS covers rate for analysis needs
+
+
+class ERRMsg:  # pylint: disable=too-few-public-methods
+    """Stores subsystem error events extracted from ERR messages."""
+
+    # A log might not contain the ERR field
+    def __init__(self) -> None:
+        self.time_us: list[int] = []
+        self.sub_sys: list[int] = []
+        self.err_code: list[int] = []
+
+
+class ModeInfo:  # pylint: disable=too-few-public-methods
+    """Stores flight mode transition events extracted from MODE messages."""
+
+    def __init__(self) -> None:
+        self.time_us: list[int] = []
+        self.mode_num: list[int] = []
+        self.rsn: list[int] = []  # reason for mode change
+
+
+class ARMStat:  # pylint: disable=too-few-public-methods
+    """Stores arming and disarming events extracted from ARM messages."""
+
+    def __init__(self) -> None:
+        self.time_us: list[int] = []
+        self.arm_state: list[int] = []
+        self.arm_chks: list[int] = []
+        self.forced: list[bool] = []
+        self.method: list[int] = []
+
+
+class LogData:  # pylint: disable=too-many-instance-attributes,too-few-public-methods
     """Contains all data extracted from an ArduPilot .bin log."""
 
     def __init__(self) -> None:
@@ -88,9 +192,19 @@ def __init__(self) -> None:
         self.frame_type: int | None = None
         # There could be multiple batteries in the vehicle, so create a separate dict for them.
         self.batteries: dict[int, BatteryData] = {}
-
-
-def extract_log(logfile: str) -> LogData:
+        self.performance_monitor = PMData()
+        # There could be multiple IMU sensors in the vehicle, so create a separate dict for them.
+        self.imu_data: dict[int, IMUData] = {}
+        # There could be multiple VIBE instances, so create a separate dict for them.
+        self.vibe_data: dict[int, VibeData] = {}
+        # There could be multiple GPS modules in the vehicle, so create a separate dict for them.
+        self.gps_data: dict[int, GPSData] = {}
+        self.err_data = ERRMsg()
+        self.arm_info = ARMStat()
+        self.mode_info = ModeInfo()
+
+
+def extract_log(logfile: str) -> LogData:  # pylint: disable=too-many-branches
     """
     Open the log file, scan every message, and return the LogData.
 
@@ -132,6 +246,34 @@ def extract_log(logfile: str) -> LogData:
             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 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 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 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)
+
         if firmware_from_ver is not None:
             log_data.firmware_info = firmware_from_ver
         else:
@@ -157,10 +299,10 @@ def process_bat(msg: Any, log_data: LogData) -> None:  # noqa: ANN401
 
     """
     # If there are multiple battery instances store them separately
-    inst = int(msg.Inst)
-    if inst not in log_data.batteries:
-        log_data.batteries[inst] = BatteryData()
-    battery = log_data.batteries[inst]
+    bat_inst = int(msg.Inst)
+    if bat_inst not in log_data.batteries:
+        log_data.batteries[bat_inst] = BatteryData()
+    battery = log_data.batteries[bat_inst]
 
     battery.time_us.append(int(msg.TimeUS))
     battery.volt.append(float(msg.Volt))
@@ -177,6 +319,164 @@ def process_bat(msg: Any, log_data: LogData) -> None:  # noqa: ANN401
     battery.state_health.append(int(getattr(msg, "SH", 0)))
 
 
+def process_performance(msg: Any, log_data: LogData) -> None:  # noqa: ANN401
+    """
+    Extract performance telemetry from a PM DataFlash log entry.
+
+    Args:
+        msg: A PM log entry object parsed from an ArduPilot .bin file
+             (returned by mavutil.mavfile.recv_match()).
+        log_data: The LogData instance to write performance data into.
+
+    """
+    pm = log_data.performance_monitor
+
+    pm.time_us.append(int(msg.TimeUS))
+    pm.load.append(int(msg.Load))
+    pm.mem.append(int(msg.Mem))
+    pm.loop_rate.append(int(msg.LR))
+    # InE renamed from IntE between firmware 4.5.4 and 4.7.0. # codespell:ignore
+    pm.int_err_bitmask.append(int(getattr(msg, "InE", getattr(msg, "IntE", 0))))  # codespell:ignore
+    pm.long_loops.append(int(msg.NLon))
+
+
+def process_imu(msg: Any, log_data: LogData) -> None:  # noqa: ANN401
+    """
+    Extract inertial Data from an IMU DataFlash log entry.
+
+    Args:
+        msg: An IMU log entry object parsed from an ArduPilot .bin file
+             (returned by mavutil.mavfile.recv_match()).
+        log_data: The LogData instance to write IMU data into.
+
+    """
+    # If there are multiple IMU sensors
+    imu_inst = int(msg.I)
+    if imu_inst not in log_data.imu_data:
+        log_data.imu_data[imu_inst] = IMUData()
+    imu = log_data.imu_data[imu_inst]
+
+    imu.time_us.append(int(msg.TimeUS))
+    imu.gyr_x.append(float(msg.GyrX))
+    imu.gyr_y.append(float(msg.GyrY))
+    imu.gyr_z.append(float(msg.GyrZ))
+    imu.acc_x.append(float(msg.AccX))
+    imu.acc_y.append(float(msg.AccY))
+    imu.acc_z.append(float(msg.AccZ))
+    imu.err_gyro.append(int(msg.EG))
+    imu.err_acc.append(int(msg.EA))
+    imu.temp.append(float(msg.T))
+    imu.gyro_hlt.append(int(getattr(msg, "GH", 0)))
+    imu.acc_hlt.append(int(getattr(msg, "AH", 0)))
+    imu.gyro_rate.append(int(msg.GHz))
+    imu.acc_rate.append(int(msg.AHz))
+
+
+def process_vibe(msg: Any, log_data: LogData) -> None:  # noqa: ANN401
+    """
+    Extract VIBE Data from an IMU DataFlash log entry.
+
+    Args:
+        msg: An VIBE log entry object parsed from an ArduPilot .bin file
+             (returned by mavutil.mavfile.recv_match()).
+        log_data: The LogData instance to write VIBE data into.
+
+    """
+    # If there are multiple instances of IMU
+    vibe_inst = int(msg.IMU)
+    if vibe_inst not in log_data.vibe_data:
+        log_data.vibe_data[vibe_inst] = VibeData()
+    vibe = log_data.vibe_data[vibe_inst]
+
+    vibe.time_us.append(int(msg.TimeUS))
+    vibe.vibe_x.append(float(msg.VibeX))
+    vibe.vibe_y.append(float(msg.VibeY))
+    vibe.vibe_z.append(float(msg.VibeZ))
+    vibe.clip.append(int(msg.Clip))
+
+
+def process_gps(msg: Any, log_data: LogData) -> None:  # noqa: ANN401
+    """
+    Extract GPS/GNSS telemetry from a GPS DataFlash log entry.
+
+    Args:
+        msg: A GPS log entry object parsed from an ArduPilot .bin file
+             (returned by mavutil.mavfile.recv_match()).
+        log_data: The LogData instance to write GPS data into.
+
+    """
+    gps_inst = int(msg.I)
+    if gps_inst not in log_data.gps_data:
+        log_data.gps_data[gps_inst] = GPSData()
+    gps = log_data.gps_data[gps_inst]
+
+    gps.time_us.append(int(msg.TimeUS))
+    gps.status.append(int(msg.Status))
+    gps.num_sats.append(int(msg.NSats))
+    gps.hor_dop.append(float(msg.HDop))
+    gps.lat.append(float(msg.Lat))
+    gps.lng.append(float(msg.Lng))
+    gps.alt.append(float(msg.Alt))
+    gps.spd.append(float(msg.Spd))
+    gps.gcrs.append(float(msg.GCrs))
+    gps.vz.append(float(msg.VZ))
+    gps.yaw.append(float(msg.Yaw))
+    gps.in_use.append(int(msg.U))
+
+
+def process_err(msg: Any, log_data: LogData) -> None:  # noqa: ANN401
+    """
+    Extract subsystem error data from an ERR DataFlash log entry.
+
+    Args:
+        msg: An ERR log entry object parsed from an ArduPilot .bin file
+             (returned by mavutil.mavfile.recv_match()).
+        log_data: The LogData instance to write error data into.
+
+    """
+    err = log_data.err_data
+
+    err.time_us.append(int(msg.TimeUS))
+    err.sub_sys.append(int(msg.Subsys))
+    err.err_code.append(int(msg.ECode))
+
+
+def process_arm_stat(msg: Any, log_data: LogData) -> None:  # noqa: ANN401
+    """
+    Extract arming status from an ARM DataFlash log entry.
+
+    Args:
+        msg: An ARM log entry object parsed from an ArduPilot .bin file
+             (returned by mavutil.mavfile.recv_match()).
+        log_data: The LogData instance to write arming data into.
+
+    """
+    arm = log_data.arm_info
+
+    arm.time_us.append(int(msg.TimeUS))
+    arm.arm_state.append(int(msg.ArmState))
+    arm.arm_chks.append(int(msg.ArmChecks))
+    arm.forced.append(bool(msg.Forced))
+    arm.method.append(int(msg.Method))
+
+
+def process_mode(msg: Any, log_data: LogData) -> None:  # noqa: ANN401
+    """
+    Extract flight mode transition data from a MODE DataFlash log entry.
+
+    Args:
+        msg: A MODE log entry object parsed from an ArduPilot .bin file
+             (returned by mavutil.mavfile.recv_match()).
+        log_data: The LogData instance to write mode data into.
+
+    """
+    mode = log_data.mode_info
+
+    mode.time_us.append(int(msg.TimeUS))
+    mode.mode_num.append(int(msg.ModeNum))
+    mode.rsn.append(int(msg.Rsn))
+
+
 def process_param(msg: Any, log_data: LogData) -> None:  # noqa: ANN401
     """
     Validate and store a single PARM log entry into log_data's parameter dicts.