diff options
Diffstat (limited to 'debug/risc-v-gsp-rm-vm/harness/state_decode.py')
| -rw-r--r-- | debug/risc-v-gsp-rm-vm/harness/state_decode.py | 1082 |
1 files changed, 1082 insertions, 0 deletions
diff --git a/debug/risc-v-gsp-rm-vm/harness/state_decode.py b/debug/risc-v-gsp-rm-vm/harness/state_decode.py new file mode 100644 index 0000000..351515a --- /dev/null +++ b/debug/risc-v-gsp-rm-vm/harness/state_decode.py @@ -0,0 +1,1082 @@ +#!/usr/bin/env python3 +"""Decode engine-visible channel state words into named fields, and diff captures. + +During GR bring-up the answer often hides in a handful of 32-bit words inside +the INSTANCE block, the RAMFC chunk and context-buffer headers: one bit turns +"nothing happens" into an engine that runs or refuses. Raw hex hides that; +this tool turns a captured dump into a named field table and makes two +captures differ field-by-field. + +Schemas: + +* ``greeneye-state-decode/1`` -- the decoded report produced by ``decode``. +* ``greeneye-state-diff/1`` -- the word/field diff produced by ``diff``. +* ``greeneye-server-snapshot/1`` -- an input snapshot emitted elsewhere:: + + {"schema": "greeneye-server-snapshot/1", "source": "...", + "chunks": [{"label": "INSTANCE", "base": "0x0", + "words": ["0x...", "..."]}]} + + (one hex string or integer per 32-bit word; accepted exactly). +* ``greeneye-state-map/1`` -- an optional field map (``--map``), see below. + +Input formats (auto-detected per ``--input``): + +(a) text lines ``LABEL[OFFSET]=VALUE`` -- label case-insensitive, offset and + value hex (``0x..``, letters, or leading-zero padded) or decimal; every + other line is ignored; +(b) a dump row with a bracketed chunk offset and words, e.g. + ``... INSTANCE[0040]: 00100000 00030000 ...`` -- this is the shape live + ladder traces print (8 words per row; any count accepted); +(c) plain hexdump rows ``00000000: 000002d2 00000000 ...`` named with + ``--label``/``--base``; +(d) raw little-endian u32 words (``--bin``, or auto-detected when a text parse + finds nothing and the bytes are control-heavy and 4-byte aligned); +(e) the snapshot JSON above. + +Offsets inside a chunk are byte offsets relative to the chunk start (all +inputs above are u32 word dumps, so offsets are multiples of 4). ``base`` is +the engine-visible address of chunk-relative offset 0: it is recorded as +provenance and for the diff, but it does not shift map lookup. + +Bundled map (deliberately small; extend or override it from a map file, never +by guessing here). Provenance is live bring-up observation (2026-10 GSP +ladder runs): + +* ``INSTANCE`` 0x010: magic ``0xface``. +* ``INSTANCE`` 0x048: GPFIFO VA, observed as ``0x100000``. +* ``RAMFC`` 0x048: ``RAMFC_TARGET`` -- bits 16/17 ``ENG_CTX_VALID`` / + ``CE_CTX_VALID``; the observed live success word ``0x00030000`` sets both. + The bit assignment is observation-driven and fully configurable via the map. + +Map file:: + + {"format": "greeneye-state-map/1", "mode": "extend" | "replace", + "entries": [ + {"chunk": "RAMFC", "label": "RAMFC_TARGET", "offset": "0x48", + "width": 4, + "bits": [{"name": "ENG_CTX_VALID", "lo": 16, "hi": 16, + "values": {"0": "clear", "1": "set"}}], + "note": "..."} + ]} + +A user entry with the same ``chunk`` and ``offset`` replaces the bundled +entry; new chunk/offset pairs extend the table. ``mode: replace`` drops the +bundled table first. An entry without ``bits`` decodes as a named value. + +Honesty: this tool names observed words; it does not prove their meaning. An +unmapped word is silence, not absence. Field offsets without evidence stay +unnamed until a map supplies them. Offline only: no hardware, no module, no +guest, no network. +""" +from __future__ import annotations + +import argparse +import json +import os +import re +import shutil +import sys +import tempfile +from typing import Any + +DECODE_FORMAT = "greeneye-state-decode/1" +DIFF_FORMAT = "greeneye-state-diff/1" +SNAPSHOT_FORMAT = "greeneye-server-snapshot/1" +MAP_FORMAT = "greeneye-state-map/1" +DEMO_FORMAT = "greeneye-state-decode-demo/1" + +WORD_BYTES = 4 +WORD_MASK = 0xFFFFFFFF +WORD_WIDTHS = (4, 8) + + +class StateDecodeError(ValueError): + """An input, map or request the tool cannot honour.""" + + +# --------------------------------------------------------------------------- +# Bundled field table +# --------------------------------------------------------------------------- +# Every entry is an observation from live GSP bring-up traces (2026-10). The +# table is intentionally tiny: unnamed offsets stay unnamed until evidence +# earns them a name, and the map file carries everything else. +BUILTIN_MAP: dict[str, Any] = { + "format": MAP_FORMAT, + "mode": "extend", + "entries": [ + { + "chunk": "INSTANCE", + "label": "MAGIC_0FACE", + "offset": "0x010", + "width": 4, + "note": "Instance magic observed as 0xface at chunk byte 0x010.", + }, + { + "chunk": "INSTANCE", + "label": "GPFIFO_VA", + "offset": "0x048", + "width": 4, + "note": "GPFIFO VA observed as 0x100000 at chunk byte 0x048.", + }, + { + "chunk": "RAMFC", + "label": "RAMFC_TARGET", + "offset": "0x048", + "width": 4, + "bits": [ + {"name": "ENG_CTX_VALID", "lo": 16, "hi": 16}, + {"name": "CE_CTX_VALID", "lo": 17, "hi": 17}, + ], + "note": ("Engine/CE context-valid bits; the observed live success " + "word 0x00030000 sets both. Bits are configurable via " + "--map."), + }, + ], +} + +# --------------------------------------------------------------------------- +# Number and token parsing +# --------------------------------------------------------------------------- + +_EQ_RE = re.compile( + r"^\s*(?P<label>[A-Za-z_][A-Za-z0-9_.-]*)\s*" + r"\[\s*(?P<off>[0-9a-fA-FxX]+)\s*\]\s*=\s*(?P<val>[0-9a-fA-FxX]+)\s*$") + +_ROW_RE = re.compile( + r"(?P<label>[A-Za-z_][A-Za-z0-9_.-]*)\s*" + r"\[\s*(?P<off>[0-9a-fA-FxX]+)\s*\]\s*:\s*" + r"(?P<words>[0-9a-fA-FxX]+(?:\s+[0-9a-fA-FxX]+)*)\s*$") + +_HEXDUMP_RE = re.compile( + r"^\s*(?P<off>[0-9a-fA-FxX]{2,16})\s*:\s*" + r"(?P<words>[0-9a-fA-FxX]+(?:\s+[0-9a-fA-FxX]+)*)\s*$") + +_HEX_RE = re.compile(r"[0-9a-fA-F]+") + +_TEXT_BYTES = frozenset(b"\t\n\r") | frozenset(range(0x20, 0x7F)) + + +def parse_number(value: Any, *, why: str = "number") -> int: + """Parse an int, a ``0x`` string, or a bare token. + + Bare tokens with a hex letter (``face``) or a leading zero and more + digits (``0040``) are read as hex, matching dump padding; other bare + digit strings are decimal. Write ``0x..`` when in doubt. + """ + if isinstance(value, bool): + raise StateDecodeError(f"bad {why} {value!r}") + if isinstance(value, int): + return value + if not isinstance(value, str): + raise StateDecodeError(f"bad {why} {value!r}") + text = value.strip() + if not text: + raise StateDecodeError(f"bad {why} {value!r}") + sign = 1 + if text[0] in "+-": + if text[0] == "-": + sign = -1 + text = text[1:] + low = text.lower() + if low.startswith("0x"): + base, digits = 16, text[2:] + elif re.fullmatch(r"[0-9]+", text): + base = 16 if text.startswith("0") and len(text) > 1 else 10 + digits = text + elif _HEX_RE.fullmatch(text): + base, digits = 16, text + else: + raise StateDecodeError(f"bad {why} {value!r}") + try: + return sign * int(digits, base) + except ValueError as exc: + raise StateDecodeError(f"bad {why} {value!r}") from exc + + +def _parse_word(value: Any, *, where: str) -> int: + number = parse_number(value, why=where) + if not 0 <= number <= WORD_MASK: + raise StateDecodeError(f"{where} {value!r} does not fit in 32 bits") + return number + + +def _hex(value: int, width: int = 4) -> str: + return f"{value:#0{2 + width * 2}x}" + + +def _parse_base(value: Any) -> int: + if value is None: + return 0 + return parse_number(value, why="base") + + +# --------------------------------------------------------------------------- +# Map loading (bundled <- override/extension) +# --------------------------------------------------------------------------- + +def _normalize_entry(raw: Any, *, where: str) -> dict[str, Any]: + if not isinstance(raw, dict): + raise StateDecodeError(f"{where} must be an object") + chunk = raw.get("chunk") + label = raw.get("label") + if not isinstance(chunk, str) or not chunk: + raise StateDecodeError(f"{where} needs a non-empty chunk") + if not isinstance(label, str) or not label: + raise StateDecodeError(f"{where} needs a non-empty label") + if raw.get("offset") is None: + raise StateDecodeError(f"{where} needs an offset") + offset = parse_number(raw["offset"], why=f"{where} offset") + width = raw.get("width", 4) + width = parse_number(width, why=f"{where} width") + if width not in WORD_WIDTHS: + raise StateDecodeError(f"{where} width {width!r} is not 4 or 8") + if offset < 0 or offset % WORD_BYTES: + raise StateDecodeError(f"{where} offset {offset:#x} is not word-aligned") + bits_raw = raw.get("bits") or [] + if not isinstance(bits_raw, list): + raise StateDecodeError(f"{where} bits must be a list") + nbits = width * 8 + bits = [] + for index, bit in enumerate(bits_raw): + if not isinstance(bit, dict): + raise StateDecodeError(f"{where} bits[{index}] must be an object") + name = bit.get("name") + if not isinstance(name, str) or not name: + raise StateDecodeError(f"{where} bits[{index}] needs a name") + lo = parse_number(bit.get("lo"), why=f"{where} bits[{index}].lo") + hi = parse_number(bit.get("hi"), why=f"{where} bits[{index}].hi") + if not 0 <= lo <= hi < nbits: + raise StateDecodeError( + f"{where} bits[{index}] range {lo}..{hi} outside width {width}") + values = bit.get("values") + norm_values: dict[int, str] = {} + if values is not None: + if not isinstance(values, dict): + raise StateDecodeError( + f"{where} bits[{index}].values must be an object") + for key, text in values.items(): + norm_values[parse_number(key, why=f"{where} values key")] = str(text) + bits.append({"name": name, "lo": lo, "hi": hi, "values": norm_values}) + note = raw.get("note") + return { + "chunk": chunk, + "label": label, + "offset": offset, + "width": width, + "bits": bits, + "note": note if isinstance(note, str) else None, + } + + +def load_map(path: str | None = None) -> tuple[dict[tuple[str, int], dict], str]: + """Build the field table; returns ``(table, source)``. + + ``table`` is keyed by ``(chunk.lower(), offset)``. Without ``path`` the + bundled table is used. With ``path``, entries merge over the bundled + table (same chunk+offset replaces, new pairs extend) unless the file sets + ``"mode": "replace"``. + """ + entries: list[Any] = list(BUILTIN_MAP["entries"]) + source = "builtin" + if path: + if not os.path.isfile(path): + raise StateDecodeError(f"map not found: {path}") + with open(path, "r", encoding="utf-8") as handle: + doc = json.load(handle) + source = os.path.abspath(path) + if isinstance(doc, list): + raw_entries, mode = doc, "extend" + elif isinstance(doc, dict): + fmt = doc.get("format", MAP_FORMAT) + if fmt != MAP_FORMAT: + raise StateDecodeError( + f"unsupported map format {fmt!r} (want {MAP_FORMAT})") + raw_entries = doc.get("entries") + mode = doc.get("mode", "extend") + else: + raise StateDecodeError(f"map {path} must be an object or a list") + if mode not in ("extend", "replace"): + raise StateDecodeError(f"map mode {mode!r} is not extend/replace") + if not isinstance(raw_entries, list): + raise StateDecodeError(f"map {path} needs an entries list") + if mode == "replace": + entries = [] + entries.extend(raw_entries) + table: dict[tuple[str, int], dict] = {} + for index, raw in enumerate(entries): + entry = _normalize_entry(raw, where=f"map entry {index}") + table[(entry["chunk"].lower(), entry["offset"])] = entry + return table, source + + +# --------------------------------------------------------------------------- +# Input loading (auto-detected per input) +# --------------------------------------------------------------------------- + +def _read_bytes(path: str) -> bytes: + if not os.path.isfile(path): + raise StateDecodeError(f"input not found: {path}") + with open(path, "rb") as handle: + return handle.read() + + +def _try_json(data: bytes) -> Any: + """Return parsed JSON when the bytes look like JSON, else ``None``.""" + try: + text = data.decode("utf-8") + except UnicodeDecodeError: + return None + stripped = text.lstrip() + if not stripped.startswith("{"): + return None + try: + return json.loads(text) + except json.JSONDecodeError as exc: + raise StateDecodeError(f"invalid JSON input: {exc}") from exc + + +def _try_text(data: bytes) -> str | None: + try: + return data.decode("utf-8") + except UnicodeDecodeError: + return None + + +def _looks_binary(data: bytes) -> bool: + if not data or len(data) % WORD_BYTES: + return False + return any(byte not in _TEXT_BYTES for byte in data) + + +def _chunk_from_binary(data: bytes, label: str, base: int, source: str) -> dict: + if not data: + raise StateDecodeError(f"binary input {source} is empty") + if len(data) % WORD_BYTES: + raise StateDecodeError( + f"binary input {source} size {len(data)} is not a multiple of 4") + words = {offset: int.from_bytes(data[offset:offset + WORD_BYTES], "little") + for offset in range(0, len(data), WORD_BYTES)} + return {"label": label, "base": base, "words": words, + "format": "binary", "source": source} + + +def _document(path: str, fmt: str, label: str | None, base: int, + chunks: list[dict]) -> dict: + return { + "path": os.path.abspath(path), + "format": fmt, + "label": label, + "base": base, + "chunks": chunks, + } + + +def parse_text(text: str, *, default_label: str | None = None, + default_base: int = 0, source: str | None = None) -> list[dict]: + """Parse the text/hexdump/row formats into chunks. + + Chunk names are case-insensitive: ``INSTANCE`` and ``inst`` land in one + chunk, displayed with the first spelling seen. Every line that matches + none of the three patterns is ignored. + """ + chunks: dict[str, dict] = {} + + def chunk_for(label: str) -> dict: + key = label.lower() + chunk = chunks.get(key) + if chunk is None: + chunk = {"label": label, "base": default_base, "words": {}, + "format": "text", "source": source} + chunks[key] = chunk + return chunk + + def put(label: str, offset: int, value: int, where: str) -> None: + if offset < 0 or offset % WORD_BYTES: + raise StateDecodeError(f"{where}: offset {offset:#x} is not word-aligned") + if not 0 <= value <= WORD_MASK: + raise StateDecodeError(f"{where}: value {value:#x} does not fit in 32 bits") + chunk_for(label)["words"][offset] = value + + for lineno, raw_line in enumerate(text.splitlines(), 1): + line = raw_line.strip() + if not line: + continue + where = f"{source or '<text>'}:{lineno}" + match = _EQ_RE.match(line) + if match: + offset = parse_number(match.group("off"), why=f"{where} offset") + value = parse_number(match.group("val"), why=f"{where} value") + put(match.group("label"), offset, value, where) + continue + match = _ROW_RE.search(line) + if match: + offset = parse_number(match.group("off"), why=f"{where} offset") + for index, token in enumerate(match.group("words").split()): + put(match.group("label"), offset + index * WORD_BYTES, + int(token, 16), where) + continue + match = _HEXDUMP_RE.match(line) + if match: + label = default_label or "RAW" + offset = int(match.group("off"), 16) + for index, token in enumerate(match.group("words").split()): + put(label, offset + index * WORD_BYTES, int(token, 16), where) + continue + return [chunks[key] for key in chunks] + + +def load_input(spec: dict) -> dict: + """Load one input spec dict (path/label/base/binary) into a document.""" + path = spec.get("path") + if not path: + raise StateDecodeError("input spec needs a path") + data = _read_bytes(path) + label = spec.get("label") + base = _parse_base(spec.get("base")) + if spec.get("binary"): + chunk = _chunk_from_binary(data, label or "RAW", base, path) + return _document(path, "binary", label, base, [chunk]) + doc = _try_json(data) + if doc is not None: + return _document_from_json(path, doc, label, base) + text = _try_text(data) + if text is not None: + chunks = parse_text(text, default_label=label, default_base=base, + source=path) + if chunks: + return _document(path, "text", label, base, chunks) + if _looks_binary(data): + chunk = _chunk_from_binary(data, label or "RAW", base, path) + return _document(path, "binary", label, base, [chunk]) + raise StateDecodeError( + f"cannot decode {path}: not text fields, not {SNAPSHOT_FORMAT} JSON, " + f"not raw u32 words") + + +def _document_from_json(path: str, doc: Any, label: str | None, + base: int) -> dict: + if isinstance(doc, dict) and doc.get("format") == DECODE_FORMAT: + raise StateDecodeError(f"{path} is a decoded report; use `diff` on it") + if not isinstance(doc, dict) or doc.get("schema") != SNAPSHOT_FORMAT: + got = doc.get("schema") if isinstance(doc, dict) else type(doc).__name__ + raise StateDecodeError( + f"unsupported JSON input {path}: schema={got!r} " + f"(want {SNAPSHOT_FORMAT})") + raw_chunks = doc.get("chunks") + if not isinstance(raw_chunks, list) or not raw_chunks: + raise StateDecodeError(f"snapshot {path} has no chunks") + chunks = [] + for index, raw in enumerate(raw_chunks): + if not isinstance(raw, dict): + raise StateDecodeError(f"{path}:chunks[{index}] must be an object") + chunk_label = raw.get("label") or label or "RAW" + chunk_base = (_parse_base(raw["base"]) if raw.get("base") is not None + else base) + words_raw = raw.get("words") + if not isinstance(words_raw, list): + raise StateDecodeError( + f"{path}:chunks[{index}] needs a words list") + words = {} + for position, word in enumerate(words_raw): + words[position * WORD_BYTES] = _parse_word( + word, where=f"{path}:chunks[{index}].words[{position}]") + chunks.append({"label": str(chunk_label), "base": chunk_base, + "words": words, "format": "snapshot-json", + "source": path}) + return _document(path, "snapshot-json", label, base, chunks) + + +# --------------------------------------------------------------------------- +# Decode +# --------------------------------------------------------------------------- + +def _read_field_value(words: dict[int, int], offset: int, + width: int) -> int | None: + """Little-endian read of a 4/8-byte field; ``None`` when words are absent.""" + data = b"" + for step in range(0, width, WORD_BYTES): + word = words.get(offset + step) + if word is None: + return None + data += word.to_bytes(WORD_BYTES, "little") + return int.from_bytes(data, "little") + + +def _bit_state(bit: dict, value: int) -> str: + values = bit.get("values") or {} + if value in values: + return values[value] + mask = (1 << (bit["hi"] - bit["lo"] + 1)) - 1 + if value == 0: + return "clear" + if value == mask: + return "set" + return f"{value:#x}" + + +def _decode_field(entry: dict, value: int) -> dict: + width = entry["width"] + field: dict[str, Any] = { + "name": entry["label"], + "offset": f"{entry['offset']:#x}", + "width": width, + "kind": "bits" if entry["bits"] else "value", + "raw": _hex(value, width), + "value": value, + } + if entry.get("note"): + field["note"] = entry["note"] + if entry["bits"]: + bitfields = [] + for bit in entry["bits"]: + raw_bit = (value >> bit["lo"]) & ((1 << (bit["hi"] - bit["lo"] + 1)) - 1) + bitfields.append({ + "name": bit["name"], + "lo": bit["lo"], + "hi": bit["hi"], + "value": raw_bit, + "state": _bit_state(bit, raw_bit), + }) + field["bitfields"] = bitfields + return field + + +def decode_chunk(chunk: dict, entries: dict[tuple[str, int], dict]) -> dict: + """Decode one internal chunk into the ``greeneye-state-decode/1`` shape.""" + label = chunk["label"] + words: dict[int, int] = chunk["words"] + matched = sorted( + (entry for (chunk_key, _off), entry in entries.items() + if chunk_key == label.lower()), + key=lambda entry: (entry["offset"], entry["label"])) + fields = [] + covered: set[int] = set() + for entry in matched: + value = _read_field_value(words, entry["offset"], entry["width"]) + if value is None: + continue + fields.append(_decode_field(entry, value)) + covered.update(range(entry["offset"], + entry["offset"] + entry["width"], WORD_BYTES)) + unmapped = [{"offset": f"{offset:#x}", "raw": _hex(words[offset])} + for offset in sorted(words) if offset not in covered] + return { + "label": label, + "base": f"{chunk['base']:#x}", + "source": chunk.get("source"), + "input_format": chunk.get("format"), + "word_width": WORD_BYTES, + "word_count": len(words), + "words": {f"{offset:#x}": _hex(value) + for offset, value in sorted(words.items())}, + "fields": fields, + "unmapped": unmapped, + "unmapped_count": len(unmapped), + } + + +def decode_specs(specs: list[dict], entries: dict[tuple[str, int], dict], + map_source: str = "builtin") -> dict: + """Load and decode input specs into a ``greeneye-state-decode/1`` report.""" + sources = [] + chunks = [] + for spec in specs: + document = load_input(spec) + sources.append({ + "path": document["path"], + "format": document["format"], + "label": document["label"], + "base": f"{document['base']:#x}", + "chunk_labels": [chunk["label"] for chunk in document["chunks"]], + }) + for chunk in document["chunks"]: + chunks.append(decode_chunk(chunk, entries)) + chunks.sort(key=lambda chunk: (chunk["label"].lower(), chunk["label"], + int(chunk["base"], 16))) + return { + "format": DECODE_FORMAT, + "map_source": map_source, + "sources": sources, + "chunks": chunks, + "limits": [ + "Field names come from the bundled/extended map; an unmapped word " + "is silence, not absence.", + "Offsets are chunk-relative bytes; base is recorded provenance and " + "does not shift field lookup.", + "No timing (L2) or device-environment (L3) claim: this is a field " + "decoder over captured words.", + ], + } + + +# --------------------------------------------------------------------------- +# Diff +# --------------------------------------------------------------------------- + +def _validate_report(doc: Any, side: str) -> None: + if not isinstance(doc, dict) or doc.get("format") != DECODE_FORMAT: + raise StateDecodeError(f"{side} is not a {DECODE_FORMAT} report") + if not isinstance(doc.get("chunks"), list): + raise StateDecodeError(f"{side} report has no chunks list") + + +def _words_to_int(chunk: dict) -> dict[int, int]: + return {int(offset, 16): int(raw, 16) + for offset, raw in chunk["words"].items()} + + +def _field_index(chunk: dict | None) -> dict[tuple[int, str], dict]: + if chunk is None: + return {} + return {(int(field["offset"], 16), field["name"]): field + for field in chunk["fields"]} + + +def _field_summary(field: dict | None) -> str | None: + if field is None: + return None + if field.get("kind") == "bits": + parts = [f"{bit['name']}={bit['state']}" + for bit in field.get("bitfields", [])] + return ", ".join(parts) if parts else field["raw"] + return f"value={field['raw']}" + + +def _changed_fields(left: dict | None, right: dict | None) -> list[dict]: + left_fields = _field_index(left) + right_fields = _field_index(right) + rows = [] + for offset, name in sorted(set(left_fields) | set(right_fields)): + left_field = left_fields.get((offset, name)) + right_field = right_fields.get((offset, name)) + if (left_field is not None and right_field is not None + and left_field["raw"] == right_field["raw"]): + continue + row: dict[str, Any] = { + "offset": f"{offset:#x}", + "name": name, + "old": left_field["raw"] if left_field else None, + "new": right_field["raw"] if right_field else None, + "old_summary": _field_summary(left_field), + "new_summary": _field_summary(right_field), + } + if (left_field is not None and right_field is not None + and left_field.get("kind") == right_field.get("kind") == "bits"): + left_bits = {bit["name"]: bit for bit in left_field["bitfields"]} + right_bits = {bit["name"]: bit for bit in right_field["bitfields"]} + bits = [] + for bit_name in sorted(set(left_bits) | set(right_bits)): + bits.append({ + "name": bit_name, + "old": (left_bits[bit_name]["state"] + if bit_name in left_bits else None), + "new": (right_bits[bit_name]["state"] + if bit_name in right_bits else None), + }) + row["bits"] = bits + rows.append(row) + return rows + + +def _word_rows(words: dict[int, int]) -> list[dict]: + return [{"offset": f"{offset:#x}", "raw": _hex(raw)} + for offset, raw in sorted(words.items())] + + +def _diff_chunk(left: dict | None, right: dict | None) -> dict: + if left is None or right is None: + present = right if left is None else left + words = _words_to_int(present) + rows = _word_rows(words) + row = { + "label": present["label"], + "base": present["base"], + "left_base": left["base"] if left else None, + "right_base": right["base"] if right else None, + "status": "right_only" if left is None else "left_only", + "added": rows if left is None else [], + "removed": rows if right is None else [], + "changed": [], + "changed_fields": _changed_fields(left, right), + } + return row + left_words = _words_to_int(left) + right_words = _words_to_int(right) + added = _word_rows({offset: right_words[offset] + for offset in right_words if offset not in left_words}) + removed = _word_rows({offset: left_words[offset] + for offset in left_words if offset not in right_words}) + changed = [{"offset": f"{offset:#x}", "old": _hex(left_words[offset]), + "new": _hex(right_words[offset])} + for offset in sorted(set(left_words) & set(right_words)) + if left_words[offset] != right_words[offset]] + changed_fields = _changed_fields(left, right) + status = "changed" if (added or removed or changed or changed_fields) \ + else "identical" + return { + "label": left["label"], + "base": left["base"], + "left_base": left["base"], + "right_base": right["base"], + "status": status, + "added": added, + "removed": removed, + "changed": changed, + "changed_fields": changed_fields, + } + + +def diff_reports(left: dict, right: dict) -> dict: + """Deterministic word/field diff of two decoded reports.""" + _validate_report(left, "left") + _validate_report(right, "right") + left_groups: dict[str, list[dict]] = {} + right_groups: dict[str, list[dict]] = {} + for chunk in left["chunks"]: + left_groups.setdefault(chunk["label"].lower(), []).append(chunk) + for chunk in right["chunks"]: + right_groups.setdefault(chunk["label"].lower(), []).append(chunk) + labels = sorted(set(left_groups) | set(right_groups)) + chunks = [] + for label in labels: + left_list = left_groups.get(label, []) + right_list = right_groups.get(label, []) + for index in range(max(len(left_list), len(right_list))): + left_chunk = left_list[index] if index < len(left_list) else None + right_chunk = right_list[index] if index < len(right_list) else None + chunks.append(_diff_chunk(left_chunk, right_chunk)) + summary = { + "chunks": len(chunks), + "changed_chunks": sum(1 for chunk in chunks + if chunk["status"] != "identical"), + "added": sum(len(chunk["added"]) for chunk in chunks), + "removed": sum(len(chunk["removed"]) for chunk in chunks), + "changed_words": sum(len(chunk["changed"]) for chunk in chunks), + "changed_fields": sum(len(chunk["changed_fields"]) for chunk in chunks), + } + return { + "format": DIFF_FORMAT, + "left": {"format": left["format"], "map_source": left.get("map_source"), + "chunks": len(left["chunks"])}, + "right": {"format": right["format"], "map_source": right.get("map_source"), + "chunks": len(right["chunks"])}, + "chunks": chunks, + "summary": summary, + "identical": all(chunk["status"] == "identical" for chunk in chunks), + "limits": [ + "Word/field diff only; an unchanged word does not prove untested " + "behaviour.", + "Chunks pair by label and occurrence order; a differing base is " + "recorded but does not re-pair.", + ], + } + + +# --------------------------------------------------------------------------- +# Rendering +# --------------------------------------------------------------------------- + +def render_decode(report: dict) -> str: + """Readable field table for humans; the JSON report stays canonical.""" + lines = [] + for chunk in report["chunks"]: + lines.append( + f"chunk {chunk['label']} base={chunk['base']} " + f"words={chunk['word_count']} mapped={len(chunk['fields'])} " + f"unmapped={chunk['unmapped_count']}") + for field in chunk["fields"]: + if field["kind"] == "bits": + detail = " ".join(f"{bit['name']}={bit['state']}" + for bit in field["bitfields"]) + else: + detail = f"value={field['raw']}" + lines.append(f" {field['offset']:>6} {field['raw']:<12} " + f"{field['name']:<20} {detail}") + nonzero = [word for word in chunk["unmapped"] + if int(word["raw"], 16)] + if nonzero: + shown = " ".join(f"{word['offset']}={word['raw']}" + for word in nonzero[:12]) + more = "" if len(nonzero) <= 12 else f" (+{len(nonzero) - 12} more)" + lines.append(f" unmapped nonzero: {shown}{more}") + return "\n".join(lines) + + +def render_diff(report: dict) -> str: + """Readable diff for humans; the JSON report stays canonical.""" + lines = [] + for chunk in report["chunks"]: + lines.append(f"chunk {chunk['label']} base={chunk['base']}: " + f"{chunk['status']}") + for word in chunk["added"]: + lines.append(f" added {word['offset']} {word['raw']}") + for word in chunk["removed"]: + lines.append(f" removed {word['offset']} {word['raw']}") + for word in chunk["changed"]: + lines.append(f" changed {word['offset']} " + f"{word['old']} -> {word['new']}") + for field in chunk["changed_fields"]: + lines.append(f" field {field['name']} @ {field['offset']}: " + f"{field['old']} -> {field['new']} " + f"({field['old_summary']} -> {field['new_summary']})") + for bit in field.get("bits", []): + lines.append(f" bit {bit['name']}: " + f"{bit['old']} -> {bit['new']}") + lines.append(f"summary: {json.dumps(report['summary'], sort_keys=True)} " + f"identical={report['identical']}") + return "\n".join(lines) + + +# --------------------------------------------------------------------------- +# Demo +# --------------------------------------------------------------------------- + +def _demo_words(word48: int) -> list[str]: + words = ["0x0"] * 20 # chunk words 0x00..0x4c + words[0x10 // WORD_BYTES] = "0xface" + words[0x48 // WORD_BYTES] = _hex(word48) + return words + + +def _demo_snapshot(name: str, *, gpfifo_va: int, ramfc_target: int) -> dict: + return { + "schema": SNAPSHOT_FORMAT, + "source": name, + "chunks": [ + {"label": "INSTANCE", "base": "0x0", + "words": _demo_words(gpfifo_va)}, + {"label": "RAMFC", "base": "0x0", + "words": _demo_words(ramfc_target)}, + ], + } + + +def run_demo(out: str | None = None) -> dict: + """Craft two snapshots in a tempdir, decode+diff them, and report. + + Expected: INSTANCE word 0x048 moves 0x100000 -> 0x200000 and RAMFC word + 0x048 moves 0x00030000 -> 0x00010000 (CE_CTX_VALID clears), i.e. two + changed words and two changed named fields. + """ + demo_dir = tempfile.mkdtemp(prefix="greeneye-state-decode-demo-") + try: + left_path = os.path.join(demo_dir, "left.json") + right_path = os.path.join(demo_dir, "right.json") + left_snapshot = _demo_snapshot("demo-left", gpfifo_va=0x100000, + ramfc_target=0x00030000) + right_snapshot = _demo_snapshot("demo-right", gpfifo_va=0x200000, + ramfc_target=0x00010000) + _write_json(left_path, left_snapshot) + _write_json(right_path, right_snapshot) + entries, map_source = load_map(None) + left = decode_specs( + [{"path": left_path, "label": None, "base": None, "binary": False}], + entries, map_source) + right = decode_specs( + [{"path": right_path, "label": None, "base": None, "binary": False}], + entries, map_source) + diff = diff_reports(left, right) + ok = (not diff["identical"] + and diff["summary"]["changed_words"] == 2 + and diff["summary"]["changed_fields"] == 2 + and diff["summary"]["added"] == 0 + and diff["summary"]["removed"] == 0) + demo = { + "format": DEMO_FORMAT, + "ok": ok, + "left": left, + "right": right, + "diff": diff, + } + if out: + os.makedirs(out, exist_ok=True) + artifacts = {} + for name, doc in (("state-decode-left.json", left), + ("state-decode-right.json", right), + ("state-diff.json", diff)): + artifacts[name] = _write_json(os.path.join(out, name), doc) + for name, path in (("state-snapshot-left.json", left_path), + ("state-snapshot-right.json", right_path)): + target = os.path.join(out, name) + shutil.copyfile(path, target) + artifacts[name] = os.path.abspath(target) + demo["artifacts"] = artifacts + return demo + finally: + shutil.rmtree(demo_dir, ignore_errors=True) + + +# --------------------------------------------------------------------------- +# CLI +# --------------------------------------------------------------------------- + +class _InputAction(argparse.Action): + """Collect repeatable ``--input``/``--bin`` specs in command order.""" + + def __init__(self, option_strings, dest, binary=False, **kwargs): + self.binary = binary + super().__init__(option_strings, dest, **kwargs) + + def __call__(self, parser, namespace, values, option_string=None): + specs = getattr(namespace, self.dest, None) + if specs is None: + specs = [] + setattr(namespace, self.dest, specs) + spec = {"path": values, "label": None, "base": None, + "binary": self.binary} + specs.append(spec) + setattr(namespace, "_current_spec", spec) + + +class _InputMetaAction(argparse.Action): + """Attach ``--label``/``--base`` to the preceding input, or set a default.""" + + def __call__(self, parser, namespace, values, option_string=None): + current = getattr(namespace, "_current_spec", None) + if current is not None: + current[self.dest] = values + setattr(namespace, self.dest, values) + + +def _apply_defaults(specs: list[dict], label: str | None, + base: str | None) -> list[dict]: + resolved = [] + for spec in specs: + item = dict(spec) + if item.get("label") is None: + item["label"] = label + if item.get("base") is None: + item["base"] = base + resolved.append(item) + return resolved + + +def _peek_report(path: str) -> dict | None: + data = _read_bytes(path) + doc = _try_json(data) + if isinstance(doc, dict) and doc.get("format") == DECODE_FORMAT: + _validate_report(doc, path) + return doc + return None + + +def _load_side(spec: dict, entries: dict[tuple[str, int], dict], + map_source: str) -> dict: + report = _peek_report(spec["path"]) + if report is not None: + return report + return decode_specs([spec], entries, map_source) + + +def _write_json(path: str, doc: Any) -> str: + absolute = os.path.abspath(path) + parent = os.path.dirname(absolute) + if parent: + os.makedirs(parent, exist_ok=True) + with open(absolute, "w", encoding="utf-8") as handle: + json.dump(doc, handle, indent=2, sort_keys=True) + handle.write("\n") + return absolute + + +def build_parser() -> argparse.ArgumentParser: + parser = argparse.ArgumentParser( + prog="state_decode.py", + description="Decode engine-visible channel state words into named " + "fields, and diff two captures (offline).") + sub = parser.add_subparsers(dest="command") + + decode = sub.add_parser( + "decode", help="decode one or more raw inputs into a field table") + decode.add_argument("--input", action=_InputAction, dest="inputs", + default=None, metavar="FILE", + help="input file; repeatable, auto-detected format") + decode.add_argument("--bin", action=_InputAction, dest="inputs", + default=None, metavar="FILE", + help="raw little-endian u32 input; repeatable") + decode.add_argument("--label", action=_InputMetaAction, default=None, + help="chunk label for the preceding --input/--bin " + "(default for the rest)") + decode.add_argument("--base", action=_InputMetaAction, default="0x0", + help="engine address of chunk offset 0 for the " + "preceding input (default 0x0)") + decode.add_argument("--map", dest="map_path", default=None, + help=f"{MAP_FORMAT} override/extension file") + decode.add_argument("--json", dest="json_path", default=None, + help="write the decoded report JSON here") + + diff = sub.add_parser( + "diff", help="diff two decoded reports or two raw inputs") + diff.add_argument("left", help=f"{DECODE_FORMAT} report or raw input") + diff.add_argument("right", help=f"{DECODE_FORMAT} report or raw input") + diff.add_argument("--left-label", dest="left_label", default=None, + help="chunk label for a raw left input") + diff.add_argument("--left-base", dest="left_base", default="0x0", + help="base for a raw left input") + diff.add_argument("--left-bin", dest="left_bin", action="store_true", + help="force the left input to raw u32 words") + diff.add_argument("--right-label", dest="right_label", default=None, + help="chunk label for a raw right input") + diff.add_argument("--right-base", dest="right_base", default="0x0", + help="base for a raw right input") + diff.add_argument("--right-bin", dest="right_bin", action="store_true", + help="force the right input to raw u32 words") + diff.add_argument("--map", dest="map_path", default=None, + help=f"{MAP_FORMAT} override/extension file") + diff.add_argument("--json", dest="json_path", default=None, + help="write the diff report JSON here") + + demo = sub.add_parser( + "demo", help="craft two snapshots, decode+diff them, print the result") + demo.add_argument("--out", default=None, + help="write demo snapshots and reports to this directory") + return parser + + +def main(argv: list[str] | None = None) -> int: + parser = build_parser() + args = parser.parse_args(argv) + try: + if args.command == "decode": + specs = _apply_defaults(list(args.inputs or []), args.label, + args.base) + if not specs: + raise StateDecodeError("decode needs at least one --input/--bin") + entries, map_source = load_map(args.map_path) + report = decode_specs(specs, entries, map_source) + print(render_decode(report)) + if args.json_path: + _write_json(args.json_path, report) + return 0 + if args.command == "diff": + entries, map_source = load_map(args.map_path) + left = _load_side({"path": args.left, "label": args.left_label, + "base": args.left_base, + "binary": args.left_bin}, entries, map_source) + right = _load_side({"path": args.right, "label": args.right_label, + "base": args.right_base, + "binary": args.right_bin}, entries, map_source) + report = diff_reports(left, right) + print(render_diff(report)) + if args.json_path: + _write_json(args.json_path, report) + return 0 if report["identical"] else 3 + if args.command == "demo": + demo = run_demo(args.out) + print(f"demo: {demo['format']} ok={demo['ok']}") + print(render_decode(demo["left"])) + print(render_diff(demo["diff"])) + print(json.dumps({"ok": demo["ok"], + "summary": demo["diff"]["summary"]}, + indent=2, sort_keys=True)) + return 0 if demo["ok"] else 3 + parser.print_help() + return 2 + except StateDecodeError as exc: + print(f"error: {exc}", file=sys.stderr) + return 2 + + +if __name__ == "__main__": + raise SystemExit(main()) |