Parse the drone's ulogcat CKCM log

The format is Parrot's own, from ulogcat/libulogcat_ckcm.c, which was deleted from Parrot-Developers/ulog in 2017; the module docstring cites it and records what the 560 KB capture settles that the source leaves open, notably that this firmware's stamps are monotonic microseconds since boot with no wall clock anywhere. Robustness is the point as much as the layout. The log is live and grew from 445 KB to 560 KB inside a minute, so a fetch landing mid-record is the normal case and ends the parse quietly. The markers are not escaped either, so the lengths are what the parser trusts. Parses the capture to the byte: 5520 entries, no resyncs, no residue.
2026-10-02 03:40:02 -06:00
parent bdfb2a9cc5
commit 38322b6885
2 changed files with 654 additions and 0 deletions
@@ -0,0 +1,276 @@
+"""The ulogcat parser: framing, the optional fields, and what a live log does to it.
+
+Frames are built here from the format rather than copied from the aircraft,
+because the aircraft's log is operational data. The one test that reads the
+real capture is skipped wherever that capture is not present, which is
+everywhere but the machine it was fetched on.
+"""
+
+import struct
+from pathlib import Path
+
+import pytest
+
+from mcbebop.files import ulog
+
+SAMPLE = Path(__file__).resolve().parents[1] / "captures" / "logs" / "ckcm.bin"
+
+
+def pstr(text: str) -> bytes:
+    raw = text.encode()
+    assert len(raw) < 256
+    return bytes([len(raw)]) + raw
+
+
+def frame(kind: int, body: bytes) -> bytes:
+    return ulog.FRAME_START + bytes([kind]) + body + ulog.FRAME_END
+
+
+def header(
+    *,
+    priority: str = "I",
+    uptime_us: int = 1_000_000,
+    tid: int = 42,
+    source: str | None = None,
+    tag: str | None = "TEST",
+    colour: bytes | None = None,
+) -> bytes:
+    flags = ulog.FLAG_DATE | ulog.FLAG_TID
+    if colour is not None:
+        flags |= ulog.FLAG_COLOUR
+    if source is not None:
+        flags |= ulog.FLAG_NAME
+    if tag is not None:
+        flags |= ulog.FLAG_TAG
+    body = bytes([flags, ord(priority)])
+    if colour is not None:
+        body += colour
+    body += struct.pack("<Qi", uptime_us, tid)
+    if source is not None:
+        body += pstr(source)
+    if tag is not None:
+        body += pstr(tag)
+    return frame(ulog.KIND_HEADER, body)
+
+
+def message(text: str, colour: bytes | None = None) -> bytes:
+    if colour is None:
+        return frame(ulog.KIND_MESSAGE, pstr(text))
+    return frame(ulog.KIND_MESSAGE_COLOURED, colour + bytes([ulog.KIND_MESSAGE]) + pstr(text))
+
+
+def entry(text: str = "hello", **kw) -> bytes:
+    return header(**kw) + message(text, colour=kw.get("colour"))
+
+
+# --- the framing -------------------------------------------------------------
+
+
+def test_an_entry_is_a_header_frame_and_a_message_frame():
+    (record,) = ulog.parse(entry("Machine: Milos board", tag="KERNEL", priority="W"))
+    assert record.tag == "KERNEL"
+    assert record.priority == "W"
+    assert record.level == "warning"
+    assert record.message == "Machine: Milos board"
+
+
+def test_the_timestamp_is_microseconds_since_boot():
+    (record,) = ulog.parse(entry(uptime_us=2_067_211_079))
+    assert record.uptime_us == 2_067_211_079
+    assert record.uptime == pytest.approx(2067.211079)
+
+
+def test_both_name_fields_are_optional_and_independent():
+    data = (
+        entry("has both", source="dragon-prog/Behaviour", tag="COMMANDS")
+        + entry("tag only", source=None, tag="KERNEL")
+        + entry("source only", source="ephemerisd", tag=None)
+    )
+    records = list(ulog.parse(data))
+    assert [(r.source, r.tag) for r in records] == [
+        ("dragon-prog/Behaviour", "COMMANDS"),
+        ("", "KERNEL"),
+        ("ephemerisd", ""),
+    ]
+
+
+def test_the_thread_id_comes_through():
+    (record,) = ulog.parse(entry(tid=1200))
+    assert record.tid == 1200
+
+
+def test_a_coloured_entry_keeps_the_headers_colour_and_still_reads_its_text():
+    # The data frame clamps each channel to a minimum of 1, so its copy of the
+    # colour disagrees with the header's for any channel the drone set to 0.
+    # The header's is the true one.
+    data = header(colour=b"\xff\xd0\x00") + message("Switch AWB to AUTO", colour=b"\xff\xd0\x01")
+    (record,) = ulog.parse(data)
+    assert record.colour == b"\xff\xd0\x00"
+    assert record.message == "Switch AWB to AUTO"
+
+
+def test_an_uncoloured_entry_has_no_colour():
+    (record,) = ulog.parse(entry("plain"))
+    assert record.colour is None
+
+
+def test_a_message_of_the_maximum_length_round_trips():
+    text = "x" * 255
+    (record,) = ulog.parse(entry(text))
+    assert record.message == text
+
+
+def test_the_trailing_newline_the_drone_printed_is_dropped_but_inner_ones_stay():
+    (record,) = ulog.parse(entry("first\nsecond\n"))
+    assert record.message == "first\nsecond"
+
+
+def test_every_entry_reports_where_in_the_file_it_came_from():
+    first = entry("a")
+    records = list(ulog.parse(first + entry("b")))
+    assert [r.offset for r in records] == [0, len(first)]
+
+
+# --- a log that is still being written ---------------------------------------
+
+
+def test_a_fetch_that_lands_mid_record_keeps_everything_before_it():
+    data = entry("complete") + entry("cut in half")
+    for cut in range(len(entry("complete")) + 1, len(data)):
+        stats = ulog.ParseStats()
+        records = list(ulog.parse(data[:cut], stats))
+        assert [r.message for r in records] == ["complete"], f"truncated at {cut}"
+        assert stats.truncated or stats.unpaired, f"truncation at {cut} went unreported"
+
+
+def test_a_truncation_in_the_very_first_record_yields_nothing_and_does_not_raise():
+    stats = ulog.ParseStats()
+    assert list(ulog.parse(entry("x")[:6], stats)) == []
+    assert stats.truncated
+
+
+def test_a_fetch_that_starts_mid_record_resyncs_on_the_next_marker():
+    data = entry("first") + entry("second")
+    stats = ulog.ParseStats()
+    records = list(ulog.parse(data[5:], stats))
+    assert [r.message for r in records] == ["second"]
+
+
+def test_garbage_between_entries_costs_one_resync_and_nothing_else():
+    # Nothing is pending here, so an unreadable frame is damage rather than a
+    # binary payload, and reading it as damage is what stops it swallowing the
+    # header that follows.
+    data = entry("before") + ulog.FRAME_START + b"\x99junk" + ulog.FRAME_END + entry("after")
+    stats = ulog.ParseStats()
+    records = list(ulog.parse(data, stats))
+    assert [r.message for r in records] == ["before", "after"]
+    assert stats.resyncs == 1
+
+
+def test_a_header_with_no_message_after_it_is_counted_not_guessed():
+    stats = ulog.ParseStats()
+    records = list(ulog.parse(header() + entry("real"), stats))
+    assert [r.message for r in records] == ["real"]
+    assert stats.unpaired == 1
+
+
+def test_a_message_with_no_header_before_it_is_dropped():
+    stats = ulog.ParseStats()
+    records = list(ulog.parse(message("orphan") + entry("real"), stats))
+    assert [r.message for r in records] == ["real"]
+    assert stats.unpaired == 1
+
+
+@pytest.mark.parametrize("bit", [ulog.FLAG_PC, ulog.FLAG_THREAD_PRIORITY, 0x80])
+def test_a_header_whose_opt_bits_describe_a_shape_no_source_documents_is_refused(bit):
+    # The renderer never writes PC or thread-priority, so nothing public says
+    # how wide they are. An unread field makes every field after it wrong, so
+    # skipping the record beats reporting a plausible lie.
+    flags = ulog.FLAG_DATE | ulog.FLAG_TID | ulog.FLAG_TAG | bit
+    bad = ulog.FRAME_START + bytes([ulog.KIND_HEADER, flags, ord("I")])
+    bad += struct.pack("<Qi", 1, 1) + pstr("TAG") + ulog.FRAME_END
+    stats = ulog.ParseStats()
+    records = list(ulog.parse(bad + message("orphaned by the refusal") + entry("real"), stats))
+    assert [r.message for r in records] == ["real"]
+    assert stats.resyncs == 1
+
+
+def test_a_binary_payload_is_kept_rather_than_losing_the_entry():
+    # A ulog_bin entry's data frame has no kind byte and no length, so there
+    # is nothing to decode and nothing to check it against. Keeping the bytes
+    # and saying so beats dropping the entry and its header with it.
+    payload = b"\x00\x01\x02\x03\x04"
+    data = header(tag="BIN") + ulog.FRAME_START + payload + ulog.FRAME_END
+    stats = ulog.ParseStats()
+    (record,) = ulog.parse(data, stats)
+    assert record.binary == payload
+    assert record.tag == "BIN"
+    assert "binary" in record.message
+    assert stats.resyncs == 0
+
+
+def test_a_buffer_with_no_marker_at_all_is_empty_rather_than_an_error():
+    stats = ulog.ParseStats()
+    assert list(ulog.parse(b"not a ulogcat file at all", stats)) == []
+    assert not stats.truncated
+
+
+def test_undecodable_text_does_not_stop_the_parse():
+    body = bytes([ulog.FLAG_TAG, ord("I")]) + struct.pack("<QI", 1, 1) + pstr("T")
+    data = frame(ulog.KIND_HEADER, body) + frame(ulog.KIND_MESSAGE, b"\x03\xff\xfe\xfd")
+    (record,) = ulog.parse(data)
+    assert record.message  # replacement characters, not an exception
+
+
+# --- priorities --------------------------------------------------------------
+
+
+def test_priority_letters_spell_out_to_syslog_levels():
+    assert ulog.PRIORITY_NAMES["E"] == "error"
+    assert ulog.rank("E") < ulog.rank("W") < ulog.rank("I") < ulog.rank("D")
+
+
+def test_there_is_no_notice_level_because_the_renderer_writes_notice_as_info():
+    # cprio[] in libulogcat_ckcm.c maps both ULOG_NOTICE and ULOG_INFO to 'I',
+    # with a FIXME saying the CKCM tooling does not know NOTICE. Offering a
+    # notice level here would be offering a filter that can never match.
+    assert "N" not in ulog.PRIORITY_NAMES
+
+
+def test_an_unknown_priority_letter_survives_and_never_filters_out_silently():
+    (record,) = ulog.parse(entry(priority="Z"))
+    assert record.priority == "Z"
+    assert record.level == "Z"
+    assert ulog.rank("Z") == ulog.UNKNOWN_RANK
+
+
+# --- statistics --------------------------------------------------------------
+
+
+def test_the_tag_histogram_covers_everything_parsed():
+    data = entry("a", tag="KERNEL") + entry("b", tag="KERNEL") + entry("c", tag="COMMANDS")
+    stats = ulog.ParseStats()
+    list(ulog.parse(data, stats))
+    assert stats.entries == 3
+    assert stats.tags == {"KERNEL": 2, "COMMANDS": 1}
+
+
+# --- the real thing ----------------------------------------------------------
+
+
+@pytest.mark.skipif(not SAMPLE.exists(), reason="no captured log on this machine")
+def test_the_captured_log_parses_to_the_byte():
+    # The framing was derived from this file, so the bar is total: every
+    # header consumed exactly, nothing skipped, no residue. A resync here
+    # would mean a field shape that was missed.
+    stats = ulog.ParseStats()
+    records = list(ulog.parse(SAMPLE.read_bytes(), stats))
+    assert stats.resyncs == 0
+    assert stats.unpaired == 0
+    assert len(records) == stats.entries > 1000
+    assert all(r.priority in ulog.PRIORITY_NAMES for r in records)
+    assert "KERNEL" in stats.tags
+    # Monotonic apart from the kernel stream, which ulogcat merges in from
+    # /proc/kmsg and which can land a few milliseconds out of order.
+    userspace = [r.uptime_us for r in records if r.tag != "KERNEL"]
+    assert userspace == sorted(userspace)