px4: add events parsing & show events with ulog_messages

The implementation mainly comes from:
- PX4/flight_review#221
- PX4/flight_review#246

.gitmodules

@@ -0,0 +1,3 @@
+[submodule "3rd_party/libevents"]
+	path = 3rd_party/libevents
+	url = https://github.com/mavlink/libevents.git

pylintrc

@@ -56,7 +56,7 @@ confidence=
 # --enable=similarities". If you want to run only the classes checker, but have
 # no Warning level messages displayed, use"--disable=all --enable=classes
 # --disable=W"
-disable=suppressed-message,fixme,trailing-newlines,locally-disabled,multiple-statements,too-few-public-methods,file-ignored,len-as-condition,useless-object-inheritance,consider-using-f-string
+disable=suppressed-message,fixme,trailing-newlines,locally-disabled,multiple-statements,too-few-public-methods,file-ignored,len-as-condition,useless-object-inheritance,consider-using-f-string,unused-argument
 
 
 [REPORTS]

pyulog/libevents_parse

@@ -0,0 +1 @@
+../3rd_party/libevents/libs/python/libevents_parse

pyulog/messages.py

@@ -6,6 +6,7 @@
 import argparse
 
 from .core import ULog
+from .px4_events import PX4Events
 #pylint: disable=invalid-name
 
 def main():
@@ -20,15 +21,29 @@ def main():
     ulog_file_name = args.filename
     disable_str_exceptions = args.ignore
 
-    msg_filter = [] # we don't need the data messages
+    msg_filter = ['event']
     ulog = ULog(ulog_file_name, msg_filter, disable_str_exceptions)
 
+    logged_messages = [(m.timestamp, m.log_level_str(), m.message) for m in ulog.logged_messages]
 
-    for m in ulog.logged_messages:
-        m1, s1 = divmod(int(m.timestamp/1e6), 60)
+    # If this is a PX4 log, try to get the events too
+    if ulog.msg_info_dict.get('sys_name', '') == 'PX4':
+        px4_events = PX4Events()
+        events = px4_events.get_logged_events(ulog)
+
+        for t, log_level, message in logged_messages:
+            # backwards compatibility: a string message with appended tab is output
+            # in addition to an event with the same message so we can ignore those
+            if message[-1] == '\t':
+                continue
+            events.append((t, log_level, message))
+
+        logged_messages = sorted(events, key=lambda m: m[0])
+
+    for t, log_level, message in logged_messages:
+        m1, s1 = divmod(int(t/1e6), 60)
         h1, m1 = divmod(m1, 60)
-        print("{:d}:{:02d}:{:02d} {:}: {:}".format(
-            h1, m1, s1, m.log_level_str(), m.message))
+        print("{:d}:{:02d}:{:02d} {:}: {:}".format(h1, m1, s1, log_level, message))
 
 
 

pyulog/px4_events.py

@@ -0,0 +1,130 @@
+""" Event parsing """
+import json
+import lzma
+import urllib.request
+from typing import Optional, Callable, Any, List, Tuple
+
+from .libevents_parse.parser import Parser
+from .core import ULog
+
+
+class PX4Events:
+    """ class to extract events from logs and combine them with metadata to get the messages """
+
+    DEFAULT_EVENTS_URL = \
+        'https://px4-travis.s3.amazonaws.com/Firmware/master/_general/all_events.json.xz'
+
+    def __init__(self):
+        self._events_profile = 'dev'
+        self._default_parser: Optional[Parser] = None
+        self._get_default_json_def_cb = self._get_default_json_definitions
+
+    @staticmethod
+    def _get_default_json_definitions(already_has_default_parser: bool) -> Optional[Any]:
+        """ Default implementation for retrieving the default json event definitions """
+
+        # If it already exists, return it to avoid re-downloading
+        if already_has_default_parser:
+            return None
+
+        with urllib.request.urlopen(PX4Events.DEFAULT_EVENTS_URL, timeout=4) as response:
+            data = response.read()
+            return json.loads(lzma.decompress(data))
+
+    def set_default_json_definitions_cb(self,
+        default_json_definitions_cb: Callable[[bool], Optional[Any]]):
+        """ Set the callback to retrieve the default event definitions json
+         data (can be used for caching) """
+        self._get_default_json_def_cb = default_json_definitions_cb
+
+    def _get_event_parser(self, ulog: ULog) -> Optional[Parser]:
+        """ get event parser instance or None on error """
+
+        if 'metadata_events' in ulog.msg_info_multiple_dict and \
+                'metadata_events_sha256' in ulog.msg_info_dict:
+            file_hash = ulog.msg_info_dict['metadata_events_sha256']
+            if len(file_hash) <= 64 and file_hash.isalnum():
+                events_metadata = ulog.msg_info_multiple_dict['metadata_events'][0]
+                event_definitions_json = json.loads(lzma.decompress(b''.join(events_metadata)))
+                parser = Parser()
+                parser.load_definitions(event_definitions_json)
+                parser.set_profile(self._events_profile)
+                return parser
+
+        # No json definitions in the log -> use default definitions
+        json_definitions = self._get_default_json_def_cb(
+            self._default_parser is not None)
+        if json_definitions is not None:
+            self._default_parser = Parser()
+            self._default_parser.load_definitions(json_definitions)
+            self._default_parser.set_profile(self._events_profile)
+
+        return self._default_parser
+
+    def get_logged_events(self, ulog: ULog) -> List[Tuple[int, str, str]]:
+        """
+        Get the events as list of messages
+        :return: list of (timestamp, log level str, message) tuples
+        """
+
+        def event_log_level_str(log_level: int):
+            return {0: 'EMERGENCY',
+                    1: 'ALERT',
+                    2: 'CRITICAL',
+                    3: 'ERROR',
+                    4: 'WARNING',
+                    5: 'NOTICE',
+                    6: 'INFO',
+                    7: 'DEBUG',
+                    8: 'PROTOCOL',
+                    9: 'DISABLED'}.get(log_level, 'UNKNOWN')
+
+        # Parse events
+        messages = []
+        try:
+            events = ulog.get_dataset('event')
+            all_ids = events.data['id']
+
+            if len(all_ids) == 0:
+                return []
+
+            # Get the parser
+            try:
+                event_parser = self._get_event_parser(ulog)
+            except Exception as exception:  # pylint: disable=broad-exception-caught
+                print('Failed to get event parser: {}'.format(exception))
+                return []
+
+            for event_idx, event_id in enumerate(all_ids):
+                log_level = (events.data['log_levels'][event_idx] >> 4) & 0xf
+                if log_level >= 8:
+                    continue
+                args = []
+                i = 0
+                while True:
+                    arg_str = 'arguments[{}]'.format(i)
+                    if arg_str not in events.data:
+                        break
+                    arg = events.data[arg_str][event_idx]
+                    args.append(arg)
+                    i += 1
+                log_level_str = event_log_level_str(log_level)
+                t = events.data['timestamp'][event_idx]
+                event = None
+                if event_parser is not None:
+                    event = event_parser.parse(event_id, bytes(args))
+                if event is None:
+                    messages.append((t, log_level_str,
+                                     '[Unknown event with ID {:}]'.format(event_id)))
+                else:
+                    # only show default group
+                    if event.group() == "default":
+                        messages.append((t, log_level_str, event.message()))
+                # we could expand this a bit for events:
+                # - show the description too
+                # - handle url's as link (currently it's shown as text, and all tags are escaped)
+        except (KeyError, IndexError, ValueError):
+            # no events in log
+            pass
+
+        return messages

test/sample_log_small_messages.txt

@@ -0,0 +1,3 @@
+0:00:22 INFO: [commander] Takeoff detected
+0:00:23 INFO: [commander] Landing detected
+0:00:25 INFO: [commander] Disarmed by landing

test/sample_logging_tagged_and_default_params_messages.txt

@@ -0,0 +1,4 @@
+0:00:00 INFO: [px4] Startup script returned successfully
+0:00:00 INFO: logging: opening log file 2022-4-29/8_45_27.ulg
+0:00:00 INFO: [logger] Start file log (type: full)
+0:00:00 INFO: [logger] Opened full log file: ./log/2022-04-29/08_45_27.ulg

test/sample_px4_events_messages.txt

@@ -0,0 +1,18 @@
+475214:49:10 INFO: logging: opening log file 2024-3-18/14_49_10.ulg
+475214:49:10 INFO: [px4] Startup script returned successfully
+475214:49:10 INFO: [logger] Start file log (type: full)
+475214:49:10 INFO: [logger] Opened full log file: ./log/2024-03-18/14_49_10.ulg
+475214:49:10 INFO: [mavlink] partner IP: 127.0.0.1
+475214:49:11 WARNING: [health_and_arming_checks] Preflight: GPS fix too low
+475214:49:18 INFO: [tone_alarm] home set
+475214:49:18 WARNING: [health_and_arming_checks] Preflight: GPS fix too low
+475214:49:21 INFO: [commander] Ready for takeoff!
+475214:49:25 INFO: Armed by internal command
+475214:49:25 INFO: Using default takeoff altitude: 2.50 m
+475214:49:25 INFO: [tone_alarm] arming warning
+475214:49:27 INFO: Takeoff detected
+475214:49:32 INFO: RTL: start return at 491 m (3 m above destination)
+475214:49:32 INFO: RTL: land at destination
+475214:49:38 INFO: Landing detected
+475214:49:40 INFO: Disarmed by landing
+475214:49:40 INFO: [tone_alarm] notify neutral

test/test_cli.py

@@ -94,7 +94,7 @@ def test_extract_gps_dump_cli(self):
         ]
         extract_gps_dump.main()
 
-    @data('sample', 'sample_appended')
+    @data('sample', 'sample_appended', 'sample_px4_events')
     def test_messages_cli(self, test_case):
         """
         Test that the output of 'ulog_messages' on sample logs match previously generated results.

test/test_px4_events.py

@@ -0,0 +1,47 @@
+"""
+Tests the PX4Events class
+"""
+
+import os
+import inspect
+import unittest
+
+from ddt import ddt, data
+
+import pyulog
+from pyulog.px4_events import PX4Events
+
+TEST_PATH = os.path.dirname(os.path.abspath(
+    inspect.getfile(inspect.currentframe())))
+
+@ddt
+class TestPX4Events(unittest.TestCase):
+    """
+    Tests the PX4Events class
+    """
+
+    @data('sample_px4_events')
+    def test_px4_events(self, base_name):
+        """
+        Test that the PX4 events are extracted.
+        """
+        ulog_file_name = os.path.join(TEST_PATH, base_name + '.ulg')
+        ulog = pyulog.ULog(ulog_file_name)
+        px4_ulog = PX4Events()
+
+        def default_json_definitions_cb(already_has_default_parser: bool):
+            raise AssertionError('Must use definitions from logs')
+        px4_ulog.set_default_json_definitions_cb(default_json_definitions_cb)
+
+        messages = px4_ulog.get_logged_events(ulog)
+        expected_messages = [
+            (1710773350346000, 'INFO', 'logging: opening log file 2024-3-18/14_49_10.ulg'),
+            (1710773365282000, 'INFO', 'Armed by internal command'),
+            (1710773365282000, 'INFO', 'Using default takeoff altitude: 2.50 m'),
+            (1710773367094000, 'INFO', 'Takeoff detected'),
+            (1710773372482000, 'INFO', 'RTL: start return at 491 m (3 m above destination)'),
+            (1710773372694000, 'INFO', 'RTL: land at destination'),
+            (1710773378482000, 'INFO', 'Landing detected'),
+            (1710773380486000, 'INFO', 'Disarmed by landing')
+        ]
+        assert messages == expected_messages