#!/usr/bin/env python3 """The *wire* projection of the serializer recovery -> objects/streams.json. `objects/layouts.json` is a **memory-layout** view: `build()` sorts fields by `off_abs` and merges two writes of the same offset into `alt_tags`. That is the right shape for a decompiler struct, and the wrong shape for a codec, because it destroys the two things the on-disk format is made of: * **order** -- the stream is a sequence, and its order is the *program* order of `Write` (with base-class / sub-writer calls spliced in at their call site). Offset order is not the same thing: 89 of the 386 classes have at least one field whose offset order differs from its write order. * **repetition** -- `SVSOJewelsOfTheCrown::Write` emits `JEWELLOCATIONID` twice, from two different members. The offset merge turns the second one into an `alt_tags` note and the record loses a field. `Lab.layout()` already returns program order; this tool just refuses to throw it away. It also drops every memory fact (`off`, `size`, `sizeof`, `gaps`, `strides`, `vftable`): a reimplementation must read and write the *format*, and must not inherit the original's ABI. What survives is exactly the wire schema -- ordinal, on-disk tag, on-disk primitive, and for framed fields the class name of the frame's contents. The disk primitive is **not** `layouts.json`'s `kind`, which is a memory type: an `int16` member and an `int8` member are both written by `Stream::WriteInt` and are four bytes on disk. It is derived from the stream vftable slot the writer called, or from the wrapper helper it called instead: +0x18 (24) string +0x1c (28) bool +0x20 (32) float +0x24 (36) int +0x28 (40) frame +0x30 (48) raw (n=8 -> int64) Usage ----- uv run python3 tools/streams.py # -> objects/streams.json uv run python3 tools/streams.py Game::StarShip # one class, readable """ import json import os import sys HERE = os.path.dirname(os.path.abspath(__file__)) sys.path.insert(0, HERE) import serializers as S # noqa: E402 OUT = os.path.join(os.path.dirname(HERE), "objects") # Wrapper helpers, by the primitive they write. `serializers.HELPERS` has the # same VAs; naming them here keeps this tool readable and independent of the # spelling used there. HELPER_PRIM = { 0x8B9D50: "i32", # WriteInt wrapper 0x8B9D00: "i32", # WriteInt16 wrapper -- widened to int32 on the wire 0x8B9C20: "bool", 0x8B9BE0: "f32", 0x8B9D70: "str", 0x8B9C60: "i64", # raw, n = 8 0x816490: "i32", # Stream::WriteNetworkObjectId -- an id, int32 on disk } SLOT_PRIM = {24: "str", 28: "bool", 32: "f32", 36: "i32", 40: "frame", 48: "raw"} def disk_prim(f): """On-disk primitive for one recovered write, or None if undecidable.""" h = f.get("helper") if h is not None: p = HELPER_PRIM.get(h) if p: return p slot = f.get("slot") if slot in SLOT_PRIM: p = SLOT_PRIM[slot] if p == "raw" and f.get("size") == 8: return "i64" return p return None def shape_of(f, prim): """How the *stream* frames this field. scalar one item: [tag][value] frame one framed item: [tag] BEEFBEEF ... 41104110 carr framed array: a frame whose first item is a "." count narr a bare int count at this level, then that many elements follow raw opaque payload whose length only the writer knows """ kind = f.get("kind", "") if kind == "vector": # A VectorHelper written through the nested slot is framed (count # inside); one written through the int wrapper is a bare count with the # elements following at the same level. return "carr" if f.get("slot") == 40 else "narr" if prim == "frame": return "frame" if prim == "raw": return "raw" return "scalar" def project(lab, cls, info): """One class -> its ordered wire schema.""" fields, ok, unknown = [], True, 0 for f in lab.layout(info["write"]): # Container element writes and loop-body reads carry `this: false`; they # describe the *element* type, which the frame's own class already # names. Computed values (a count, a constant) are not members but they # ARE items on the wire, so they are kept, tagged as such. prim = disk_prim(f) rec = { "tag": f.get("tag", "."), "prim": prim, "shape": shape_of(f, prim), "member": bool(f.get("this")), } if f.get("inner"): rec["of"] = f["inner"] if f.get("kind") in ("const",): rec["computed"] = True if f.get("unresolved") or prim is None: rec["unresolved"] = True unknown += 1 ok = False fields.append(rec) anon = sum(1 for f in fields if f["tag"] == ".") return { "class": cls, "fields": fields, "n": len(fields), "anon": anon, "unresolved": unknown, "named": anon == 0 and bool(fields), "complete": ok and bool(fields), } def run(): lab = S.Lab() out = {} for cls, info in lab.infos.items(): out[cls] = project(lab, cls, info) # Helper-only POD types (Vector3, OutputRates, ...) have no vftable of their # own and are reachable only through a specialised StreamableHelper thunk. # They are frame contents for other classes, so the codec needs them. for t, h in lab.helper_writers().items(): if t in out or not h["write"]: continue out[t] = project(lab, t, {"write": h["write"]}) return out def main(): if len(sys.argv) > 1 and sys.argv[1] != "-": lab = S.Lab() name = sys.argv[1] info = lab.infos.get(name) or lab.helper_writers().get(name) if not info: sys.exit(f"no serializer for {name!r}") p = project(lab, name, info) print(f"{name} ({p['n']} items, {p['anon']} anonymous)") for i, f in enumerate(p["fields"]): extra = f" <{f['of']}>" if f.get("of") else "" flag = " UNRESOLVED" if f.get("unresolved") else "" flag += "" if f["member"] else " [element]" flag += " [computed]" if f.get("computed") else "" print(f" {i:3d} {f['tag']!r:<20} {str(f['prim']):<6} " f"{f['shape']:<6}{extra}{flag}") return 0 out = run() os.makedirs(OUT, exist_ok=True) dest = os.path.join(OUT, "streams.json") with open(dest, "w") as fh: json.dump(out, fh, indent=1, sort_keys=True) named = sum(1 for v in out.values() if v["named"]) items = sum(v["n"] for v in out.values()) print(f"wrote {dest}: {len(out)} classes, {items} wire items, " f"{named} fully named") return 0 if __name__ == "__main__": sys.exit(main() or 0)