| #!/usr/bin/env python3 | |
| # | |
| # SPDX-FileCopyrightText: Copyright The Zephyr Project Contributors | |
| # | |
| # SPDX-License-Identifier: Apache-2.0 | |
| """ | |
| Zephyr CTF trace viewer. | |
| Reads a Common Trace Format (CTF) binary stream produced by Zephyr's tracing | |
| subsystem (for example the ``tracing.bin`` dumped by | |
| ``samples/subsys/tracing`` when built with the CTF format and the semihosting | |
| backend) and renders it graphically on the console - a lightweight, | |
| terminal-only take on tools such as SEGGER SystemView or Eclipse Trace Compass. | |
| The viewer focuses on visualization. Most of the screen is a Gantt-style time | |
| graph with one lane per thread (plus lanes for the idle thread and for ISRs); | |
| coloured bars show which thread is running over time and the transitions | |
| between them are the context switches. A movable time cursor (the playhead) | |
| selects a point in the trace, with a small info strip beneath the chart that | |
| can be toggled off to give the graph the whole screen. A metrics panel fills | |
| the area below the lanes with a CPU-busy gauge, context-switch / event rates | |
| and per-thread stacked utilization bars computed over the visible window. | |
| The trace can be replayed: ``space`` starts/stops autoplay, which advances the | |
| playhead in real time and scrolls the chart to follow it, and ``[`` / ``]`` | |
| adjust the playback speed. Pressing ``v`` switches to a raw CTF log view: a | |
| scrollable table of every decoded event (timestamp, id, name and fields), with | |
| the event nearest the playhead highlighted. | |
| Unlike ``babeltrace2`` this script needs no external dependencies: it decodes | |
| the packed little-endian CTF records itself. The per-event field layout is read | |
| from the TSDL ``metadata`` file that ships next to the tracing subsystem | |
| (``subsys/tracing/ctf/tsdl/metadata``) so the decoder automatically tracks any | |
| events added there; a built-in fallback table covers the core scheduling events | |
| so the time graph still works even when the metadata file cannot be found. | |
| Usage:: | |
| # interactive viewer (a terminal/TTY is required) | |
| ./trace_viewer.py build/tracing.bin | |
| # follow a trace live while the application is still writing it | |
| ./trace_viewer.py -f build/tracing.bin | |
| # non-interactive ASCII timeline, e.g. to pipe or redirect | |
| ./trace_viewer.py build/tracing.bin --text | |
| # point at a specific metadata file | |
| ./trace_viewer.py build/tracing.bin --metadata path/to/metadata | |
| """ | |
| import argparse | |
| import os | |
| import re | |
| import struct | |
| import sys | |
| def safe_open(path, mode, **kwargs): | |
| """Open a command-line file path after canonicalising and validating it. | |
| The viewer only ever reads pre-existing files named on the command line. | |
| Resolving symlinks with ``realpath`` and confirming the target is a regular | |
| file keeps a stray CLI argument (a directory, device node, or dangling | |
| symlink) from being opened, and yields a clear error instead of a deep | |
| traceback. Centralising every read here keeps the trusted/untrusted | |
| boundary in one obvious place. | |
| """ | |
| real = os.path.realpath(path) | |
| if not os.path.isfile(real): | |
| raise SystemExit(f"error: {path!r} is not a readable file") | |
| return open(real, mode, **kwargs) # noqa: SIM115 | |
| # Record framing, identical for every event: | |
| # uint64_t timestamp (ns) <- present when CONFIG_TRACING_CTF_TIMESTAMP=y | |
| # uint16_t id | |
| # <packed, byte-aligned event specific fields> | |
| HDR = struct.Struct("<QH") | |
| HDR_NO_TS = struct.Struct("<H") | |
| # Map a TSDL integer typedef to a struct format character and byte size. | |
| TYPES = { | |
| "int8_t": ("b", 1), | |
| "uint8_t": ("B", 1), | |
| "uint16_t": ("H", 2), | |
| "uint32_t": ("I", 4), | |
| "int32_t": ("i", 4), | |
| "uint64_t": ("Q", 8), | |
| } | |
| # Events that drive the running-thread timeline / lane layout. | |
| THREAD_SWITCHED_IN = 0x11 | |
| THREAD_SWITCHED_OUT = 0x10 | |
| ISR_ENTER = 0x1B | |
| ISR_EXIT = 0x1C | |
| ISR_EXIT_TO_SCHEDULER = 0x1D | |
| IDLE = 0x1E | |
| # Events carrying thread identity we want in the thread table. | |
| THREAD_INFO = 0x19 | |
| THREAD_CREATE = 0x13 | |
| THREAD_NAME_SET = 0x1A | |
| THREAD_PRIO_SET = 0x12 # k_thread_priority_set bracket (carries prio) | |
| THREAD_SCHED_PRIO_SET = 0xE9 # scheduler priority change, incl. inheritance | |
| # Minimal fallback layout for the scheduling events, used only when the TSDL | |
| # metadata file cannot be located. Each entry is (name, [(field, type), ...]). | |
| FALLBACK_EVENTS = { | |
| 0x10: ("thread_switched_out", [("thread_id", "uint32_t"), ("name", "str20")]), | |
| 0x11: ("thread_switched_in", [("thread_id", "uint32_t"), ("name", "str20")]), | |
| 0x12: ( | |
| "thread_priority_set", | |
| [("thread_id", "uint32_t"), ("name", "str20"), ("prio", "int8_t")], | |
| ), | |
| 0x13: ("thread_create", [("thread_id", "uint32_t"), ("name", "str20")]), | |
| 0x14: ("thread_abort", [("thread_id", "uint32_t"), ("name", "str20")]), | |
| 0x19: ( | |
| "thread_info", | |
| [ | |
| ("thread_id", "uint32_t"), | |
| ("name", "str20"), | |
| ("stack_base", "uint32_t"), | |
| ("stack_size", "uint32_t"), | |
| ], | |
| ), | |
| 0x1A: ("thread_name_set", [("thread_id", "uint32_t"), ("name", "str20")]), | |
| 0x1B: ("isr_enter", []), | |
| 0x1C: ("isr_exit", []), | |
| 0x1D: ("isr_exit_to_scheduler", []), | |
| 0x1E: ("idle", []), | |
| 0x7F: ("thread_sleep_enter", [("timeout", "uint32_t")]), | |
| 0x80: ("thread_sleep_exit", [("timeout", "uint32_t"), ("ret", "int32_t")]), | |
| } | |
| class EventDef: | |
| """Decoded layout for one CTF event id.""" | |
| __slots__ = ("eid", "name", "fields", "size") | |
| def __init__(self, eid, name, fields): | |
| self.eid = eid | |
| self.name = name | |
| # fields: list of (field_name, kind) where kind is a struct char for | |
| # scalars or ("str", n) for a fixed length string. Every field is a | |
| # fixed size, so the whole record body has a known length (self.size), | |
| # which the streaming reader uses to spot incomplete trailing records. | |
| self.fields = [] | |
| self.size = 0 | |
| for fname, ftype in fields: | |
| if ftype == "str20": | |
| ftype = ("str", 20) | |
| if isinstance(ftype, tuple) and ftype[0] == "str": | |
| self.fields.append((fname, ftype)) | |
| self.size += ftype[1] | |
| else: | |
| ch, _ = TYPES[ftype] | |
| self.fields.append((fname, ch)) | |
| self.size += struct.calcsize(ch) | |
| def decode(self, buf, off): | |
| """Decode fields from buf starting at off; return (dict, new_off).""" | |
| out = {} | |
| for fname, kind in self.fields: | |
| if isinstance(kind, tuple) and kind[0] == "str": | |
| n = kind[1] | |
| raw = buf[off : off + n] | |
| off += n | |
| out[fname] = raw.split(b"\x00", 1)[0].decode("ascii", "replace") | |
| else: | |
| sz = struct.calcsize(kind) | |
| (val,) = struct.unpack_from("<" + kind, buf, off) | |
| off += sz | |
| out[fname] = val | |
| return out, off | |
| def parse_metadata(path): | |
| """Parse the TSDL metadata file into {id: EventDef}.""" | |
| with safe_open(path, "r", errors="replace") as f: | |
| text = f.read() | |
| defs = {} | |
| # Match each "event { ... };" block. The body can contain a nested | |
| # "struct { ... }", so balance braces manually rather than with a | |
| # non-greedy regex (which would stop at the inner closing brace). | |
| for kw in re.finditer(r"\bevent\s*\{", text): | |
| start = kw.end() | |
| depth = 1 | |
| i = start | |
| while i < len(text) and depth: | |
| if text[i] == "{": | |
| depth += 1 | |
| elif text[i] == "}": | |
| depth -= 1 | |
| i += 1 | |
| body = text[start : i - 1] | |
| name_m = re.search(r"name\s*=\s*([A-Za-z0-9_]+)\s*;", body) | |
| id_m = re.search(r"id\s*=\s*(0[xX][0-9a-fA-F]+|\d+)\s*;", body) | |
| if not name_m or not id_m: | |
| continue | |
| eid = int(id_m.group(1), 0) | |
| name = name_m.group(1) | |
| fields = [] | |
| struct_m = re.search(r"fields\s*:=\s*struct\s*\{(.*?)\}\s*;", body, re.DOTALL) | |
| if struct_m: | |
| for fm in re.finditer( | |
| r"([A-Za-z_][A-Za-z0-9_]*)\s+([A-Za-z_][A-Za-z0-9_]*)\s*(?:\[(\d+)\])?\s*;", | |
| struct_m.group(1), | |
| ): | |
| ftype, fname, arr = fm.group(1), fm.group(2), fm.group(3) | |
| if ftype == "ctf_bounded_string_t": | |
| fields.append((fname, ("str", int(arr) if arr else 1))) | |
| elif ftype in TYPES: | |
| fields.append((fname, ftype)) | |
| # Unknown types are skipped; they would desync decoding, but the | |
| # Zephyr metadata only uses the integer aliases and strings. | |
| defs[eid] = EventDef(eid, name, fields) | |
| return defs | |
| def build_event_defs(metadata_path): | |
| if metadata_path and os.path.exists(metadata_path): | |
| try: | |
| defs = parse_metadata(metadata_path) | |
| if defs: | |
| return defs, metadata_path | |
| except Exception as exc: # pragma: no cover - defensive | |
| sys.stderr.write(f"warning: could not parse {metadata_path}: {exc}\n") | |
| # Fallback: scheduling events only. | |
| defs = {eid: EventDef(eid, name, flds) for eid, (name, flds) in FALLBACK_EVENTS.items()} | |
| return defs, None | |
| def find_metadata(binary_path, override): | |
| if override: | |
| return override | |
| rel = os.path.join("subsys", "tracing", "ctf", "tsdl", "metadata") | |
| candidates = [] | |
| # $ZEPHYR_BASE, if set. | |
| zbase = os.environ.get("ZEPHYR_BASE") | |
| if zbase: | |
| candidates.append(os.path.join(zbase, rel)) | |
| # Climb parents from this script (scripts/tracing/) looking for the source | |
| # tree metadata, so the viewer works from anywhere in the tree. | |
| d = os.path.dirname(os.path.abspath(__file__)) | |
| for _ in range(6): | |
| candidates.append(os.path.join(d, rel)) | |
| d = os.path.dirname(d) | |
| # Alongside the trace itself (the layout babeltrace2 expects). | |
| candidates.append(os.path.join(os.path.dirname(os.path.abspath(binary_path)), "metadata")) | |
| candidates.append("metadata") | |
| for c in candidates: | |
| if os.path.exists(c): | |
| return c | |
| return None | |
| class Event: | |
| __slots__ = ("ts", "eid", "name", "fields") | |
| def __init__(self, ts, eid, name, fields): | |
| self.ts = ts | |
| self.eid = eid | |
| self.name = name | |
| self.fields = fields | |
| class Trace: | |
| """Parsed trace: ordered events plus reconstructed running timeline.""" | |
| def __init__(self): | |
| self.events = [] | |
| self.threads = {} # tid -> {name, prio, stack_base, stack_size} | |
| self.segments = [] # (start_ts, end_ts, tid) running thread spans | |
| self.isr_spans = [] # (start_ts, end_ts) ISR active spans | |
| self.states = {} # tid -> [(start, end, state, reason), ...] | |
| self.state_starts = {} # tid -> [start, ...] (for bisect) | |
| self.t0 = 0 | |
| self.t1 = 0 | |
| def thread(self, tid): | |
| return self.threads.setdefault( | |
| tid, {"name": "", "prio": None, "stack_base": None, "stack_size": None} | |
| ) | |
| class TraceReader: | |
| """Incrementally decode a CTF stream into a Trace. | |
| Because every record has a known fixed length, the reader can be fed bytes | |
| in arbitrary chunks: complete records are decoded and any partial trailing | |
| record is kept until the rest arrives. This is what lets the viewer follow | |
| a tracing.bin file that the application is still writing. | |
| All derived structures (running segments, ISR spans and the per-thread | |
| state timeline) are maintained incrementally. The currently-open state of | |
| each thread is published as a provisional segment extending to the latest | |
| timestamp so the live edge renders; it is dropped and re-derived on the | |
| next feed. | |
| """ | |
| def __init__(self, defs, has_ts=True): | |
| self.defs = defs | |
| self.has_ts = has_ts | |
| self.hdr = HDR if has_ts else HDR_NO_TS | |
| self.tr = Trace() | |
| self._buf = b"" # undecoded bytes (possibly a partial record) | |
| self._fake_ts = 0 | |
| self._desync = False | |
| # Some platforms back the CTF timestamp with a free-running cycle | |
| # counter that wraps, giving a sawtooth instead of a monotonic clock. | |
| # Unwrap it: every time the raw value jumps backwards, add the previous | |
| # raw value as an offset so the timeline stays monotonic. For a trace | |
| # whose timestamps are already monotonic this is a no-op. | |
| self._prev_raw = None | |
| self._ts_off = 0 | |
| # running-thread / ISR incremental state | |
| self._cur_tid = None | |
| self._seg_start = None | |
| self._isr_depth = 0 | |
| self._isr_start = None | |
| # per-thread state-machine incremental state | |
| self._st_cur = {} # tid -> [state, since, reason] | |
| self._st_hint = {} # tid -> (state, reason) | |
| self._running = None | |
| self._provisional = [] # tids whose open state is appended for render | |
| # --- incremental state-machine helpers -------------------------------- | |
| def _st_close(self, tid, ts): | |
| st = self._st_cur.get(tid) | |
| if st is not None and ts > st[1]: | |
| self.tr.states.setdefault(tid, []).append((st[1], ts, st[0], st[2])) | |
| self.tr.state_starts.setdefault(tid, []).append(st[1]) | |
| def _st_set(self, tid, ts, state, reason=""): | |
| self._st_close(tid, ts) | |
| self._st_cur[tid] = [state, ts, reason] | |
| def _drop_provisional(self): | |
| for tid in self._provisional: | |
| segs = self.tr.states.get(tid) | |
| if segs: | |
| segs.pop() | |
| self.tr.state_starts[tid].pop() | |
| self._provisional = [] | |
| def _add_provisional(self): | |
| last = self.tr.t1 | |
| for tid, st in self._st_cur.items(): | |
| state, since, reason = st | |
| if last > since and state != ST_DEAD: | |
| self.tr.states.setdefault(tid, []).append((since, last, state, reason)) | |
| self.tr.state_starts.setdefault(tid, []).append(since) | |
| self._provisional.append(tid) | |
| def _state_machine(self, ts, nm, fields, tid): | |
| if nm == "thread_switched_in": | |
| self._running = tid | |
| if tid is not None: | |
| self._st_set(tid, ts, ST_RUN) | |
| self._st_hint.pop(tid, None) | |
| elif nm == "thread_switched_out": | |
| t = tid if tid is not None else self._running | |
| if t is not None and self._st_cur.get(t, [None])[0] == ST_RUN: | |
| h = self._st_hint.pop(t, None) | |
| if h: | |
| self._st_set(t, ts, h[0], h[1]) | |
| else: | |
| self._st_set(t, ts, ST_RDY) # preempted, still runnable | |
| self._running = None | |
| elif nm in _SLEEP_ENTERS: | |
| if self._running is not None: | |
| to = fields.get("timeout", fields.get("ms", fields.get("us", ""))) | |
| self._st_hint[self._running] = (ST_SLP, f"sleep {to}") | |
| elif nm.endswith("_blocking"): | |
| if self._running is not None: | |
| self._st_hint[self._running] = (ST_BLK, _block_reason_nm(nm, fields)) | |
| elif nm in _PEND_EVENTS: | |
| if tid is not None: | |
| h = self._st_hint.get(tid) | |
| self._st_set(tid, ts, ST_BLK, h[1] if h else "blocked") | |
| elif nm in _READY_EVENTS: | |
| if tid is not None: | |
| self._st_set(tid, ts, ST_RDY) | |
| self._st_hint.pop(tid, None) | |
| elif nm in _SUSPEND_EVENTS: | |
| if tid is not None: | |
| self._st_set(tid, ts, ST_SUS) | |
| elif nm in _ABORT_EVENTS: | |
| if tid is not None: | |
| self._st_set(tid, ts, ST_DEAD) | |
| elif nm == "thread_create": | |
| if tid is not None and tid not in self._st_cur: | |
| self._st_set(tid, ts, ST_RDY) | |
| def _consume(self, ts, eid, name, fields): | |
| tr = self.tr | |
| tr.events.append(Event(ts, eid, name, fields)) | |
| if len(tr.events) == 1: | |
| tr.t0 = ts | |
| tr.t1 = ts | |
| tid = fields.get("thread_id") | |
| if tid is not None: | |
| t = tr.thread(tid) | |
| nm = fields.get("name") | |
| if nm: | |
| t["name"] = nm | |
| if eid in (THREAD_PRIO_SET, THREAD_SCHED_PRIO_SET) and "prio" in fields: | |
| t["prio"] = fields["prio"] | |
| if eid == THREAD_INFO: | |
| t["stack_base"] = fields.get("stack_base") | |
| t["stack_size"] = fields.get("stack_size") | |
| # Running-thread timeline from context switches. | |
| if eid == THREAD_SWITCHED_IN: | |
| if self._cur_tid is not None and self._seg_start is not None: | |
| tr.segments.append((self._seg_start, ts, self._cur_tid)) | |
| self._cur_tid = tid | |
| self._seg_start = ts | |
| elif eid == THREAD_SWITCHED_OUT: | |
| if self._cur_tid is not None and self._seg_start is not None: | |
| tr.segments.append((self._seg_start, ts, self._cur_tid)) | |
| self._cur_tid = None | |
| self._seg_start = None | |
| # ISR overlay spans (nested enters collapse to one active span). | |
| if eid == ISR_ENTER: | |
| if self._isr_depth == 0: | |
| self._isr_start = ts | |
| self._isr_depth += 1 | |
| elif eid in (ISR_EXIT, ISR_EXIT_TO_SCHEDULER): | |
| if self._isr_depth > 0: | |
| self._isr_depth -= 1 | |
| if self._isr_depth == 0 and self._isr_start is not None: | |
| tr.isr_spans.append((self._isr_start, ts)) | |
| self._isr_start = None | |
| self._state_machine(ts, name, fields, tid) | |
| def _decode_buf(self): | |
| data = self._buf | |
| off = 0 | |
| n = len(data) | |
| hsz = self.hdr.size | |
| new = 0 | |
| while off + hsz <= n: | |
| if self.has_ts: | |
| ts, eid = self.hdr.unpack_from(data, off) | |
| else: | |
| (eid,) = self.hdr.unpack_from(data, off) | |
| ts = None | |
| edef = self.defs.get(eid) | |
| if edef is None: | |
| # Unknown id: the record length is unknown, so we cannot safely | |
| # skip it. Stop and keep the bytes; flag the desync for the UI. | |
| self._desync = True | |
| break | |
| rec = hsz + edef.size | |
| if off + rec > n: | |
| break # incomplete trailing record; wait for more | |
| if ts is None: | |
| ts = self._fake_ts | |
| self._fake_ts += 1 | |
| else: | |
| if self._prev_raw is not None and ts < self._prev_raw: | |
| self._ts_off += self._prev_raw # counter wrapped | |
| self._prev_raw = ts | |
| ts += self._ts_off | |
| fields, _ = edef.decode(data, off + hsz) | |
| off += rec | |
| self._consume(ts, eid, edef.name, fields) | |
| new += 1 | |
| self._buf = data[off:] | |
| return new | |
| def feed(self, data): | |
| """Decode as many complete records as `data` (appended) allows. | |
| Returns the number of new events decoded. | |
| """ | |
| if not data: | |
| return 0 | |
| self._buf += data | |
| self._drop_provisional() | |
| new = self._decode_buf() | |
| self._add_provisional() | |
| return new | |
| def parse_trace(path, defs, has_ts=True): | |
| """Read a complete trace file in one shot (non-live path).""" | |
| reader = TraceReader(defs, has_ts) | |
| with safe_open(path, "rb") as fh: | |
| reader.feed(fh.read()) | |
| return reader.tr | |
| # Thread state codes. The glyph ramp goes from solid (on CPU) to light (idle), | |
| # so darker == closer to running; colour disambiguates further. | |
| ST_RUN = "run" # executing on a CPU | |
| ST_RDY = "rdy" # runnable, waiting for a CPU (preempted / just readied) | |
| ST_BLK = "blk" # pending on a kernel object (see reason) | |
| ST_SLP = "slp" # sleeping on a timeout (k_sleep and friends) | |
| ST_SUS = "sus" # explicitly suspended | |
| ST_DEAD = "dead" # aborted / not yet created | |
| STATE_GLYPH = {ST_RUN: "█", ST_RDY: "▓", ST_BLK: "▒", ST_SLP: "░", ST_SUS: "—", ST_DEAD: " "} | |
| STATE_NAME = { | |
| ST_RUN: "run", | |
| ST_RDY: "ready", | |
| ST_BLK: "blocked", | |
| ST_SLP: "sleep", | |
| ST_SUS: "susp", | |
| ST_DEAD: "dead", | |
| } | |
| # Precedence for choosing one state when a column's time bucket spans several; | |
| # running wins ties so context switches stay visible. | |
| STATE_PREC = {ST_RUN: 5, ST_BLK: 4, ST_RDY: 3, ST_SLP: 2, ST_SUS: 1, ST_DEAD: 0} | |
| # curses colour pair per state (set up in run_curses). | |
| STATE_PAIR = {ST_RUN: 1, ST_RDY: 2, ST_BLK: 3, ST_SLP: 4, ST_SUS: 5} | |
| _SLEEP_ENTERS = { | |
| "k_sleep_enter", | |
| "thread_sleep_enter", | |
| "thread_msleep_enter", | |
| "thread_usleep_enter", | |
| } | |
| _READY_EVENTS = { | |
| "thread_sched_ready", | |
| "thread_ready", | |
| "thread_wakeup", | |
| "thread_sched_wakeup", | |
| "thread_resume", | |
| "thread_sched_resume", | |
| } | |
| _SUSPEND_EVENTS = {"thread_suspend", "thread_sched_suspend"} | |
| _ABORT_EVENTS = {"thread_abort", "thread_sched_abort"} | |
| _PEND_EVENTS = {"thread_sched_pend", "thread_pending"} | |
| def _block_reason_nm(nm, f): | |
| """Best-effort 'why is this thread blocked' string for a blocking event.""" | |
| if nm.startswith("semaphore"): | |
| return f"sem 0x{f.get('id', 0):x}" | |
| if nm.startswith("mutex"): | |
| return f"mutex 0x{f.get('id', 0):x}" | |
| if nm.startswith("msgq"): | |
| return f"msgq 0x{f.get('id', 0):x}" | |
| if nm.startswith("condvar"): | |
| return f"condvar 0x{f.get('id', 0):x}" | |
| if nm.startswith("event_wait"): | |
| return f"event 0x{f.get('event_id', 0):x}" | |
| if nm.startswith("thread_join"): | |
| return f"join 0x{f.get('thread_id', 0):x}" | |
| if nm.startswith("mem_slab"): | |
| return f"memslab 0x{f.get('id', 0):x}" | |
| if nm.startswith("work"): | |
| return "work" | |
| return "blocked" | |
| def state_at(tr, tid, ts): | |
| """Return (state, reason) of thread tid at time ts, or (None, '').""" | |
| import bisect | |
| starts = tr.state_starts.get(tid) | |
| if not starts: | |
| return None, "" | |
| i = bisect.bisect_right(starts, ts) - 1 | |
| if i < 0: | |
| return None, "" | |
| s, e, state, reason = tr.states[tid][i] | |
| if s <= ts < e or (i == len(starts) - 1 and ts >= s): | |
| return state, reason | |
| return None, "" | |
| def thread_running_at(tr, ts): | |
| """Return the thread id running at time ts (single-CPU), or None.""" | |
| for tid in tr.threads: | |
| st, _ = state_at(tr, tid, ts) | |
| if st == ST_RUN: | |
| return tid | |
| return None | |
| # -------------------------------------------------------------------------- | |
| # Rendering helpers shared by the text and curses front-ends. | |
| # -------------------------------------------------------------------------- | |
| def lane_order(tr): | |
| """Return thread ids ordered for display (most-active first).""" | |
| busy = {} | |
| for s, e, tid in tr.segments: | |
| busy[tid] = busy.get(tid, 0) + max(0, e - s) | |
| # Threads that ran, busiest first; then any known-but-idle threads. | |
| ran = sorted(busy, key=lambda t: -busy[t]) | |
| others = [t for t in tr.threads if t not in busy] | |
| return ran + others | |
| def thread_label(tr, tid): | |
| name = tr.threads.get(tid, {}).get("name") or "" | |
| if not name: | |
| # Idle threads are often unnamed; flag the well-known idle id heuristic | |
| # by name only, otherwise show the handle. | |
| name = "(unnamed)" | |
| return name | |
| def fmt_time(ns): | |
| """Human friendly timestamp.""" | |
| if ns >= 1_000_000: | |
| return f"{ns / 1_000_000:.3f}ms" | |
| if ns >= 1_000: | |
| return f"{ns / 1_000:.3f}us" | |
| return f"{ns}ns" | |
| def render_rows(tr, order, view0, view1, width): | |
| """Return {tid: list[str-cells]} bars plus an ISR row, for the time window. | |
| Each cell is one of ' ' (idle), '#'/full block (running). We use shading to | |
| indicate partial coverage of a column's time bucket. | |
| """ | |
| span = max(1, view1 - view0) | |
| col_ns = span / width | |
| shades = " ░▒▓█" | |
| rows = {} | |
| for tid in order: | |
| rows[tid] = [0.0] * width | |
| isr_row = [0.0] * width | |
| def paint(acc, s, e): | |
| if e <= view0 or s >= view1: | |
| return | |
| s = max(s, view0) | |
| e = min(e, view1) | |
| c0 = (s - view0) / col_ns | |
| c1 = (e - view0) / col_ns | |
| ic0 = int(c0) | |
| ic1 = min(width - 1, int(c1)) | |
| for c in range(ic0, ic1 + 1): | |
| cell0 = c | |
| cell1 = c + 1 | |
| cov = min(cell1, c1) - max(cell0, c0) | |
| if cov > 0: | |
| acc[c] = min(1.0, acc[c] + cov) | |
| for s, e, tid in tr.segments: | |
| if tid in rows: | |
| paint(rows[tid], s, e) | |
| for s, e in tr.isr_spans: | |
| paint(isr_row, s, e) | |
| def to_chars(acc): | |
| out = [] | |
| for v in acc: | |
| if v <= 0.0: | |
| out.append(" ") | |
| else: | |
| idx = min(4, 1 + int(v * 3.999)) | |
| out.append(shades[idx]) | |
| return out | |
| char_rows = {tid: to_chars(acc) for tid, acc in rows.items()} | |
| return char_rows, to_chars(isr_row) | |
| def render_state_rows(tr, lanes, view0, view1, width): | |
| """Return {tid: [state-or-None per column]} for the visible window. | |
| For each column the dominant state (by time covered, ties broken by | |
| STATE_PREC) is chosen, so a lane reads as run/ready/blocked/sleep bars. | |
| """ | |
| import bisect | |
| span = max(1, view1 - view0) | |
| col_ns = span / width | |
| out = {} | |
| for tid in lanes: | |
| cells = [None] * width | |
| segs = tr.states.get(tid) | |
| if segs: | |
| starts = tr.state_starts[tid] | |
| acc = {} # col -> {state: coverage} | |
| i = max(0, bisect.bisect_right(starts, view0) - 1) | |
| n = len(segs) | |
| while i < n: | |
| s, e, state, _reason = segs[i] | |
| i += 1 | |
| if s >= view1: | |
| break | |
| if e <= view0 or state == ST_DEAD: | |
| continue | |
| cs = max(s, view0) | |
| ce = min(e, view1) | |
| c0 = (cs - view0) / col_ns | |
| c1 = (ce - view0) / col_ns | |
| for c in range(int(c0), min(width - 1, int(c1)) + 1): | |
| cov = min(c + 1, c1) - max(c, c0) | |
| if cov <= 0: | |
| continue | |
| d = acc.get(c) | |
| if d is None: | |
| d = {} | |
| acc[c] = d | |
| d[state] = d.get(state, 0) + cov | |
| for c, d in acc.items(): | |
| best = None | |
| bestv = -1.0 | |
| for st, v in d.items(): | |
| if v > bestv or (v == bestv and STATE_PREC[st] > STATE_PREC.get(best, -1)): | |
| best, bestv = st, v | |
| cells[c] = best | |
| out[tid] = cells | |
| return out | |
| def window_stats(tr, view0, view1): | |
| """Per-thread time in each state over [view0, view1]. | |
| Returns ({tid: {state: ns}}, span_ns). Threads that exist in the window | |
| have segments covering it, so their per-state times sum to the span. | |
| """ | |
| import bisect | |
| span = max(1, view1 - view0) | |
| per = {} | |
| for tid, segs in tr.states.items(): | |
| starts = tr.state_starts[tid] | |
| i = max(0, bisect.bisect_right(starts, view0) - 1) | |
| n = len(segs) | |
| acc = {} | |
| while i < n: | |
| s, e, st, _r = segs[i] | |
| i += 1 | |
| if s >= view1: | |
| break | |
| if e <= view0: | |
| continue | |
| d = min(e, view1) - max(s, view0) | |
| if d > 0: | |
| acc[st] = acc.get(st, 0) + d | |
| if acc: | |
| per[tid] = acc | |
| return per, span | |
| # -------------------------------------------------------------------------- | |
| # Non-interactive ASCII output. | |
| # -------------------------------------------------------------------------- | |
| def run_text(tr, width): | |
| order = lane_order(tr) | |
| label_w = max([len("THREAD")] + [len(thread_label(tr, t)) for t in order]) + 1 | |
| label_w = min(label_w, 18) | |
| tl_w = max(20, width - label_w - 12) | |
| print( | |
| f"Zephyr CTF trace ({len(tr.events)} events, " | |
| f"{len(tr.threads)} threads, " | |
| f"{fmt_time(tr.t1 - tr.t0)} span)" | |
| ) | |
| print(f"t0 = {fmt_time(tr.t0)} t1 = {fmt_time(tr.t1)}") | |
| print() | |
| state_rows = render_state_rows(tr, order, tr.t0, tr.t1, tl_w) | |
| _, isr_row = render_rows(tr, [], tr.t0, tr.t1, tl_w) | |
| def lbl(s): | |
| s = s[: label_w - 1] | |
| return s.ljust(label_w) | |
| legend = " ".join( | |
| f"{STATE_GLYPH[s]}={STATE_NAME[s]}" for s in (ST_RUN, ST_RDY, ST_BLK, ST_SLP, ST_SUS) | |
| ) | |
| print("states: " + legend) | |
| print() | |
| axis = "".join("|" if i % 10 == 0 else "." for i in range(tl_w)) | |
| print( | |
| " " * label_w | |
| + "0" | |
| + " " * (tl_w - len(fmt_time(tr.t1 - tr.t0)) - 1) | |
| + fmt_time(tr.t1 - tr.t0) | |
| ) | |
| print(lbl("THREAD") + axis) | |
| for tid in order: | |
| cells = state_rows[tid] | |
| if all(c is None for c in cells): | |
| continue | |
| line = "".join(STATE_GLYPH.get(c, " ") if c else " " for c in cells) | |
| print(lbl(thread_label(tr, tid)) + line) | |
| if any(c != " " for c in isr_row): | |
| print(lbl("[ISR]") + "".join(isr_row)) | |
| print() | |
| print("Threads") | |
| print(f" {'handle':<12}{'name':<18}{'prio':>5} {'stack_base':<12}{'stack_sz':>9}") | |
| for tid in order: | |
| t = tr.threads[tid] | |
| prio = "" if t["prio"] is None else str(t["prio"]) | |
| sb = "" if t["stack_base"] is None else f"0x{t['stack_base']:08x}" | |
| ss = "" if t["stack_size"] is None else str(t["stack_size"]) | |
| print(f" 0x{tid:08x} {thread_label(tr, tid):<18}{prio:>5} {sb:<12}{ss:>9}") | |
| # Event histogram by type. | |
| print() | |
| print("Event counts") | |
| counts = {} | |
| for ev in tr.events: | |
| counts[ev.name] = counts.get(ev.name, 0) + 1 | |
| for name, c in sorted(counts.items(), key=lambda kv: -kv[1]): | |
| print(f" {c:>7} {name}") | |
| # -------------------------------------------------------------------------- | |
| # Interactive curses front-end. | |
| # -------------------------------------------------------------------------- | |
| def _fmt_fields(ev): | |
| """One-line rendering of an event's decoded fields.""" | |
| parts = [] | |
| for k, v in ev.fields.items(): | |
| if isinstance(v, int) and ( | |
| k.endswith("id") or k.endswith("base") or k in ("port", "pin", "iface", "pkt") | |
| ): | |
| parts.append(f"{k}=0x{v:x}") | |
| else: | |
| parts.append(f"{k}={v}") | |
| return " ".join(parts) | |
| def run_curses(stdscr, reader, fh=None): | |
| import bisect | |
| import curses | |
| import time | |
| tr = reader.tr | |
| live = fh is not None | |
| curses.curs_set(0) | |
| stdscr.keypad(True) | |
| use_color = curses.has_colors() | |
| if use_color: | |
| curses.start_color() | |
| curses.use_default_colors() | |
| # Lanes are coloured by thread *state* (not identity) so run / ready / | |
| # blocked / sleep / suspended read at a glance. | |
| curses.init_pair(1, curses.COLOR_GREEN, -1) # running | |
| curses.init_pair(2, curses.COLOR_YELLOW, -1) # ready | |
| curses.init_pair(3, curses.COLOR_RED, -1) # blocked | |
| curses.init_pair(4, curses.COLOR_CYAN, -1) # sleeping | |
| curses.init_pair(5, curses.COLOR_MAGENTA, -1) # suspended | |
| curses.init_pair(20, curses.COLOR_BLACK, curses.COLOR_RED) # ISR | |
| curses.init_pair(21, curses.COLOR_BLACK, curses.COLOR_CYAN) # header/cursor | |
| curses.init_pair(22, curses.COLOR_WHITE, curses.COLOR_BLUE) # selected | |
| curses.init_pair(23, curses.COLOR_BLACK, curses.COLOR_GREEN) # play | |
| order = lane_order(tr) | |
| ts_list = [ev.ts for ev in tr.events] | |
| full_span = max(1, tr.t1 - tr.t0) | |
| view0 = tr.t0 | |
| view1 = tr.t1 if tr.t1 > tr.t0 else tr.t0 + 1 | |
| cursor_ns = float(tr.t0) | |
| sel_row = 0 # selected thread lane | |
| show_events = True # event-density row | |
| show_info = True # compact info panel under the chart | |
| show_metrics = True # metrics panel in the gap below the lanes | |
| mode = "gantt" # "gantt" or "log" | |
| log_scroll = 0 # top event index shown in the log view | |
| # Autoplay: cursor advances by `speed` trace-ns per real second. Default to | |
| # replaying the whole trace in ~20 seconds. | |
| playing = False | |
| speed = max(1.0, full_span / 20.0) | |
| last_tick = time.monotonic() | |
| # Live follow: when reading a file that is still being written, keep the | |
| # view pinned to the latest events. Panning/zooming detaches; 'f' re-pins. | |
| live_follow = live | |
| nthreads = len(tr.threads) | |
| def clampview(): | |
| nonlocal view0, view1 | |
| span = view1 - view0 | |
| if span < 100: | |
| span = 100 | |
| view1 = view0 + span | |
| if view0 < tr.t0: | |
| view0 = tr.t0 | |
| view1 = view0 + span | |
| if view1 > tr.t1: | |
| view1 = tr.t1 | |
| view0 = max(tr.t0, view1 - span) | |
| def running_at(ts): | |
| return thread_running_at(tr, ts) | |
| def nearest_idx(ts): | |
| if not ts_list: | |
| return 0 | |
| i = bisect.bisect_left(ts_list, ts) | |
| if i <= 0: | |
| return 0 | |
| if i >= len(ts_list): | |
| return len(ts_list) - 1 | |
| return i if (ts_list[i] - ts) < (ts - ts_list[i - 1]) else i - 1 | |
| def fmt_speed(sp): | |
| return f"{fmt_time(int(sp))}/s" | |
| while True: | |
| # --- Poll the file for newly written events ----------------------- | |
| if live: | |
| try: | |
| chunk = fh.read() | |
| except OSError: | |
| chunk = b"" | |
| if reader.feed(chunk): | |
| ts_list.extend(ev.ts for ev in tr.events[len(ts_list) :]) | |
| full_span = max(1, tr.t1 - tr.t0) | |
| if len(tr.threads) != nthreads: | |
| nthreads = len(tr.threads) | |
| order = lane_order(tr) | |
| if live_follow and tr.t1 > tr.t0: | |
| span = max(100, view1 - view0) | |
| view1 = tr.t1 | |
| view0 = max(tr.t0, view1 - span) | |
| cursor_ns = float(tr.t1) | |
| log_scroll = len(tr.events) # keep log view at the newest row | |
| # --- Advance autoplay clock --------------------------------------- | |
| now = time.monotonic() | |
| dt = now - last_tick | |
| last_tick = now | |
| if playing: | |
| cursor_ns += speed * dt | |
| if cursor_ns >= tr.t1: | |
| cursor_ns = float(tr.t1) | |
| playing = False | |
| # Scroll the window so the playhead stays visible (~70% across). | |
| span = view1 - view0 | |
| if cursor_ns > view1 or cursor_ns < view0: | |
| view0 = cursor_ns - span * 0.7 | |
| view1 = view0 + span | |
| clampview() | |
| stdscr.erase() | |
| h, w = stdscr.getmaxyx() | |
| if h < 10 or w < 50: | |
| stdscr.addstr(0, 0, "Terminal too small (need >=50x10).") | |
| stdscr.refresh() | |
| stdscr.timeout(-1) | |
| if stdscr.getch() in (ord("q"), 27): | |
| return | |
| continue | |
| clampview() | |
| span = view1 - view0 | |
| cur_i = nearest_idx(int(cursor_ns)) | |
| # --- Header (both modes) ------------------------------------------ | |
| if live: | |
| play_tag = " LIVE>> " if live_follow else " LIVE " | |
| elif playing: | |
| play_tag = f" PLAY {fmt_speed(speed)} " | |
| else: | |
| play_tag = " PAUSED " | |
| hdr = ( | |
| f" CTF {('LOG' if mode == 'log' else 'GANTT')} " | |
| f"ev={len(tr.events)} thr={len(tr.threads)} " | |
| f"t={fmt_time(int(cursor_ns) - tr.t0)}" | |
| ) | |
| hattr = curses.color_pair(21) if use_color else curses.A_REVERSE | |
| stdscr.addstr(0, 0, hdr[: w - 1].ljust(w - 1 - len(play_tag)), hattr) | |
| hot = playing or (live and live_follow) | |
| pattr = ( | |
| curses.color_pair(23) | |
| if (hot and use_color) | |
| else (curses.A_REVERSE if not use_color else curses.color_pair(22)) | |
| ) | |
| stdscr.addstr(0, max(0, w - 1 - len(play_tag)), play_tag[: w - 1], pattr) | |
| if mode == "log": | |
| _draw_log(stdscr, tr, h, w, cur_i, log_scroll, use_color, curses) | |
| else: | |
| ev_lo = bisect.bisect_left(ts_list, view0) | |
| ev_hi = bisect.bisect_right(ts_list, view1) | |
| sel_row = _draw_gantt( | |
| stdscr, | |
| tr, | |
| h, | |
| w, | |
| view0, | |
| view1, | |
| span, | |
| cursor_ns, | |
| order, | |
| sel_row, | |
| show_events, | |
| show_info, | |
| show_metrics, | |
| use_color, | |
| curses, | |
| running_at, | |
| ev_lo, | |
| ev_hi, | |
| ) | |
| # Footer help. | |
| latest = "[End/f]latest " if live else "[End]latest " | |
| if mode == "log": | |
| help_line = f"{latest}[space]play [up/dn]scroll [PgUp/Dn] [g]antt-view [q]uit" | |
| else: | |
| help_line = ( | |
| f"{latest}[space]play [<-/->]cursor [+/-]zoom " | |
| "[up/dn]lane [a]ll [v]log [m]etrics [i]nfo [q]uit" | |
| ) | |
| stdscr.addstr(h - 1, 0, help_line[: w - 1], hattr) | |
| stdscr.refresh() | |
| # --- Input -------------------------------------------------------- | |
| # Block when idle; poll periodically when live or auto-playing. | |
| stdscr.timeout(33 if playing else (150 if live else -1)) | |
| k = stdscr.getch() | |
| if k == -1: | |
| continue | |
| # End / f: jump to the latest state and re-sync with the newest events | |
| # (and resume live-follow). Works in both the Gantt and log views. | |
| if k in (ord("f"), curses.KEY_END): | |
| jump_span = max(100, view1 - view0) | |
| view1 = tr.t1 | |
| view0 = max(tr.t0, view1 - jump_span) | |
| cursor_ns = float(tr.t1) | |
| log_scroll = len(tr.events) | |
| if live: | |
| live_follow = True | |
| continue | |
| # Navigating away from the live edge detaches follow. | |
| if live: | |
| nav = ( | |
| { | |
| curses.KEY_UP, | |
| curses.KEY_DOWN, | |
| curses.KEY_PPAGE, | |
| curses.KEY_NPAGE, | |
| curses.KEY_HOME, | |
| } | |
| if mode == "log" | |
| else { | |
| curses.KEY_LEFT, | |
| curses.KEY_RIGHT, | |
| ord(","), | |
| ord("."), | |
| ord("+"), | |
| ord("="), | |
| ord("-"), | |
| ord("_"), | |
| ord("a"), | |
| curses.KEY_HOME, | |
| curses.KEY_NPAGE, | |
| curses.KEY_PPAGE, | |
| } | |
| ) | |
| if k in nav: | |
| live_follow = False | |
| step = max(1, span // 40) | |
| pan = max(1, span // 4) | |
| # Keys common to both views. | |
| if k in (ord("q"), 27): | |
| return | |
| elif k == ord(" "): | |
| playing = not playing | |
| last_tick = time.monotonic() | |
| elif k in (ord("]"), ord(">")): | |
| speed = min(full_span * 4.0, speed * 1.5) | |
| elif k in (ord("["), ord("<")): | |
| speed = max(1.0, speed / 1.5) | |
| elif k in (ord("v"), ord("g"), ord("\t")): | |
| mode = "log" if mode == "gantt" else "gantt" | |
| log_scroll = max(0, cur_i - (h // 2)) | |
| continue | |
| if mode == "log": | |
| page = max(1, h - 4) | |
| if k == curses.KEY_UP: | |
| log_scroll = max(0, log_scroll - 1) | |
| elif k == curses.KEY_DOWN: | |
| log_scroll = min(max(0, len(tr.events) - 1), log_scroll + 1) | |
| elif k == curses.KEY_PPAGE: | |
| log_scroll = max(0, log_scroll - page) | |
| elif k == curses.KEY_NPAGE: | |
| log_scroll = min(max(0, len(tr.events) - 1), log_scroll + page) | |
| elif k == curses.KEY_HOME: | |
| log_scroll = 0 | |
| continue | |
| # Gantt-only keys. | |
| if k == curses.KEY_LEFT: | |
| cursor_ns = max(view0, cursor_ns - step) | |
| elif k == curses.KEY_RIGHT: | |
| cursor_ns = min(view1, cursor_ns + step) | |
| elif k == ord(","): | |
| view0 -= pan | |
| view1 -= pan | |
| clampview() | |
| cursor_ns = max(view0, min(view1, cursor_ns)) | |
| elif k == ord("."): | |
| view0 += pan | |
| view1 += pan | |
| clampview() | |
| cursor_ns = max(view0, min(view1, cursor_ns)) | |
| elif k in (ord("+"), ord("=")): | |
| new_span = max(100, span // 2) | |
| view0 = cursor_ns - new_span // 2 | |
| view1 = view0 + new_span | |
| clampview() | |
| elif k in (ord("-"), ord("_")): | |
| new_span = min(full_span, span * 2) | |
| view0 = cursor_ns - new_span // 2 | |
| view1 = view0 + new_span | |
| clampview() | |
| elif k == ord("a"): | |
| view0, view1 = tr.t0, tr.t1 | |
| elif k == curses.KEY_HOME: | |
| cursor_ns = float(view0) | |
| elif k == curses.KEY_UP: | |
| sel_row = max(0, sel_row - 1) | |
| elif k == curses.KEY_DOWN: | |
| sel_row += 1 | |
| elif k == ord("e"): | |
| show_events = not show_events | |
| elif k == ord("i"): | |
| show_info = not show_info | |
| elif k == ord("m"): | |
| show_metrics = not show_metrics | |
| elif k == curses.KEY_NPAGE: | |
| view0 += pan * 2 | |
| view1 += pan * 2 | |
| clampview() | |
| elif k == curses.KEY_PPAGE: | |
| view0 -= pan * 2 | |
| view1 -= pan * 2 | |
| clampview() | |
| def _draw_gantt( | |
| stdscr, | |
| tr, | |
| h, | |
| w, | |
| view0, | |
| view1, | |
| span, | |
| cursor_ns, | |
| order, | |
| sel_row, | |
| show_events, | |
| show_info, | |
| show_metrics, | |
| use_color, | |
| curses, | |
| running_at, | |
| ev_lo, | |
| ev_hi, | |
| ): | |
| """Render the Gantt timeline; returns the clamped selected lane index.""" | |
| label_w = 14 | |
| tl_w = w - label_w - 1 | |
| def st_attr(state): | |
| if use_color and state in STATE_PAIR: | |
| a = curses.color_pair(STATE_PAIR[state]) | |
| return a | curses.A_BOLD if state == ST_RUN else a | |
| return curses.A_BOLD if state == ST_RUN else curses.A_NORMAL | |
| # Time-axis ruler with both end labels. | |
| stdscr.addstr(1, 0, "t".ljust(label_w)) | |
| ruler = "".join("|" if i % 10 == 0 else "-" for i in range(tl_w)) | |
| stdscr.addstr(1, label_w, ruler[:tl_w]) | |
| left_lbl = fmt_time(view0 - tr.t0) | |
| right_lbl = fmt_time(view1 - tr.t0) | |
| stdscr.addstr(2, label_w, left_lbl[:tl_w]) | |
| if len(right_lbl) < tl_w: | |
| stdscr.addstr(2, label_w + tl_w - len(right_lbl), right_lbl) | |
| # The info panel at the bottom is intentionally small so the chart gets | |
| # the bulk of the screen; toggle it off entirely with 'i'. | |
| panel_h = 5 if show_info else 0 | |
| lanes_top = 3 | |
| overhead = lanes_top + panel_h + 1 # +1 footer | |
| visible = [ | |
| t for t in order if any(tt == t and s <= view1 and e >= view0 for s, e, tt in tr.segments) | |
| ] or order | |
| max_lanes = max(1, h - overhead - 1) # -1 for the events row | |
| view_lanes = visible[:max_lanes] | |
| sel_row = max(0, min(sel_row, len(view_lanes) - 1)) if view_lanes else 0 | |
| state_rows = render_state_rows(tr, view_lanes, view0, view1, tl_w) | |
| _, isr_row = render_rows(tr, [], view0, view1, tl_w) | |
| cursor_col = int((cursor_ns - view0) / max(1, span) * tl_w) | |
| cursor_col = max(0, min(tl_w - 1, cursor_col)) | |
| row_y = lanes_top | |
| for idx, tid in enumerate(view_lanes): | |
| sel = idx == sel_row | |
| lbl = thread_label(tr, tid)[: label_w - 1].ljust(label_w) | |
| lblattr = ( | |
| curses.color_pair(22) | |
| if (sel and use_color) | |
| else (curses.A_BOLD if sel else curses.A_NORMAL) | |
| ) | |
| stdscr.addstr(row_y, 0, lbl, lblattr) | |
| # Draw each lane as runs of same-state glyphs, coloured per state. | |
| cells = state_rows[tid] | |
| c = 0 | |
| while c < tl_w: | |
| st = cells[c] | |
| run = c + 1 | |
| while run < tl_w and cells[run] == st: | |
| run += 1 | |
| glyph = STATE_GLYPH.get(st, " ") if st else " " | |
| if glyph != " ": | |
| stdscr.addstr(row_y, label_w + c, glyph * (run - c), st_attr(st)) | |
| c = run | |
| # Playhead marker. | |
| st = cells[cursor_col] if cursor_col < len(cells) else None | |
| mark = STATE_GLYPH.get(st, "|") if st else "|" | |
| stdscr.addstr( | |
| row_y, | |
| label_w + cursor_col, | |
| mark, | |
| curses.color_pair(21) if use_color else curses.A_REVERSE, | |
| ) | |
| row_y += 1 | |
| if any(c != " " for c in isr_row) and row_y < h - panel_h - 1: | |
| stdscr.addstr(row_y, 0, "[ISR]".ljust(label_w), curses.A_BOLD) | |
| iattr = curses.color_pair(20) if use_color else curses.A_REVERSE | |
| stdscr.addstr(row_y, label_w, "".join(isr_row)[:tl_w], iattr) | |
| row_y += 1 | |
| if show_events and row_y < h - panel_h - 1: | |
| marks = [" "] * tl_w | |
| bucket = {} | |
| for ev in tr.events[ev_lo:ev_hi]: | |
| if ev.eid in ( | |
| THREAD_SWITCHED_IN, | |
| THREAD_SWITCHED_OUT, | |
| ISR_ENTER, | |
| ISR_EXIT, | |
| ISR_EXIT_TO_SCHEDULER, | |
| IDLE, | |
| ): | |
| continue | |
| c = int((ev.ts - view0) / max(1, span) * tl_w) | |
| c = max(0, min(tl_w - 1, c)) | |
| bucket[c] = bucket.get(c, 0) + 1 | |
| dens = " .:+*#@" | |
| for c, n in bucket.items(): | |
| marks[c] = dens[min(len(dens) - 1, n)] | |
| marks[cursor_col] = "|" if marks[cursor_col] == " " else marks[cursor_col] | |
| stdscr.addstr(row_y, 0, "events".ljust(label_w), curses.A_DIM) | |
| stdscr.addstr(row_y, label_w, "".join(marks)[:tl_w]) | |
| row_y += 1 | |
| # --- Metrics panel (fills the gap above the info strip) --------------- | |
| if show_metrics: | |
| gap_bot = (h - panel_h) if show_info else (h - 1) | |
| if gap_bot - row_y >= 3: | |
| _draw_metrics( | |
| stdscr, | |
| tr, | |
| row_y, | |
| gap_bot, | |
| w, | |
| view0, | |
| view1, | |
| view_lanes, | |
| ev_lo, | |
| ev_hi, | |
| curses, | |
| st_attr, | |
| ) | |
| # --- Compact info panel ----------------------------------------------- | |
| if show_info: | |
| py = h - panel_h | |
| stdscr.hline(py, 0, curses.ACS_HLINE, w) | |
| py += 1 | |
| # Colour-coded state legend. | |
| stdscr.addstr(py, 0, " states: ") | |
| x = 9 | |
| for st in (ST_RUN, ST_RDY, ST_BLK, ST_SLP, ST_SUS): | |
| seg = f"{STATE_GLYPH[st]} {STATE_NAME[st]} " | |
| stdscr.addstr(py, x, seg, st_attr(st)) | |
| x += len(seg) | |
| py += 1 | |
| # Running thread at the playhead. | |
| cur_tid = running_at(int(cursor_ns)) | |
| runlbl = thread_label(tr, cur_tid) if cur_tid is not None else "(none)" | |
| runhandle = f"0x{cur_tid:08x}" if cur_tid is not None else "" | |
| stdscr.addstr( | |
| py, | |
| 0, | |
| (f" running: {runlbl} {runhandle}".ljust(w - 1))[: w - 1], | |
| curses.color_pair(21) if use_color else curses.A_REVERSE, | |
| ) | |
| py += 1 | |
| # Selected lane: its state and, when blocked, why. | |
| sel_tid = view_lanes[sel_row] if view_lanes else None | |
| if sel_tid is not None: | |
| t = tr.threads[sel_tid] | |
| prio = "-" if t["prio"] is None else str(t["prio"]) | |
| ss = "-" if t["stack_size"] is None else f"{t['stack_size']}B" | |
| st, reason = state_at(tr, sel_tid, int(cursor_ns)) | |
| stext = STATE_NAME.get(st, "-") | |
| if st == ST_BLK and reason: | |
| stext = f"blocked on {reason}" | |
| elif st == ST_SLP and reason: | |
| stext = reason | |
| det = ( | |
| f" lane: {thread_label(tr, sel_tid)} 0x{sel_tid:08x} " | |
| f"prio={prio} stack={ss} -> {stext}" | |
| ) | |
| stdscr.addstr(py, 0, det[: w - 1].ljust(w - 1), st_attr(st) if st else curses.A_BOLD) | |
| return sel_row | |
| def _draw_metrics(stdscr, tr, y0, y1, w, view0, view1, lanes, ev_lo, ev_hi, curses, st_attr): | |
| """Draw key metrics over the visible window in the empty area below the | |
| lanes: a CPU-busy gauge, context-switch / event rates, and per-thread | |
| stacked utilization bars (run/ready/blocked/sleep).""" | |
| per, span = window_stats(tr, view0, view1) | |
| idle_ids = {t for t in tr.threads if tr.threads[t].get("name") == "idle"} | |
| run_total = sum(a.get(ST_RUN, 0) for a in per.values()) | |
| idle_run = sum(per.get(t, {}).get(ST_RUN, 0) for t in idle_ids) | |
| busy = max(0.0, min(1.0, (run_total - idle_run) / span)) | |
| switches = sum(1 for ev in tr.events[ev_lo:ev_hi] if ev.eid == THREAD_SWITCHED_IN) | |
| nev = ev_hi - ev_lo | |
| secs = span / 1e9 | |
| def bar(y, x, width, parts, empty="░"): | |
| cx, used = x, 0 | |
| for st, fr in parts: | |
| cnt = min(int(round(fr * width)), width - used) | |
| if cnt <= 0: | |
| continue | |
| stdscr.addstr(y, cx, STATE_GLYPH.get(st, "█") * cnt, st_attr(st)) | |
| cx += cnt | |
| used += cnt | |
| if used < width: | |
| stdscr.addstr(y, cx, empty * (width - used), curses.A_DIM) | |
| y = y0 | |
| stdscr.hline(y, 0, curses.ACS_HLINE, w) | |
| y += 1 | |
| if y >= y1: | |
| return | |
| gw = 14 | |
| head = f" CPU {busy * 100:3.0f}% " | |
| stdscr.addstr(y, 0, head, curses.A_BOLD) | |
| bar(y, len(head), gw, [(ST_RUN, busy)]) | |
| sw_rate = f"{switches / secs:.0f}/s" if secs > 0 else "-" | |
| ev_rate = f"{nev / secs:.0f}/s" if secs > 0 else "-" | |
| tail = f" ctxsw {switches} ({sw_rate}) events {nev} ({ev_rate}) window {fmt_time(span)}" | |
| stdscr.addstr(y, len(head) + gw, tail[: max(0, w - len(head) - gw - 1)]) | |
| y += 1 | |
| # Per-thread utilization: stacked state composition over the window. | |
| bw = max(8, w - 26) | |
| for tid in lanes: | |
| if y >= y1: | |
| break | |
| acc = per.get(tid) | |
| if not acc: | |
| continue | |
| util = acc.get(ST_RUN, 0) / span | |
| name = thread_label(tr, tid)[:11] | |
| stdscr.addstr(y, 0, f" {name:<11}{util * 100:3.0f}% ", curses.A_NORMAL) | |
| parts = [(st, acc.get(st, 0) / span) for st in (ST_RUN, ST_RDY, ST_BLK, ST_SLP, ST_SUS)] | |
| bar(y, 17, bw, parts, empty=" ") | |
| y += 1 | |
| def _draw_log(stdscr, tr, h, w, cur_i, log_scroll, use_color, curses): | |
| """Render the raw CTF event log as a scrollable table.""" | |
| stdscr.addstr( | |
| 1, | |
| 0, | |
| (f"{'#':>7} {'time':>12} {'id':>5} {'event':<28} fields")[: w - 1], | |
| curses.A_UNDERLINE, | |
| ) | |
| rows = h - 3 # header(0), column titles(1), footer(h-1) | |
| # Clamp so a large scroll (e.g. "go to latest") lands on the last full page. | |
| top = max(0, min(log_scroll, max(0, len(tr.events) - rows))) | |
| for r in range(rows): | |
| i = top + r | |
| if i >= len(tr.events): | |
| break | |
| ev = tr.events[i] | |
| line = ( | |
| f"{i:>7} {fmt_time(ev.ts - tr.t0):>12} " | |
| f"0x{ev.eid:03x} {ev.name:<28} {_fmt_fields(ev)}" | |
| ) | |
| attr = curses.A_NORMAL | |
| if i == cur_i: | |
| attr = curses.color_pair(21) if use_color else curses.A_REVERSE | |
| stdscr.addstr(2 + r, 0, line[: w - 1], attr) | |
| def main(): | |
| ap = argparse.ArgumentParser(description="Zephyr CTF trace viewer", allow_abbrev=False) | |
| ap.add_argument("binary", help="CTF trace binary (e.g. build/tracing.bin)") | |
| ap.add_argument("--metadata", help="path to TSDL metadata file") | |
| ap.add_argument( | |
| "--text", action="store_true", help="print a static ASCII timeline instead of the TUI" | |
| ) | |
| ap.add_argument( | |
| "--width", type=int, default=0, help="output width for --text (default: terminal width)" | |
| ) | |
| ap.add_argument( | |
| "--no-timestamp", | |
| action="store_true", | |
| help="trace was built without CONFIG_TRACING_CTF_TIMESTAMP", | |
| ) | |
| ap.add_argument( | |
| "-f", | |
| "--follow", | |
| action="store_true", | |
| help="follow the file live as the application writes it " | |
| "(waits for the file/first events to appear)", | |
| ) | |
| args = ap.parse_args() | |
| meta_path = find_metadata(args.binary, args.metadata) | |
| defs, used = build_event_defs(meta_path) | |
| if used is None: | |
| sys.stderr.write( | |
| "warning: TSDL metadata not found; using built-in scheduling-event " | |
| "table only (pass --metadata to decode all events)\n" | |
| ) | |
| has_ts = not args.no_timestamp | |
| if args.follow: | |
| if args.text or not sys.stdout.isatty(): | |
| sys.stderr.write("--follow requires an interactive terminal\n") | |
| return 1 | |
| return _run_follow(args.binary, defs, has_ts) | |
| tr = parse_trace(args.binary, defs, has_ts=has_ts) | |
| if not tr.events: | |
| sys.stderr.write("no events decoded; is this a CTF trace? try --no-timestamp\n") | |
| return 1 | |
| if args.text or not sys.stdout.isatty(): | |
| width = args.width or (os.get_terminal_size().columns if sys.stdout.isatty() else 100) | |
| run_text(tr, width) | |
| return 0 | |
| reader = TraceReader(defs, has_ts) | |
| with safe_open(args.binary, "rb") as fh: | |
| reader.feed(fh.read()) | |
| import curses | |
| curses.wrapper(run_curses, reader) | |
| return 0 | |
| def _run_follow(path, defs, has_ts): | |
| """Open the trace, catch up on existing bytes, then follow it live.""" | |
| import curses | |
| import time | |
| deadline = None | |
| while not os.path.exists(path): | |
| if deadline is None: | |
| sys.stderr.write(f"waiting for {path} to appear (start the application)...\n") | |
| deadline = True | |
| time.sleep(0.3) | |
| reader = TraceReader(defs, has_ts) | |
| fh = safe_open(path, "rb") | |
| try: | |
| reader.feed(fh.read()) # catch up on whatever already exists | |
| curses.wrapper(run_curses, reader, fh) | |
| finally: | |
| fh.close() | |
| return 0 | |
| if __name__ == "__main__": | |
| sys.exit(main()) |