#!/usr/bin/env python3 """Emit include/generated/sots_stream_schema.h for sots-engine from objects/streams.json. Sibling of `tools/gen_addresses.py`, same discipline: a generated file carrying **facts only**, a provenance header, never hand-edited, regenerated on every merge. What crosses the channel is the *wire schema* -- for each serializable class, the ordered sequence of items its `Write` puts on the stream: on-disk tag, on-disk primitive, and how the item is framed. What deliberately does **not** cross is every memory fact in `objects/layouts.json`: field offsets, `sizeof`, gaps and container strides. `sots-engine` is our own C++, not a byte-for-byte decomp; it must read and write the *format* faithfully and must not inherit the original's ABI in its runtime types. (The shim is the one component that legitimately needs the original ABI, because it reads the running game's memory -- those few offsets already have a home, as `offset` entries in `ghidra/addresses.json`.) The header is a *specification*, not a program. See the note on `SOTS_WIRE_UNCONDITIONAL` below and findings/objects/wire-schema-channel.md: a linear pass over `Write` cannot see its branches, so the recovered sequence is a **superset** of what any one record contains. The engine's hand-written `io()` shapes remain the codec; this table is what proves they agree with the binary, item for item, in order. Usage ----- uv run python3 tools/streams.py # -> objects/streams.json uv run python3 tools/gen_stream_schema.py \ ../sots-engine/include/generated/sots_stream_schema.h """ import datetime import json import os import subprocess import sys ROOT = os.path.join(os.path.dirname(os.path.abspath(__file__)), "..") PRIM = {None: "Unknown", "i32": "I32", "i64": "I64", "f32": "F32", "bool": "Bool", "str": "Str", "frame": "Frame", "raw": "Raw"} SHAPE = {"scalar": "Scalar", "frame": "Frame", "carr": "CArr", "narr": "NArr", "raw": "Raw"} def cstr(s): return '"' + s.replace("\\", "\\\\").replace('"', '\\"') + '"' def main(): streams = json.load(open(os.path.join(ROOT, "objects", "streams.json"))) layouts = json.load(open(os.path.join(ROOT, "objects", "layouts.json"))) try: rev = subprocess.check_output( ["git", "-C", ROOT, "rev-parse", "--short", "HEAD"]).decode().strip() except Exception: rev = "unknown" out = [ "// GENERATED -- do not edit. Wire schema of Sword of the Stars.exe (GOG 1.8.1).", f"// Source: sots-re objects/streams.json @ {rev}, generated " f"{datetime.date.today()} by tools/gen_stream_schema.py", "//", "// For each serializable class, the ordered sequence of items its Mars::IStreamable", "// Write() puts on the stream. Recovered mechanically from the serializers; the", "// order is the program order of Write, with base-class and sub-writer calls spliced", "// in at their call site, which is the on-disk order.", "//", "// FACTS ONLY. No field offsets, no sizeof, no struct strides: this header describes", "// the on-disk FORMAT, never the original's memory layout. The engine's own types are", "// free to be laid out however they like.", "//", "// SOTS_WIRE_UNCONDITIONAL -- read this before treating the table as a program.", "// The recovery is a linear pass over Write, so it cannot see Write's branches. A", "// field the game emits only under a condition (StarShip's BQ2, gated by hbq) is", "// listed unconditionally here. The sequence is therefore a SUPERSET of what any", "// single record contains, and a codec driven straight off it would desynchronise.", "// Container loops are flattened the same way: a count item is followed by its", "// element items as siblings, with `member == false` marking the elements, rather", "// than nested inside the container. Use this table to CHECK a hand-written codec,", "// not to generate one.", "//", "// `grade` is the recovery tier (verified / clean / unnamed / partial / empty) and", "// read_agree/read_comparable is the class's Read-vs-Write offset cross-check.", "#pragma once", "#include ", "#include ", "", "namespace sots::wire {", "", "// On-disk primitive, from the Stream vftable slot the writer called:", "// +0x18 str +0x1c bool +0x20 f32 +0x24 i32 +0x28 frame +0x30 raw (n=8 -> i64)", "// NOTE this is the DISK type. A member held as int16 or int8 in the original is", "// written by WriteInt and is I32 on the wire.", "enum class Prim : uint8_t { Unknown, I32, I64, F32, Bool, Str, Frame, Raw };", "", "// How the item is framed on the stream.", "// Scalar [len][tag][value], padded to 4", "// Frame [len][tag][pad] BEEFBEEF ... 41104110", "// CArr a Frame whose first item is a \".\" count, then that many elements", "// NArr a bare count item at this level, then that many elements follow it", "// Raw opaque payload whose length only the writer knows", "enum class Shape : uint8_t { Scalar, Frame, CArr, NArr, Raw };", "", "struct Field {", " const char* tag; // on-disk tag; \".\" when the writer passed NULL", " Prim prim;", " Shape shape;", " const char* of; // frame contents class name, or nullptr", " bool member; // false: a container element / loop-body write", " bool computed; // not a member: a count or a constant the writer derived", " bool unresolved; // the recovery could not type this item", "};", "", "struct Class {", " const char* name;", " const Field* fields;", " uint16_t count;", " uint16_t anon; // items whose tag is \".\"", " const char* grade;", " uint16_t read_agree;", " uint16_t read_comparable;", "};", "", ] names = sorted(streams) for cls in names: c = streams[cls] ident = "k_" + "".join(ch if ch.isalnum() else "_" for ch in cls) if not c["fields"]: out.append(f"inline constexpr Field {ident}[1] = " "{{nullptr, Prim::Unknown, Shape::Scalar, nullptr, false, false, false}};" f" // {cls}: Write emits no item") continue out.append(f"inline constexpr Field {ident}[] = {{ // {cls}") for f in c["fields"]: of = cstr(f["of"]) if f.get("of") else "nullptr" out.append(" {%s, Prim::%s, Shape::%s, %s, %s, %s, %s}," % ( cstr(f["tag"]), PRIM[f["prim"]], SHAPE[f["shape"]], of, "true" if f["member"] else "false", "true" if f.get("computed") else "false", "true" if f.get("unresolved") else "false")) out.append("};") out.append("") out.append("inline constexpr Class kClasses[] = {") for cls in names: c = streams[cls] L = layouts.get(cls, {}) ident = "k_" + "".join(ch if ch.isalnum() else "_" for ch in cls) n = len(c["fields"]) out.append(" {%s, %s, %d, %d, %s, %d, %d}," % ( cstr(cls), ident, n, c["anon"], cstr(L.get("grade") or "unknown"), L.get("read_agree") or 0, L.get("read_comparable") or 0)) out.append("};") out.append(f"inline constexpr size_t kClassCount = {len(names)};") out += [ "", "// Linear lookup by class name. The table is sorted, but it is small and this is", "// only ever called from tests and tools.", "inline const Class* find(const char* name) {", " for (const Class& c : kClasses) {", " const char* a = c.name;", " const char* b = name;", " while (*a && *a == *b) { ++a; ++b; }", " if (!*a && !*b) return &c;", " }", " return nullptr;", "}", "", "inline const char* prim_name(Prim p) {", " switch (p) {", " case Prim::Unknown: return \"?\";", " case Prim::I32: return \"i32\";", " case Prim::I64: return \"i64\";", " case Prim::F32: return \"f32\";", " case Prim::Bool: return \"bool\";", " case Prim::Str: return \"str\";", " case Prim::Frame: return \"frame\";", " case Prim::Raw: return \"raw\";", " }", " return \"?\";", "}", "", "inline const char* shape_name(Shape s) {", " switch (s) {", " case Shape::Scalar: return \"scalar\";", " case Shape::Frame: return \"frame\";", " case Shape::CArr: return \"carr\";", " case Shape::NArr: return \"narr\";", " case Shape::Raw: return \"raw\";", " }", " return \"?\";", "}", "", "} // namespace sots::wire", "", ] dest = sys.argv[1] if len(sys.argv) > 1 else os.path.join( ROOT, "objects", "generated", "sots_stream_schema.h") os.makedirs(os.path.dirname(dest), exist_ok=True) open(dest, "w").write("\n".join(out)) nf = sum(len(c["fields"]) for c in streams.values()) print(f"wrote {dest} ({len(names)} classes, {nf} wire items)") return 0 if __name__ == "__main__": sys.exit(main() or 0)