diff options
Diffstat (limited to 'debug/risc-v-gsp-rm-vm/harness/server_snapshot.py')
| -rw-r--r-- | debug/risc-v-gsp-rm-vm/harness/server_snapshot.py | 698 |
1 files changed, 698 insertions, 0 deletions
diff --git a/debug/risc-v-gsp-rm-vm/harness/server_snapshot.py b/debug/risc-v-gsp-rm-vm/harness/server_snapshot.py new file mode 100644 index 0000000..1099fb0 --- /dev/null +++ b/debug/risc-v-gsp-rm-vm/harness/server_snapshot.py @@ -0,0 +1,698 @@ +#!/usr/bin/env python3 +"""Collect a canonical server-state snapshot from cycle evidence directories. + +The differential loop (live run vs emulated run, run N vs run N+1) needs one +machine-readable view of the server state a cycle produced. Cycles already +write that state as text: labelled field variables (``LABEL[offset]=0xVALUE``), +labelled hexdump blocks (an ``INSTANCE[0000]: w w w w``-style row, or a header +line naming the label followed by ``OFFSET: w w w w`` rows -- e.g. +``LOGINIT-raw`` / ``LOGRM-raw`` raw task-log dumps), and ``vgsp-nocat`` record +lines carrying ``recType=`` / ``errorCode=``. This harness discovers those +files under evidence/session directories, normalizes every dump to +little-endian u32 words, and emits one JSON snapshot that two runs can be +compared with ``diff``. + +Schemas +------- + +``greeneye-server-snapshot/1``:: + + {"schema": "greeneye-server-snapshot/1", + "source": "<string>", + "created": "<ISO-8601>", + "chunks": [{"label": "INSTANCE", "base": "0x0", + "words": ["0x00000000", ...]}], + "records": [{"kind": "record", "raw": "...", + "rec_type": 5, "error_code": 0x57f1344, + "source_name": "ASSERT"}]} + +``greeneye-server-snapshot-diff/1``:: + + {"schema": "greeneye-server-snapshot-diff/1", + "a": {"source": ..., "created": ...}, "b": {...}, + "chunks": [{"label": "INSTANCE", "status": "changed|added|removed|unchanged", + "base": {"a": "0x0", "b": "0x0"}, + "words": {"a": 1024, "b": 1024}, + "added": [{"index": 8, "value": "0x00000000"}], + "removed": [{"index": 8, "value": "0x00000001"}], + "changed": [{"index": 8, "a": "0x...", "b": "0x..."}]}], + "records": {"added": [...], "removed": [...]}, + "summary": {"identical": false, "chunks_added": 0, ...}} + +Normalization rules (the consumer depends on them): + +* words are 8-hex-digit lowercase strings of little-endian u32 values; +* a ``LABEL[offset]=0xVALUE`` value wider than 32 bits becomes two words + (low word first); its offset orders the entries and selects ``base``, and + the values are appended in offset order (the offset is a table index there, + not a byte address); +* hexdump rows are placed by byte offset (``(offset - base) // 4``) when + 4-byte aligned, so a row missing in one run still aligns the rest; +* ``base`` is the smallest offset seen for the chunk (``0x0`` when none); +* chunks are sorted by label; a repeated label is merged only when its base + and word list are byte-identical, otherwise it is kept as ``LABEL#2``, + ``LABEL#3`` ... and the collision is reported in the run stats; +* non-empty records are deduplicated by raw line and sorted by raw line; +* ``created`` is the only variable field: pin it with ``--created`` or + ``SOURCE_DATE_EPOCH`` for deterministic output. + +Honesty: this is an evidence normalizer, not a state oracle. It preserves +what the cycle wrote -- offsets are honoured, identical duplicates are merged, +differing duplicates are renamed, and dump-looking lines that fail to parse +are counted (``unparsed``) instead of being silently dropped. It never +executes anything, touches no GPU and reads only files given on the command +line. +""" +from __future__ import annotations + +import argparse +import fnmatch +import json +import os +import re +import sys +import tempfile +import time +from typing import Any, Iterable + +TOOL_ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) +if TOOL_ROOT not in sys.path: + sys.path.insert(0, TOOL_ROOT) + +SNAPSHOT_FORMAT = "greeneye-server-snapshot/1" +DIFF_FORMAT = "greeneye-server-snapshot-diff/1" +DEMO_FORMAT = "greeneye-server-snapshot-demo/1" + +#: File-name globs discovered under each ``--dir`` unless extended with +#: ``--include``. fnmatch's ``*`` matches leading dots, so the rolling host +#: dmesg captures (``.dmesg-*.rolling``) are covered by ``*.rolling``. +DEFAULT_INCLUDES = ("*.txt", "*.log", "*.rolling") + +#: ``INSTANCE[0000]: w w w w`` -- one labelled dump row inside an arbitrary +#: prefix (dmesg timestamp, subsystem, case id). Offsets are hexadecimal. +_INLINE_ROW_RE = re.compile( + r"(?P<label>[A-Za-z_][A-Za-z0-9_.]*)\[(?P<offset>(?:0x)?[0-9A-Fa-f]+)\]:" + r"[ \t]*(?P<tail>[0-9A-Fa-f]{1,8}(?:[ \t]+[0-9A-Fa-f]{1,8})*)[ \t]*$" +) + +#: ``LABEL[offset]=0xVALUE`` (optionally ``LABEL[0].suffix=``), several per +#: line allowed (``root[0]=... pd2[0]=... pd0[0].small=...``). +_LABELED_VALUE_RE = re.compile( + r"(?P<label>[A-Za-z_][A-Za-z0-9_.]*)\[(?P<offset>(?:0x)?[0-9A-Fa-f]+)\]" + r"(?P<suffix>\.[A-Za-z_][A-Za-z0-9_]*)?[ \t]*=[ \t]*" + r"(?P<value>0[xX][0-9A-Fa-f]+)\b" +) + +#: A bare hexdump row (``00000010: w w w w``) inside a labelled block. +_BLOCK_ROW_RE = re.compile( + r"^(?P<offset>[0-9A-Fa-f]{4,16}):[ \t]+" + r"(?P<tail>[0-9A-Fa-f]{1,8}(?:[ \t]+[0-9A-Fa-f]{1,8})*)[ \t]*$" +) + +#: A block header naming the label for the rows that follow: ``LOGINIT-raw:``, +#: ``LOGRM-raw:``, ``RAMFC:`` ... Used only as one-line lookahead. +_BLOCK_HEADER_RE = re.compile(r"^(?P<label>[A-Za-z_][A-Za-z0-9_.-]*)[ \t]*:") + +#: A GSP-RM error record line (``rec seq=1 recType=5 errorCode=0x...``). +_RECORD_RE = re.compile( + r"\brecType=(?P<rec_type>\d+)\b.*?\berrorCode=(?P<error_code>0[xX][0-9A-Fa-f]+)" +) + +_SOURCE_RE = re.compile(r'\bsource="(?P<source_name>[^"]*)"') + +#: Loose shape check for a line that looks like a dump but failed to parse; +#: these are counted so silent data loss is visible in the run stats. +_UNPARSED_HINT_RE = re.compile( + r"\[[0-9A-Fa-fx]+\][ \t]*[=:]|^[0-9A-Fa-f]{4,}:|\brecType=" +) + + +class SnapshotError(ValueError): + """An input file or snapshot the tool cannot honour.""" + + +def _new_stats() -> dict[str, Any]: + return { + "files": 0, + "lines": 0, + "chunks": 0, + "words": 0, + "records": 0, + "unparsed": 0, + "merged": 0, + "collisions": 0, + "conflicts": 0, + "misaligned": 0, + "collision_list": [], + } + + +class _Builder: + """Words captured for one label occurrence before normalization. + + Entries are ``(offset, [u32 word, ...])`` in encounter order. Hexdump + builders are positional: entries are placed at ``(offset - base) // 4`` + when the byte offset is 4-byte aligned; otherwise they are appended after + the highest placed word. Value-line builders (``LABEL[offset]=0xVALUE``) + are sequential: the offset only orders the entries and selects ``base`` + (its unit is a table entry index there, not a byte), so each entry's words + are appended in offset order. A later value that would overwrite a + different value at the same slot is dropped and counted as a + ``conflict``. + """ + + __slots__ = ("label", "positional", "entries", "conflicts", "misaligned") + + def __init__(self, label: str, positional: bool = True): + self.label = label + self.positional = positional + self.entries: list[tuple[int, list[str]]] = [] + self.conflicts = 0 + self.misaligned = 0 + + def add(self, offset: int, words: list[str]) -> None: + if not words: + return + for existing_offset, existing_words in self.entries: + if existing_offset == offset and existing_words == words: + return + self.entries.append((offset, words)) + + def build(self) -> tuple[int, list[str]]: + base = min(offset for offset, _ in self.entries) + placed: dict[int, str] = {} + append_at = 0 + conflicts = 0 + misaligned = 0 + for offset, words in sorted(self.entries, key=lambda entry: entry[0]): + if not self.positional: + index = append_at + elif (offset - base) % 4 == 0: + index = (offset - base) // 4 + else: + index = append_at + misaligned += 1 + for step, word in enumerate(words): + slot = index + step + if slot in placed and placed[slot] != word: + conflicts += 1 + elif slot not in placed: + placed[slot] = word + append_at = max(append_at, index + len(words)) + self.conflicts = conflicts + self.misaligned = misaligned + return base, [placed[index] for index in sorted(placed)] + + +def _word(value: int) -> str: + return f"0x{value & 0xFFFFFFFF:08x}" + + +def _value_words(value: int) -> list[str]: + """Split a value into little-endian u32 words (low word first).""" + words = [] + while True: + words.append(_word(value)) + value >>= 32 + if value == 0: + return words + + +def _split_words(tail: str) -> list[str]: + return [f"0x{int(token, 16):08x}" for token in tail.split()] + + +def _row_offset(text: str) -> int: + return int(text, 16) + + +def _labeled_offset(text: str) -> int: + if text.lower().startswith("0x"): + return int(text, 16) + try: + return int(text, 10) + except ValueError: + return int(text, 16) + + +def _match_record(line: str) -> dict[str, Any] | None: + match = _RECORD_RE.search(line) + if not match: + return None + record: dict[str, Any] = { + "kind": "record", + "raw": line, + "rec_type": int(match.group("rec_type")), + "error_code": int(match.group("error_code"), 16), + } + source = _SOURCE_RE.search(line) + if source: + record["source_name"] = source.group("source_name") + return record + + +def _parse_text(text: str, stats: dict[str, Any]) -> tuple[list[_Builder], list[dict]]: + occurrences: list[_Builder] = [] + records: list[dict] = [] + by_value_label: dict[str, _Builder] = {} + current_block: _Builder | None = None + inline_block: _Builder | None = None + pending_header: str | None = None + + for line in text.splitlines(): + stripped = line.strip() + if not stripped: + current_block = pending_header = inline_block = None + continue + + record = _match_record(stripped) + if record is not None: + records.append(record) + current_block = pending_header = inline_block = None + continue + + match = _INLINE_ROW_RE.search(stripped) + if match: + label = match.group("label") + words = _split_words(match.group("tail")) + if inline_block is not None and inline_block.label == label: + inline_block.add(_row_offset(match.group("offset")), words) + else: + inline_block = _Builder(label) + inline_block.add(_row_offset(match.group("offset")), words) + occurrences.append(inline_block) + current_block = pending_header = None + continue + + match = _BLOCK_ROW_RE.match(stripped) + if match: + if current_block is None: + if pending_header is None: + stats["unparsed"] += 1 + continue + current_block = _Builder(pending_header) + occurrences.append(current_block) + current_block.add(int(match.group("offset"), 16), + _split_words(match.group("tail"))) + pending_header = inline_block = None + continue + + value_matches = list(_LABELED_VALUE_RE.finditer(stripped)) + if value_matches: + for value_match in value_matches: + label = value_match.group("label") + if value_match.group("suffix"): + label += value_match.group("suffix") + builder = by_value_label.get(label) + if builder is None: + builder = _Builder(label, positional=False) + by_value_label[label] = builder + occurrences.append(builder) + builder.add(_labeled_offset(value_match.group("offset")), + _value_words(int(value_match.group("value"), 16))) + current_block = pending_header = inline_block = None + continue + + if _UNPARSED_HINT_RE.search(stripped): + stats["unparsed"] += 1 + + current_block = inline_block = None + header = _BLOCK_HEADER_RE.match(stripped) + pending_header = header.group("label") if header else None + + return occurrences, records + + +def _chunk_sort_key(label: str) -> tuple[str, int, str]: + stem, separator, tail = label.rpartition("#") + if separator and tail.isdigit(): + return stem, int(tail), label + return label, 0, label + + +def _finalize(occurrences: Iterable[_Builder], + stats: dict[str, Any]) -> list[dict[str, Any]]: + chunks: dict[str, dict[str, Any]] = {} + for builder in occurrences: + base, words = builder.build() + stats["conflicts"] += builder.conflicts + stats["misaligned"] += builder.misaligned + if not words: + continue + key = builder.label + suffix = 1 + merged = False + while key in chunks: + existing = chunks[key] + if existing["base"] == f"{base:#x}" and existing["words"] == words: + stats["merged"] += 1 + merged = True + break + suffix += 1 + key = f"{builder.label}#{suffix}" + if merged: + continue + chunks[key] = {"label": key, "base": f"{base:#x}", "words": words} + if suffix > 1: + stats["collisions"] += 1 + stats["collision_list"].append( + {"label": builder.label, "renamed": key, + "base": f"{base:#x}", "words": len(words)}) + return [chunks[key] for key in sorted(chunks, key=_chunk_sort_key)] + + +def discover(dirs: Iterable[str] | None, files: Iterable[str] | None, + includes: Iterable[str] | None = None) -> list[str]: + """Return the sorted absolute paths matched by the dirs/files/includes.""" + patterns = tuple(DEFAULT_INCLUDES) + tuple(includes or ()) + found: set[str] = set() + for directory in dirs or (): + if not os.path.isdir(directory): + raise SnapshotError(f"not a directory: {directory}") + for root, _subdirs, names in os.walk(directory): + for name in names: + if any(fnmatch.fnmatch(name, pattern) for pattern in patterns): + found.add(os.path.abspath(os.path.join(root, name))) + for path in files or (): + if not os.path.isfile(path): + raise SnapshotError(f"not a file: {path}") + found.add(os.path.abspath(path)) + return sorted(found) + + +def _default_source(dirs: Iterable[str] | None, + files: Iterable[str] | None) -> str: + roots = sorted(os.path.abspath(path) + for path in list(dirs or ()) + list(files or ())) + return ",".join(roots) if roots else "empty" + + +def _created_iso() -> str: + """Build timestamp; stable when ``SOURCE_DATE_EPOCH`` is set.""" + raw = os.environ.get("SOURCE_DATE_EPOCH") + try: + epoch = int(raw) if raw else int(time.time()) + except ValueError: + epoch = 0 + return time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime(epoch)) + + +def collect(dirs: Iterable[str] | None = None, + files: Iterable[str] | None = None, + includes: Iterable[str] | None = None, + source: str | None = None, + created: str | None = None) -> tuple[dict[str, Any], dict[str, Any]]: + """Collect every dump in ``dirs``/``files`` into one snapshot. + + Returns ``(snapshot, stats)``; ``stats`` carries the parse accounting + (files, lines, chunks, words, records, unparsed candidates, merged + duplicates, renamed collisions, conflicts, misaligned offsets) and is + printed by the CLI but is deliberately not part of the consumed schema. + """ + paths = discover(dirs, files, includes) + if not paths: + raise SnapshotError( + "no input files matched the requested --dir/--file/--include set") + stats = _new_stats() + occurrences: list[_Builder] = [] + records: dict[str, dict] = {} + for path in paths: + try: + with open(path, "r", encoding="utf-8", errors="replace") as handle: + text = handle.read() + except OSError as exc: + raise SnapshotError(f"cannot read {path}: {exc}") from exc + stats["files"] += 1 + stats["lines"] += len(text.splitlines()) + file_occurrences, file_records = _parse_text(text, stats) + occurrences.extend(file_occurrences) + for record in file_records: + records.setdefault(record["raw"], record) + chunks = _finalize(occurrences, stats) + stats["chunks"] = len(chunks) + stats["words"] = sum(len(chunk["words"]) for chunk in chunks) + stats["records"] = len(records) + snapshot = { + "schema": SNAPSHOT_FORMAT, + "source": source if source is not None else _default_source(dirs, files), + "created": created if created is not None else _created_iso(), + "chunks": chunks, + "records": sorted(records.values(), key=lambda record: record["raw"]), + } + return snapshot, stats + + +def diff_snapshots(a: dict[str, Any], b: dict[str, Any]) -> dict[str, Any]: + """Compare two ``greeneye-server-snapshot/1`` snapshots word by word. + + Emits a ``greeneye-server-snapshot-diff/1`` report: per shared chunk the + added/removed/changed word lists (with word indices), whole-chunk + added/removed rows, per-chunk and total counts, and record deltas. Output + is sorted by label and by word index, so two diffs of the same inputs are + byte-identical. + """ + for name, snapshot in (("a", a), ("b", b)): + if not isinstance(snapshot, dict) or snapshot.get("schema") != SNAPSHOT_FORMAT: + raise SnapshotError(f"snapshot {name} is not {SNAPSHOT_FORMAT}") + a_chunks = {chunk["label"]: chunk for chunk in a.get("chunks", [])} + b_chunks = {chunk["label"]: chunk for chunk in b.get("chunks", [])} + rows: list[dict[str, Any]] = [] + counts = { + "chunks_added": 0, "chunks_removed": 0, "chunks_changed": 0, + "chunks_unchanged": 0, "words_added": 0, "words_removed": 0, + "words_changed": 0, + } + for label in sorted(set(a_chunks) | set(b_chunks), key=_chunk_sort_key): + left = a_chunks.get(label) + right = b_chunks.get(label) + if left is None: + added = [{"index": index, "value": word} + for index, word in enumerate(right["words"])] + row = {"label": label, "status": "added", + "base": {"a": None, "b": right["base"]}, + "words": {"a": 0, "b": len(right["words"])}, + "added": added, "removed": [], "changed": []} + counts["chunks_added"] += 1 + counts["words_added"] += len(added) + elif right is None: + removed = [{"index": index, "value": word} + for index, word in enumerate(left["words"])] + row = {"label": label, "status": "removed", + "base": {"a": left["base"], "b": None}, + "words": {"a": len(left["words"]), "b": 0}, + "added": [], "removed": removed, "changed": []} + counts["chunks_removed"] += 1 + counts["words_removed"] += len(removed) + else: + a_words, b_words = left["words"], right["words"] + added, removed, changed = [], [], [] + for index in range(max(len(a_words), len(b_words))): + if index >= len(a_words): + added.append({"index": index, "value": b_words[index]}) + elif index >= len(b_words): + removed.append({"index": index, "value": a_words[index]}) + elif a_words[index] != b_words[index]: + changed.append({"index": index, "a": a_words[index], + "b": b_words[index]}) + unchanged = (not added and not removed and not changed + and left["base"] == right["base"]) + row = {"label": label, + "status": "unchanged" if unchanged else "changed", + "base": {"a": left["base"], "b": right["base"]}, + "words": {"a": len(a_words), "b": len(b_words)}, + "added": added, "removed": removed, "changed": changed} + if unchanged: + counts["chunks_unchanged"] += 1 + else: + counts["chunks_changed"] += 1 + counts["words_added"] += len(added) + counts["words_removed"] += len(removed) + counts["words_changed"] += len(changed) + rows.append(row) + + a_records = {record.get("raw", "") for record in a.get("records", [])} + b_records = {record.get("raw", "") for record in b.get("records", [])} + records_added = sorted(b_records - a_records) + records_removed = sorted(a_records - b_records) + counts["records_added"] = len(records_added) + counts["records_removed"] = len(records_removed) + identical = (counts["chunks_added"] == 0 and counts["chunks_removed"] == 0 + and counts["chunks_changed"] == 0 + and not records_added and not records_removed) + return { + "schema": DIFF_FORMAT, + "a": {"source": a.get("source"), "created": a.get("created")}, + "b": {"source": b.get("source"), "created": b.get("created")}, + "chunks": rows, + "records": {"added": records_added, "removed": records_removed}, + "summary": {"identical": identical, **counts}, + } + + +# --------------------------------------------------------------------------- +# Demo: two tiny snapshots in a tempdir, collected and diffed +# --------------------------------------------------------------------------- + +DEMO_A = """\ +[demo] MOCK[0000]: 00000001 00000002 00000003 00000004 +alpha[0]=0x00000011 +alpha[4]=0x00000022 +""" + +DEMO_B = """\ +[demo] MOCK[0000]: 00000001 000000ff 00000003 00000004 00000005 +alpha[0]=0x00000011 +alpha[4]=0x00000033 +""" + + +def _write_text(path: str, text: str) -> str: + with open(path, "w", encoding="utf-8") as handle: + handle.write(text) + return os.path.abspath(path) + + +def _write_json(path: str, payload: dict[str, Any]) -> str: + with open(path, "w", encoding="utf-8") as handle: + json.dump(payload, handle, indent=2, sort_keys=True) + handle.write("\n") + return os.path.abspath(path) + + +def run_demo(workdir: str) -> dict[str, Any]: + """Craft two tiny evidence files in ``workdir``, collect both and diff.""" + a_path = _write_text(os.path.join(workdir, "run-a.txt"), DEMO_A) + b_path = _write_text(os.path.join(workdir, "run-b.txt"), DEMO_B) + snapshot_a, _stats_a = collect(files=[a_path], source="demo-a", + created="2026-01-01T00:00:00Z") + snapshot_b, _stats_b = collect(files=[b_path], source="demo-b", + created="2026-01-01T00:00:00Z") + return { + "schema": DEMO_FORMAT, + "workdir": os.path.abspath(workdir), + "snapshot_a": snapshot_a, + "snapshot_b": snapshot_b, + "diff": diff_snapshots(snapshot_a, snapshot_b), + } + + +def _load_json(path: str) -> dict[str, Any]: + try: + with open(path, "r", encoding="utf-8") as handle: + return json.load(handle) + except (OSError, json.JSONDecodeError) as exc: + raise SnapshotError(f"cannot read {path}: {exc}") from exc + + +def build_parser() -> argparse.ArgumentParser: + parser = argparse.ArgumentParser( + prog="server_snapshot.py", + description="Collect and diff canonical server-state snapshots " + "(offline evidence normalizer).", + ) + sub = parser.add_subparsers(dest="command") + + collect_cmd = sub.add_parser( + "collect", help="normalize evidence files into one snapshot") + collect_cmd.add_argument("--dir", action="append", default=[], + help="evidence/session directory (repeatable)") + collect_cmd.add_argument("--file", action="append", default=[], + help="explicit evidence file (repeatable)") + collect_cmd.add_argument( + "--include", action="append", default=[], + help="additional file-name glob under --dir (repeatable); " + f"defaults: {', '.join(DEFAULT_INCLUDES)}") + collect_cmd.add_argument("--source", help="override the snapshot source") + collect_cmd.add_argument( + "--created", help="override the ISO-8601 timestamp (SOURCE_DATE_EPOCH " + "is honoured otherwise)") + collect_cmd.add_argument("--json", help="write the snapshot here instead " + "of stdout") + + diff_cmd = sub.add_parser("diff", help="compare two snapshot JSON files") + diff_cmd.add_argument("--a", required=True, help="first snapshot JSON") + diff_cmd.add_argument("--b", required=True, help="second snapshot JSON") + diff_cmd.add_argument("--json", help="write the diff here instead of stdout") + + demo_cmd = sub.add_parser( + "demo", help="craft two tiny snapshots in a tempdir and diff them") + demo_cmd.add_argument("--out", help="keep the crafted files in this directory") + return parser + + +def _collect_command(args) -> int: + if not args.dir and not args.file: + raise SnapshotError("collect needs at least one --dir or --file") + snapshot, stats = collect(dirs=args.dir, files=args.file, + includes=args.include, source=args.source, + created=args.created) + if args.json: + path = _write_json(args.json, snapshot) + summary = { + "schema": snapshot["schema"], + "source": snapshot["source"], + "created": snapshot["created"], + "chunks": len(snapshot["chunks"]), + "words": stats["words"], + "records": len(snapshot["records"]), + "unparsed": stats["unparsed"], + "collisions": stats["collisions"], + "snapshot_json": path, + } + print(json.dumps(summary, indent=2, sort_keys=True)) + else: + print(json.dumps(snapshot, indent=2, sort_keys=True)) + print(f"server_snapshot: {stats['files']} file(s), {stats['lines']} line(s), " + f"{stats['chunks']} chunk(s), {stats['words']} word(s), " + f"{stats['records']} record(s), {stats['unparsed']} unparsed, " + f"{stats['merged']} merged, {stats['collisions']} collision(s), " + f"{stats['conflicts']} conflict(s)", file=sys.stderr) + return 0 + + +def _diff_command(args) -> int: + report = diff_snapshots(_load_json(args.a), _load_json(args.b)) + if args.json: + path = _write_json(args.json, report) + print(json.dumps({"schema": report["schema"], **report["summary"], + "diff_json": path}, indent=2, sort_keys=True)) + else: + print(json.dumps(report, indent=2, sort_keys=True)) + return 0 if report["summary"]["identical"] else 3 + + +def _demo_command(args) -> int: + if args.out: + os.makedirs(args.out, exist_ok=True) + demo = run_demo(args.out) + else: + with tempfile.TemporaryDirectory(prefix="server-snapshot-demo-") as workdir: + demo = run_demo(workdir) + for row in demo["diff"]["chunks"]: + if row["status"] != "unchanged": + print(f"{row['status']:<9} {row['label']:<16} " + f"+{len(row['added'])} -{len(row['removed'])} " + f"~{len(row['changed'])}") + print(json.dumps(demo["diff"]["summary"], indent=2, sort_keys=True)) + return 0 + + +def main(argv: list[str] | None = None) -> int: + parser = build_parser() + args = parser.parse_args(argv) + try: + if args.command == "collect": + return _collect_command(args) + if args.command == "diff": + return _diff_command(args) + if args.command == "demo": + return _demo_command(args) + parser.print_help() + return 2 + except SnapshotError as exc: + print(f"error: {exc}", file=sys.stderr) + return 2 + + +if __name__ == "__main__": + raise SystemExit(main()) |