diff --git a/src/mcbebop/files/ulog.py b/src/mcbebop/files/ulog.py new file mode 100644 index 0000000..5467429 --- /dev/null +++ b/src/mcbebop/files/ulog.py @@ -0,0 +1,378 @@ +"""Parrot's ulogcat CKCM log, as firmware 4.7.1 writes it. + +The aircraft keeps its system log at `internal_000/Debug/current/ckcm/ckcm.bin`, +produced by `/usr/bin/ckcm_log.sh` running `ulogcat -uk -v ckcm`. It is the +only place several facts about the machine are stated without a shell: which +board it is, which config it loaded, what its command handler thought it was +doing. So it is worth reading, and reading it needs the framing. + +CKCM is a serial-debug format from Parrot's car-kit era, and the renderer that +writes it was deleted from `Parrot-Developers/ulog` in October 2017. The file +`ulogcat/libulogcat_ckcm.c` at the commit before `cec8877` is the reference +for everything below, and it says in a comment that it copies its constants +because the real CKCM headers were never published. Independently, the +framing was derived from a 560 KB sample off this aircraft and holds across +all of it: 5520 entries, every header consumed to the byte, no residue. + + frame := D5 E6 E5 F6 + +A log entry is TWO frames written back to back in one `write()`: a header +frame and then the data frame carrying its message. There is no file header +and no rotation framing; the file is the raw concatenation. + + header (kind E7) := opt:u8 priority:u8 [colour:3 if opt&01] + stamp:u64le tid:i32le + [tname:pstr if opt&20] [tag:pstr if opt&40] + + data (kind 02) := pstr + data (kind 24) := colour:3 02 pstr + data (other) := raw bytes, for a ulog_bin entry + +`pstr` is one length byte then that many bytes, so a message is capped at 255 +bytes and anything longer is truncated by the drone, not by us. + +`priority` is a single ASCII letter, and it is lossy on purpose: the renderer +maps EMERG, ALERT and CRIT all to `C`, and NOTICE and INFO both to `I`, so a +notice cannot be told from an info here. + +`stamp` is microseconds. Which clock it counts from depends on the ulogger +driver build, and this firmware's answer is legible in the data: userspace +stamps top out around 2067 s rather than 1.7e9, and they agree to a constant +8 ms with the uptime stamps 779 of the `ARlibs` messages carry in their own +text. So this aircraft logs **monotonic time since boot**, and there is no +wall clock anywhere in the file. Kernel entries carry the printk uptime, which +is the same clock but merged in from `/proc/kmsg` and so can land a few +milliseconds out of order against the userspace stream. + +There is no pid field. Only `tid` is stored, and the process name is smuggled +into `tname` as `process/thread` when the two differ, which is why a `/` in +`source` is the only signal that an entry came from a named thread rather than +a process main. Both strings are genuinely optional: kernel messages carry a +tag (`KERNEL`) and no name, a few processes carry a name and no tag. `-l` is +not passed, so tags are raw and `tag == "KERNEL"` is the only way to tell a +kernel entry from a userspace one. + +The three colour bytes are RGB, and the drone colours its camera and ISP +output: 23 of the 5520 sample entries carry them, white for ulogcat's own +banner lines and oranges for the rest. The data frame clamps each channel to a +minimum of 1, so a header colour of `ff d0 00` arrives as `ff d0 01` there; +the header's value is the true one and the one kept. + +Two opt bits, `02` (PC) and `10` (thread priority), are defined but never +written by this renderer, so no public source says what they would look like. +A header carrying either is refused rather than guessed at. + +Robustness is not optional here: the log is live. The sample grew from 445 KB +to 560 KB inside a minute, so a fetch lands mid-record as a matter of course +and a truncated tail has to end the parse quietly rather than raise. Nor are +the markers escaped, which is why the lengths are what this parser trusts and +the end marker is only what it checks them against. +""" + +from __future__ import annotations + +import logging +import struct +from collections.abc import Iterator +from dataclasses import dataclass, field + +logger = logging.getLogger(__name__) + +FRAME_START = b"\xd5\xe6" +FRAME_END = b"\xe5\xf6" + +KIND_HEADER = 0xE7 # CKCM calls this PLOG +KIND_MESSAGE = 0x02 # RT_STR +KIND_MESSAGE_COLOURED = 0x24 # RT_COLOR, then an RT_STR + +#: Not a byte in the file. A ulog_bin entry's data frame carries no kind and +#: no length, so this parser needs a name for "the frame was raw payload". +KIND_BINARY = -1 + +FLAG_COLOUR = 0x01 +FLAG_PC = 0x02 # never written by this renderer; layout unknown +FLAG_DATE = 0x04 +FLAG_TID = 0x08 +FLAG_THREAD_PRIORITY = 0x10 # never written by this renderer; layout unknown +FLAG_NAME = 0x20 +FLAG_TAG = 0x40 + +#: Bits whose wire shape no public source describes. A header that sets one +#: cannot be read, and reading the fields after it would be a guess. +FLAGS_UNSUPPORTED = FLAG_PC | FLAG_THREAD_PRIORITY | 0x80 + +COLOUR_LEN = 3 + +#: The letters the renderer writes, spelled out. `C` also covers EMERG and +#: ALERT, and `I` also covers NOTICE, so neither is recoverable from here. +PRIORITY_NAMES = { + "C": "critical", + "E": "error", + "W": "warning", + "I": "info", + "D": "debug", +} + +#: Syslog severity numbers, so "at least a warning" can be a comparison. +#: Lower is more severe, and an unknown letter sorts as the least severe +#: thing there is so that a filter never silently drops it. +PRIORITY_RANK = {"C": 2, "E": 3, "W": 4, "I": 6, "D": 7} +UNKNOWN_RANK = 99 + + +def rank(priority: str) -> int: + return PRIORITY_RANK.get(priority, UNKNOWN_RANK) + + +@dataclass(frozen=True) +class LogRecord: + """One log entry, the header and message frames put back together.""" + + uptime_us: int + priority: str + """Syslog level as the single letter the file stores. `level` spells it.""" + tag: str + source: str + """`process/thread`, or empty when the entry carries no source name.""" + tid: int + message: str + """Text, or a short stand-in when the entry's payload was binary.""" + offset: int + """Byte offset of the entry's header frame, for pointing at a bad record.""" + colour: bytes | None = None + """RGB the drone asked for, when it asked. Camera and ISP code does.""" + binary: bytes | None = None + """Raw payload of a ulog_bin entry, which carries no text and no length.""" + + @property + def uptime(self) -> float: + """Seconds since the drone booted.""" + return self.uptime_us / 1_000_000 + + @property + def level(self) -> str: + return PRIORITY_NAMES.get(self.priority, self.priority) + + +@dataclass +class ParseStats: + """What the parse had to put up with. Worth reporting on a live log.""" + + entries: int = 0 + resyncs: int = 0 + """Frames that did not decode, each costing a hunt for the next marker.""" + truncated: bool = False + """The buffer ended mid-record, which is normal for a log still being written.""" + unpaired: int = 0 + """Headers with no message frame after them, or the reverse.""" + tags: dict[str, int] = field(default_factory=dict) + + +class _Truncated(Exception): + """The frame runs past the end of the buffer.""" + + +class _Malformed(Exception): + """The frame decoded to something that is not a frame.""" + + +def _pstr(data: bytes, pos: int) -> tuple[str, int]: + if pos >= len(data): + raise _Truncated + length = data[pos] + end = pos + 1 + length + if end > len(data): + raise _Truncated + return data[pos + 1 : end].decode("utf-8", "replace"), end + + +def _frame(data: bytes, start: int, *, expect_data: bool) -> tuple[int, int, int]: + """Locate the frame whose start marker is at `start`. + + Returns `(kind, body_start, body_end)`, with `KIND_BINARY` for a payload + that carries no kind byte. For every other frame the end marker is + verified, which is the only check that the lengths were read correctly: + get them wrong and the two bytes that should close the frame are not + there. + + A binary payload has no kind, no length and so nothing to check, which + makes it indistinguishable from corruption. `expect_data` is what keeps + that from costing anything: a data frame only ever follows a header, so + only there is an unreadable frame read as a payload. Anywhere else it is + treated as damage and the caller resyncs, which is what keeps a corrupt + region from swallowing the next header along with it. + """ + pos = start + len(FRAME_START) + if pos >= len(data): + raise _Truncated + kind = data[pos] + body = pos + 1 + + if kind == KIND_HEADER: + end = _header_end(data, body) + elif kind == KIND_MESSAGE: + _, end = _pstr(data, body) + elif ( + kind == KIND_MESSAGE_COLOURED + and body + COLOUR_LEN < len(data) + and data[body + COLOUR_LEN] == KIND_MESSAGE + ): + _, end = _pstr(data, body + COLOUR_LEN + 1) + elif expect_data: + return _binary_frame(data, pos) + else: + raise _Malformed + + if end + len(FRAME_END) > len(data): + raise _Truncated + if data[end : end + len(FRAME_END)] != FRAME_END: + raise _Malformed + return kind, body, end + + +def _binary_frame(data: bytes, body: int) -> tuple[int, int, int]: + end = data.find(FRAME_END, body) + if end < 0: + raise _Truncated + return KIND_BINARY, body, end + + +def _header_end(data: bytes, body: int) -> int: + pos = body + 2 # opt, priority + if pos > len(data): + raise _Truncated + flags = data[body] + if flags & FLAGS_UNSUPPORTED: + raise _Malformed + if flags & FLAG_COLOUR: + pos += COLOUR_LEN + pos += 12 # stamp u64, tid i32 + if pos > len(data): + raise _Truncated + if flags & FLAG_NAME: + _, pos = _pstr(data, pos) + if flags & FLAG_TAG: + _, pos = _pstr(data, pos) + return pos + + +def _header(data: bytes, body: int, end: int) -> tuple[str, str, str, int, int, bytes | None]: + flags = data[body] + priority = chr(data[body + 1]) + pos = body + 2 + colour = None + if flags & FLAG_COLOUR: + colour = data[pos : pos + COLOUR_LEN] + pos += COLOUR_LEN + stamp_us, tid = struct.unpack_from(" str: + pos = body + COLOUR_LEN + 1 if kind == KIND_MESSAGE_COLOURED else body + text, _ = _pstr(data, pos) + # Most messages end in a newline the drone's own printf put there. + return text.rstrip("\n") + + +def parse(data: bytes, stats: ParseStats | None = None) -> Iterator[LogRecord]: + """Walk a ulogcat buffer, yielding one record per log entry. + + Never raises on bad input. A frame that does not decode costs a hunt for + the next start marker, and a buffer that ends mid-record ends the walk; + both are counted in `stats` if one is passed, because on a live log the + difference between "the drone logged nothing more" and "we caught it + mid-write" is worth telling the caller about. + """ + stats = stats if stats is not None else ParseStats() + pos = data.find(FRAME_START) + if pos < 0: + return + pending: tuple[int, tuple] | None = None + + while pos >= 0: + try: + kind, body, end = _frame(data, pos, expect_data=pending is not None) + except _Truncated: + # Only the tail of a live file should land here. If a start marker + # follows, the "truncation" was really a misread frame. + nxt = data.find(FRAME_START, pos + len(FRAME_START)) + if nxt < 0: + stats.truncated = True + break + stats.resyncs += 1 + pos = nxt + continue + except _Malformed: + stats.resyncs += 1 + pos = data.find(FRAME_START, pos + len(FRAME_START)) + continue + + if kind == KIND_HEADER: + if pending is not None: + stats.unpaired += 1 + try: + pending = (pos, _header(data, body, end)) + except (_Truncated, _Malformed): + stats.resyncs += 1 + pending = None + elif pending is None: + stats.unpaired += 1 + else: + offset, (priority, tag, source, tid, stamp_us, colour) = pending + pending = None + binary = data[body:end] if kind == KIND_BINARY else None + if binary is None: + try: + message = _message(data, kind, body) + except (_Truncated, _Malformed): # pragma: no cover - _frame checked it + stats.resyncs += 1 + pos = end + len(FRAME_END) + continue + else: + message = f"<{len(binary)} bytes of binary log payload>" + stats.entries += 1 + stats.tags[tag] = stats.tags.get(tag, 0) + 1 + yield LogRecord( + uptime_us=stamp_us, + priority=priority, + tag=tag, + source=source, + tid=tid, + message=message, + offset=offset, + colour=colour, + binary=binary, + ) + + pos = end + len(FRAME_END) + if pos >= len(data): + break + if data[pos : pos + len(FRAME_START)] != FRAME_START: + nxt = data.find(FRAME_START, pos) + if nxt < 0: + stats.truncated = True + break + stats.resyncs += 1 + pos = nxt + + if pending is not None: + stats.unpaired += 1 + if stats.resyncs or stats.unpaired: + logger.debug( + "ulog parse: %d entries, %d resyncs, %d unpaired, truncated=%s", + stats.entries, + stats.resyncs, + stats.unpaired, + stats.truncated, + ) diff --git a/tests/test_ulog.py b/tests/test_ulog.py new file mode 100644 index 0000000..3b50608 --- /dev/null +++ b/tests/test_ulog.py @@ -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(" 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(" 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)