kota-technician-tool/kota_tech/core/logline.py

"""Serial byte stream -> framed, cleaned, parsed log lines.

ESP-IDF log format: "<L> (<ms since boot>) <TAG>: <message>", optionally
wrapped in ANSI colour codes. ROM/bootloader output (boot banner, reset
reason) has no level prefix and is classified separately, since a boot
banner is the only reliable reboot marker.

Firmware 1.0.16+ adds structured diagnostic lines logged under the "KOTA"
tag ("I (812) KOTA: boot fw=1.0.16 build=4f10989e ... last_restart=none")
and prefixes every ESP_LOGE message with a stable error ID ("[E-SD-03] ").
Both are parsed here so the diagnostics engine reads fields directly
instead of matching message text.
"""

from __future__ import annotations

import re
import time
from dataclasses import dataclass, field
from enum import Enum

from .redact import redact

ANSI_RE = re.compile(r"\x1b\[[0-9;]*[A-Za-z]")
ESP_LOG_RE = re.compile(r"^([EWIDV]) \((\d+)\) ([^:]{1,40}?): ?(.*)$")
# ESP32-S3 ROM prints e.g. "ESP-ROM:esp32s3-20210327" then
# "rst:0x1 (POWERON),boot:0x8 (SPI_FAST_FLASH_BOOT)".
BOOT_BANNER_RE = re.compile(r"^ESP-ROM:")
RESET_REASON_RE = re.compile(r"^rst:0x([0-9a-fA-F]+) \(([A-Z0-9_]+)\)")
# "[E-SD-03] Failed to initialize the card ..." (firmware error_ids.csv).
ERROR_ID_RE = re.compile(r"^\[(E-[A-Z0-9]+-\d{2,})\] ?")
KOTA_TAG = "KOTA"
PANIC_RE = re.compile(
    r"Guru Meditation Error|abort\(\) was called|ESP_ERROR_CHECK failed|"
    r"^Backtrace:|Rebooting\.\.\.$"
)


class LineKind(str, Enum):
    LOG = "log"          # ESP_LOGx line
    BOOT = "boot"        # ROM boot banner (start of a new boot)
    RESET = "reset"      # ROM reset-reason line
    PANIC = "panic"      # crash / abort output
    KOTA = "kota"        # structured "KOTA: <record> key=value ..." line
    RAW = "raw"          # anything else (task list rows, line protocol, ...)
    HOST = "host"        # event generated by this tool (connect, reset, ...)


def host_event(text: str) -> LogLine:
    return LogLine(f"--- {text} ---", LineKind.HOST)


@dataclass(frozen=True)
class LogLine:
    text: str                      # cleaned + redacted
    kind: LineKind
    host_time: float = field(default_factory=time.time)
    level: str | None = None       # E/W/I/D/V for LOG lines
    uptime_ms: int | None = None
    tag: str | None = None
    message: str | None = None
    reset_reason: str | None = None
    error_id: str | None = None    # "E-SD-03" when the message carries one
    record: str | None = None      # KOTA lines: "boot", "i2c", "health", ...
    fields: dict[str, str] = field(default_factory=dict)   # KOTA key=value pairs
    words: tuple[str, ...] = ()    # KOTA bare words ("verify", "OK", "->", ...)

    @property
    def match_text(self) -> str:
        """Text for message rules: the error-ID prefix removed, so rules
        written for older firmware keep matching."""
        if self.error_id and self.message is not None and self.tag:
            return self.text.replace(f"[{self.error_id}] ", "", 1)
        return self.text


def parse_kota(body: str) -> tuple[str, dict[str, str], tuple[str, ...]]:
    """'rtc verify running=1 delta=2s ok=1' -> ('rtc', {...}, ('verify',)).

    Values never contain spaces. The mic line has a parenthesised aside
    ("(GPIO3=1, expected 1)"): brackets and commas are stripped from keys
    and values, and the aside's words are kept as bare words.
    """
    tokens = body.split()
    if not tokens:
        return "", {}, ()
    fields: dict[str, str] = {}
    words: list[str] = []
    for tok in tokens[1:]:
        key, eq, value = tok.partition("=")
        if eq and key.strip("(,"):
            key = key.lstrip("(")
            # "0x68=DS1307(missing)" keeps its bracket; "(GPIO3=1," loses it.
            value = value.rstrip(",")
            if tok.startswith("(") or value.endswith(")") and "(" not in value:
                value = value.rstrip(")")
            fields[key] = value
        else:
            words.append(tok.strip("(),"))
    return tokens[0], fields, tuple(w for w in words if w)


def clean(raw: str) -> str:
    return ANSI_RE.sub("", raw).replace("\r", "").rstrip()


def parse_line(raw: str, host_time: float | None = None) -> LogLine:
    text = redact(clean(raw))
    t = time.time() if host_time is None else host_time
    if m := ESP_LOG_RE.match(text):
        level, ms, tag, msg = m.groups()
        tag = tag.strip()
        if tag == KOTA_TAG:
            record, fields, words = parse_kota(msg)
            return LogLine(text, LineKind.KOTA, t, level, int(ms), tag, msg,
                           record=record, fields=fields, words=words)
        eid = None
        if em := ERROR_ID_RE.match(msg):
            eid, msg = em.group(1), msg[em.end():]
        return LogLine(text, LineKind.LOG, t, level, int(ms), tag, msg, error_id=eid)
    if BOOT_BANNER_RE.match(text):
        return LogLine(text, LineKind.BOOT, t)
    if m := RESET_REASON_RE.match(text):
        return LogLine(text, LineKind.RESET, t, reset_reason=m.group(2))
    if PANIC_RE.search(text):
        return LogLine(text, LineKind.PANIC, t)
    return LogLine(text, LineKind.RAW, t)


class LineFramer:
    """Accumulates bytes and yields complete lines.

    A partial line is released after `idle_flush_s` without new bytes so
    prompts / unterminated output are not held back forever.

    Lines are also split where a new log line starts in the middle
    ("I (2302) chip[DL]: I (2302) chip[DL]: Posting..."): the two S3 cores
    interleave output, and the USB-Serial-JTAG FIFO keeps the tail of an old
    line while no host reads, releasing it glued in front of the next live
    output. After a (re)open, the fragment in front of the first such split
    is stale and dropped. Timing cannot identify it, because the FIFO only
    flushes when the board next prints, possibly seconds later.
    """

    def __init__(self, idle_flush_s: float = 0.3, max_line: int = 4096,
                 drop_first_partial: bool = False):
        self._buf = bytearray()
        self._last_rx = 0.0
        self.idle_flush_s = idle_flush_s
        self.max_line = max_line
        self._first_line_pending = drop_first_partial

    def cancel_drop(self) -> None:
        self._first_line_pending = False

    def feed(self, data: bytes, now: float | None = None) -> list[str]:
        now = time.monotonic() if now is None else now
        if data:
            self._buf.extend(data)
            self._last_rx = now
        out: list[str] = []
        while (i := self._buf.find(b"\n")) >= 0:
            out += self._emit(self._buf[:i])
            del self._buf[: i + 1]
        if len(self._buf) >= self.max_line:
            out += self._emit(self._buf)
            self._buf.clear()
        return out

    def poll_idle(self, now: float | None = None) -> list[str]:
        now = time.monotonic() if now is None else now
        if self._buf and now - self._last_rx >= self.idle_flush_s:
            return self.flush()
        return []

    def flush(self) -> list[str]:
        if not self._buf:
            return []
        out = self._emit(self._buf)
        self._buf.clear()
        return out

    def _emit(self, raw: bytes | bytearray) -> list[str]:
        parts = split_glued(bytes(raw).decode("utf-8", errors="replace"))
        if self._first_line_pending:
            self._first_line_pending = False
            if len(parts) > 1:
                parts = parts[1:]  # stale FIFO fragment from before the open
        return parts


# Start of a log line (optionally ANSI-coloured) or ROM banner.
_LINE_START = re.compile(r"(?:\x1b\[[0-9;]*m)?(?:[EWIDV] \(\d+\) \S|ESP-ROM:)")


def split_glued(text: str) -> list[str]:
    # Left-to-right scan so an ANSI colour prefix stays with its line.
    cuts = [m.start() for m in _LINE_START.finditer(text) if m.start() > 0]
    if not cuts:
        return [text]
    bounds = [0, *cuts, len(text)]
    return [text[a:b] for a, b in zip(bounds, bounds[1:]) if a < b]