#!/usr/bin/env python3 """save_reader.py -- reader for Sword of the Stars (2006, v1.8) .sav files. The game serializes its world through a self-describing "Streamable" stream: container whole file is a gzip stream (an already-inflated blob is accepted too, e.g. R1's *.sav.inflate.dat test artifacts) named value [int32 len][name bytes] [value] [NUL pad to 4] complex frame [int32 len][name bytes][NUL pad to 4] 0xBEEFBEEF ... child items ... 0x41104110 (= ~0xBEEFBEEF) arrays a named int32 count followed by count x element; the "ComplexArray" flavour wraps count+elements in a frame Everything is little-endian; text is windows-1252. Values carry NO type byte -- only their name -- so a reader either knows the type from a schema or has to guess it. This module does both: * Walker generic, resync-capable tokenizer. Walks the inflated bytes as a tree of Node(name, kind, value, offset). Uses the BEEFBEEF/41104110 frames as hard synchronisation points, a catalog of known (context, name) -> type hints, and a lookahead plausibility test for unknown names. Unknown or malformed regions become `raw` nodes; parsing continues at the next frame marker or plausible tag. Nothing in the schema is required for the walk to succeed. * Shapes the recovered struct field orders (Summary, CreateParams, Sim -> Player / Sys / Flt / Ship ..., CDT) applied on top of the generic tree to produce typed dicts. Fields marked A() carry binary-confirmed on-disk names (struct-recovery.md, schema-gaps-resolved.md) and are matched by name; fields marked R() are matched by position (NULL-named "." items, or names known only from the community editor). Padding convention: the NUL padding is computed over the whole item ("joint": [len][name][value][pad]) -- confirmed on the real saves. The alternative ("split": pad after the name and after the value) is kept for --padding auto, which walks the file both ways and keeps the cleaner parse. Two facts about string VALUES matter for the plausibility test: they may be empty (len 0 = four zero bytes, byte-identical to int 0) and they may hold any windows-1252 byte (system names use 0x92 = right single quote). Only TAG bytes are constrained to printable ASCII; a value never vetoes a layout. Items the game writes with a NULL name carry the tag ".". See SAVE_FORMAT.md and findings/objects/schema-gaps-resolved.md. usage: save_reader.py SAVE [--dump] [--json] [--strict] [--padding MODE] [--out FILE] [--inflate FILE] Stdlib only. """ from __future__ import annotations import argparse import gzip import json import math import struct import sys import zlib from typing import Any, Callable, Optional __all__ = [ "BEGIN_MARK", "END_MARK", "SaveFormatError", "Issue", "Node", "Walker", "Shape", "Seq", "CArr", "NArr", "If", "Opt", "Until", "Repeat", "Rest", "A", "R", "ROOT", "apply_schema", "read_save", "read_bytes", "inflate", "dump_tree", "plain", "pad4", ] BEGIN_MARK = 0xBEEFBEEF END_MARK = 0x41104110 BEGIN_BYTES = struct.pack(" classified as int, not float def pad4(n: int) -> int: return (n + 3) & ~3 class SaveFormatError(ValueError): def __init__(self, msg: str, offset: Optional[int] = None): self.offset = offset if offset is not None: msg = f"{msg} (at inflated offset 0x{offset:x})" super().__init__(msg) class Issue: __slots__ = ("level", "path", "offset", "msg") def __init__(self, level: str, path: str, offset: int, msg: str): self.level, self.path, self.offset, self.msg = level, path, offset, msg def __repr__(self) -> str: return f"[{self.level}] @0x{self.offset:x} {self.path}: {self.msg}" def as_dict(self) -> dict: return {"level": self.level, "path": self.path, "offset": self.offset, "msg": self.msg} class Node: """One item of the generic tree. kind is one of PRIMITIVES, 'complex' or 'raw'.""" __slots__ = ("name", "kind", "value", "offset", "size", "children", "raw", "alt", "hinted") def __init__(self, name, kind, value=None, offset=0, size=0, children=None, raw=b"", alt=None, hinted=False): self.name = name # str, "" for an empty tag, None for a tagless frame self.kind = kind self.value = value self.offset = offset # inflated offset of the item's tag (or frame marker) self.size = size # bytes consumed incl. tag, padding and frame markers self.children = children if children is not None else [] self.raw = raw # value bytes (scalars / raw) for re-typing self.alt = alt # alternative reading of a 4-byte word self.hinted = hinted # type came from the catalog/schema, not a guess @property def is_complex(self) -> bool: return self.kind == "complex" def __repr__(self) -> str: return f"Node({self.name!r}, {self.kind}, {self.value!r} @0x{self.offset:x})" # --- container --------------------------------------------------------------- def inflate(data: bytes) -> bytes: """gzip -> bytes. Data without the gzip magic is returned unchanged.""" if data[:2] == b"\x1f\x8b": try: return gzip.decompress(data) except (EOFError, OSError, zlib.error) as e: raise SaveFormatError(f"gzip container is damaged: {e}") from None return data # --- schema description objects --------------------------------------------- # Field types: a primitive name, "vec3" (frame of 3 floats, named or not), # "any" (keep generic), "raw", a Shape (framed struct), Seq (unframed group), # CArr (framed count+elements), NArr (unframed count+elements). class Field: __slots__ = ("name", "type", "key", "auth", "flex") def __init__(self, name, type_, key=None, auth=False, flex=False): self.name = name # expected on-disk tag (auth) or R1 name (not auth) self.type = type_ self.key = key or name # key in the typed dict self.auth = auth self.flex = flex # Shape/CArr may also appear inline (unframed) def __repr__(self): return f"{'A' if self.auth else 'R'}({self.name!r})" def A(name, type_, key=None, flex=False) -> Field: """Field whose on-disk name is binary-confirmed (struct-recovery.md).""" return Field(name, type_, key, auth=True, flex=flex) def R(name, type_, key=None, flex=False) -> Field: """Field known only by the community editor's C# name (matched by position).""" return Field(name, type_, key, auth=False, flex=flex) class Seq: """Unframed ordered group of fields (R1 'leaf' structs, array element bodies).""" __slots__ = ("fields",) def __init__(self, fields): self.fields = list(fields) class Shape: """Framed struct: name tag + BEEFBEEF + fields + 41104110.""" __slots__ = ("name", "fields", "_prefix", "_by_name") def __init__(self, name, fields): self.name = name self.fields = list(fields) self._prefix = None self._by_name = None def prefix(self) -> list: """Positional child hints up to the first variable-length construct.""" if self._prefix is None: self._prefix = _prefix_of(self.fields) return self._prefix def by_name(self) -> dict: if self._by_name is None: d = {} _collect_names(self.fields, d) self._by_name = d return self._by_name class CArr: __slots__ = ("elem",) def __init__(self, elem): self.elem = elem # primitive, Shape, Seq, or Field class NArr: __slots__ = ("elem",) def __init__(self, elem): self.elem = elem class If: """Conditional group: parsed iff out[key] == equals (default True), or, with greater_than=N, iff out[key] is a number > N (e.g. PrMax > 0).""" __slots__ = ("key", "inner", "equals", "greater_than") def __init__(self, key, inner, equals=True, greater_than=None): self.key, self.inner, self.equals, self.greater_than = key, inner, equals, greater_than def holds(self, out: dict) -> bool: v = out.get(self.key) if self.greater_than is not None: return isinstance(v, (int, float)) and not isinstance(v, bool) and v > self.greater_than return v == self.equals class Opt: """Optional named item: consumed only when the next tag matches.""" __slots__ = ("field",) def __init__(self, name, type_, key=None): self.field = Field(name, type_, key, auth=True) class Until: """Collect generic items until a tag named stop appears (exclusive).""" __slots__ = ("key", "stop") def __init__(self, key, stop): self.key, self.stop = key, stop class Repeat: """Uncounted repetition: consume `elem` (a Field or Seq whose first field is an A() tag) again and again while the next item carries that tag. Used for lists the writer emits with no count (turnstats `stats` frames until the frame END, the 7 `ISsp`/`ISsu` pairs, the root `CD` frames).""" __slots__ = ("elem", "key", "lead") def __init__(self, elem, key=None): self.elem = elem first = elem.fields[0] if isinstance(elem, Seq) else elem if not isinstance(first, Field): raise TypeError("Repeat needs a Field or a Seq starting with a Field") self.lead = first.name self.key = key or first.key class Rest: """Collect all remaining items generically.""" __slots__ = ("key",) def __init__(self, key="_rest"): self.key = key def _prefix_of(fields) -> list: out = [] for f in fields: if isinstance(f, Field): if f.flex: break t = f.type if isinstance(t, Seq): sub = _prefix_of(t.fields) out.extend(sub) if len(sub) != _count_items(t.fields): break elif isinstance(t, NArr): out.append("int") break elif t in ("any", "raw"): out.append(None) else: out.append(t) # primitive kind, "vec3", Shape, CArr else: break # If / Opt / Until / Rest end positional certainty return out def _count_items(fields) -> int: n = 0 for f in fields: if isinstance(f, Field) and isinstance(f.type, Seq): n += _count_items(f.type.fields) else: n += 1 return n def _collect_names(fields, d): for f in fields: if isinstance(f, Field): d.setdefault(f.name, f.type) d.setdefault(f.name.lower(), f.type) t = f.type if isinstance(t, Seq): _collect_names(t.fields, d) elif isinstance(t, NArr): e = t.elem if isinstance(e, Seq): _collect_names(e.fields, d) elif isinstance(e, Field): _collect_names([e], d) elif isinstance(f, Opt): _collect_names([f.field], d) elif isinstance(f, If): inner = f.inner _collect_names(inner.fields if isinstance(inner, Seq) else [inner], d) elif isinstance(f, Repeat): e = f.elem _collect_names(e.fields if isinstance(e, Seq) else [e], d) # --- schema: recovered struct layouts ----------------------------------------- # Sources: save-editor-structs.md (R1/R2 community editors) with the corrections # from struct-recovery.md (binary). A() names are on-disk tags confirmed in the # exe; R() names are R1's C# field names (disk spelling unknown -> positional). # Game::PlayerColorID: "." int index; iff -1 the three "." ints r,g,b follow. # Framed under ClrID (Player), indcl (IndependenceInfo) and FxCrID (SlotDef). PlayerColor = Shape(None, [ R("idx", "int"), If("idx", Seq([R("r", "int"), R("g", "int"), R("b", "int")]), equals=-1), Rest(), ]) # -- Summary = StrategyGameInfo::Write @0x00829960 (schema-gaps-resolved.md §5.2) # Every tag below is confirmed in the binary. Items whose writer passes a NULL # name are on disk as "." and are matched positionally (R()). PlayerSettings = Shape("Settings", [ # StrategyPlayerGameSettings: 4 x "." int R("treasury", "int"), R("colonies", "int"), R("techs", "int"), R("difficulty", "int"), Rest(), ]) Slot = Shape("Slot", [ # SlotDef::Write @0x008276d0 A("IsPlay", "bool"), A("IsDead", "bool"), A("IsReq", "bool"), A("IsRec", "bool"), A("IsFxNm", "bool"), A("FxNm", "string"), A("IsFxSp", "bool"), A("FxSp", "int"), A("IsFxCr", "bool"), A("FxCrID", PlayerColor), A("IsFxBd", "bool"), A("FxBd", "string"), A("IsFxAv", "bool"), A("FxAv", "string"), A("Tag", "int"), A("Pwd", "string"), A("Team", "int"), A("Settings", PlayerSettings), Rest(), ]) PlayerInfo = Shape(None, [A("Slot", Slot), A("Rank", "int"), Rest()]) # element tag "." Tmrs = Shape("TMRS", [A("TSTL", "float"), A("TCTL", "float"), A("TQTL", "float"), A("TQTLE", "float"), Rest()]) Session = Shape("Session", [A("TMRS", Tmrs), Rest()]) Summary = Shape("Summary", [ A("GameName", "string"), A("Turn", "int"), A("NumSys", "int"), A("Checksum", "int"), A("Players", CArr(PlayerInfo)), A("Session", Session), A("MapShape", "int"), A("IncMod", "float"), A("ResMod", "float"), A("Alliances", "bool"), A("Teams", "bool"), A("Encounters", "bool"), A("Scenario", "string"), Rest(), ]) # -- CreateParams = StrategyGameCreateParams::Write @0x0082ae40 (§5.1) ---------- Planet = Shape(None, [ # SystemParams element, all tags "." # p1 is a STRING, not an int (Game::SystemParams wire schema: frame, str, i32, # i32, f32 -- lane G's wire-schema conformance check, findings/objects/ # wire-schema-channel.md §3.1). It is the empty string in every save we hold, # and an empty string is four zero bytes -- byte-identical to the int 0 -- so # this round-tripped by luck. A named system would desynchronise the reader. R("pos", "vec3"), R("p1", "string"), R("p2", "int"), R("p3", "int"), R("p4", "float"), Rest(), ]) MapP = Shape("MapP", [ # StarMapParams::Write @0x00727a10, all tags "." R("mapType", "int"), R("planets", CArr(Planet)), # VectorHelper R("players", NArr(CArr("int"))), # "." count, n x "." frame{ "." count, n x "." int } R("nodePaths", CArr("any")), # VectorHelper, empty in the real saves Rest(), ]) Scrp = Shape("scrp", [ # StrategyScriptParams::Write @0x0082ad40 A("spc", NArr(Seq([A("spsn", "string"), A("sppn", "string"), A("sppv", "string")])), key="params"), Rest(), ]) CreateParams = Shape("CreateParams", [ A("Name", "string"), A("ID", "int"), A("RSeed", "int"), A("AID", "int"), A("Key", "string"), # Key may be "" A("MapP", MapP), A("MapS", "int"), A("MapF", "string"), A("NSys", "int"), A("REnc", "float"), A("SDist", "float"), A("SSize", "float"), A("SRes", "float"), A("SSuit", "float"), A("MaxP", "int"), A("ASpec", "int"), A("bAlly", "bool"), A("NTeam", "int"), A("tmgrp", "bool"), A("PSav", "int"), A("PCol", "int"), A("PTech", "int"), A("IncM", "float"), A("ResM", "float"), A("scrp", Scrp), Rest(), ]) # -- shared sim types (struct-recovery §1.4, §1.5, §1.6, §1.7) ------------------ PopG = Shape("PopG", [A("PopT", "int"), A("PopS", "int"), A("PopC", "int64"), Rest()]) Population = Shape(None, [A("PopNG", NArr(A("PopG", PopG)), key="groups"), Rest()]) Morale = Shape(None, [A("mnsp", NArr(Seq([A("msp", "int"), A("mv", "int")])), key="entries"), Rest()]) MoraleEvent = Shape(None, [ A("mid", "int"), A("mtr", "int"), A("mn", "int"), A("mtp", "int"), A("mfx", Morale), A("mdsc", "string"), Rest(), ]) BuildOrder = Shape(None, [ A("desID", "int"), A("con", "int"), A("conleft", "int"), A("sav", "int"), A("ordID", "int"), Rest(), ]) BuildQueue = Shape("BQ", [A("ords", CArr(BuildOrder)), Rest()]) # `ords` = framed VectorHelper (§10) IndependenceInfo = Shape("indi", [ A("indsp", "int"), A("indcl", PlayerColor), A("indnm", "string"), A("indav", "string"), A("indba", "string"), Rest(), ]) Rts = Shape("Rts", [ A("SRs", "float"), A("SRt", "float"), A("SRsc", "float"), A("SRtf", "float"), A("SRi", "float"), A("SRoh", "float"), A("SRnr", "int"), Rest(), ]) PlayerView = Shape("pview", [ A("VTrn", "int"), A("Pop", "int"), A("Pop2", Population), A("Infra", "float"), A("Suit", "float"), A("Res", "int"), Opt("ARes", "int"), A("ARes2", "int"), A("MRes", "int"), A("NoRebAI", "bool"), A("pbon", "int"), A("pbon2", Population), A("ibon", "float"), A("TerrFl", "int"), A("footer", "bool"), Rest(), ]) # -- star system (struct-recovery §1, disk order) -------------------------------- Sys = Shape("Sys", [ A("Pos", "vec3"), A("R", "float"), A("G", "float"), A("B", "float"), A("A", "float"), A("Idx", "int"), A("Size", "int"), Opt("ISuit", "float"), A("Suit", "float"), A("Res", "int"), Opt("ARes", "int"), A("ARes2", "int"), A("MRes", "int"), A("NoRebAI", "bool"), A("TRes", "int"), A("Pop", "int"), A("Pop2", Population), A("Infra", "float"), A("PvPop", "int"), A("PvPop2", Population), A("PvInfra", "float"), A("PvSuit", "float"), A("PvRes", "int"), A("PvARes2", "int"), A("PvMRes", "int"), A("PvNoRebAI", "bool"), A("Rts", Rts), A("Abdn", "bool"), A("Dstyd", "bool"), A("TnsOH", "int"), A("OutMod", "float"), A("RepCur", "float"), A("RepMax", "float"), A("ntdev", "int"), A("pbon", "int"), A("pbon2", Population), A("ibon", "float"), A("ltis", "int"), A("rbfl", "int"), A("rbtn", "int"), A("rbfr", "int"), A("rbwn", "int"), A("hsrg", "bool"), A("haltc", NArr(Seq([A("haltt", "int"), A("haltv", "bool")])), key="halt"), A("vnh", "bool"), If("vnh", Seq([A("vnd", "bool"), A("vnex3", "bool"), A("vnpex3", "bool")])), A("Name", "string"), A("VFlags", "int"), A("EFlags", "int"), A("AFlags", "int"), A("FFlags", "int"), A("GFlags", "int"), A("Bats2", "int64"), A("rcex", "int64"), A("MnRFlags", "int"), A("RfRFlags", "int"), A("ClkFlags", "int"), A("EggScio", "int"), A("TerrFl", "int"), A("TAcq", "int"), A("TFAcq", "int"), A("TDst", "int"), A("dcs", Population), A("dsu", "float"), A("cm", Morale), A("PvCM", Morale), A("cme2", CArr(MoraleEvent), flex=True), A("spies2", "any"), A("PID", "int"), A("DefF", "int"), A("DefSF", "int"), Opt("BQ", BuildQueue), A("nadct", NArr(Seq([A("ads", "int"), A("adt", "int")])), key="adct"), A("NumPlgs2", NArr(Seq([A("PlgT", "int"), A("Plg", "any")])), key="plagues"), A("NumFlts", NArr(A("Flt", "int")), key="fleets"), A("NumGFs", NArr(A("GF", "int")), key="gates"), A("NumSnF", NArr(A("SnF", "int")), key="stations"), A("NumMnF", NArr(A("MnF", "int")), key="monitors"), # NVO entry: indi is an inline member of the map value and is written for # every colony, NOT gated by isind (ServerSystem::Write @0x00749630). A("NVO", NArr(Seq([ A("PID", "int"), A("TShn", "int"), A("OID", "int"), A("isind", "bool"), A("indi", IndependenceInfo), ])), key="colonies"), A("NVE", NArr(Seq([A("EPid", "int"), A("ETS", "int"), A("Eid", "int")])), key="nve"), A("NVs", NArr(Seq([R("pid", "int"), R("pview", PlayerView)])), key="views"), A("hindi", "bool"), If("hindi", A("indi", IndependenceInfo)), Rest(), ]) # -- player / empire (struct-recovery §2, disk order) ---------------------------- Alliances = Shape("Team", [A("ALid", "int"), A("AL", "int"), A("NA", "int"), A("CF", "int"), Rest()]) DipDetail = Seq # helper alias for readability DipStat = Shape(None, [ A("other", "int"), A("lastnap", "int"), A("lastnapbty", "int"), A("bknnap", "int"), A("btynap", "int"), A("lastally", "int"), A("lastallybty", "int"), A("bknally", "int"), A("btyally", "int"), A("lastcf", "int"), A("lastcfbty", "int"), A("bkncf", "int"), A("btycf", "int"), A("deadhome", "int"), Rest(), ]) Prep = Shape(None, [ A("oid", "int"), A("pid", "int"), A("flds", "int"), A("sav", "int"), A("home", "int"), A("ncol", "int"), A("mpwr", "int"), A("mcls", "int"), A("mmsl", "int"), A("nshp", "int"), A("nsat", "int"), Rest(), ]) # Objective records (elements of the framed odes/owep/otch arrays). The lead tag # is `otnF` on disk (R1 had `ontF`; positional matching hid the typo -- caught by # the C++ round-trip writer and confirmed in the bytes, SAVE_FORMAT.md §10). Odes = Shape(None, [A("otnF", "int"), A("otnL", "int"), A("odid", "int"), A("opid", "int"), Rest()]) # `odet` is a BOOL, not an int: Game::ObservedWeapon / Game::ObservedTech call the # bool writer (wire-schema-channel.md §3.2). It is byte-safe here only because the # tag is 4 characters, which makes a bool item and an int item both 12 bytes; a 3- # or 5-character tag would not have been so forgiving. Owep = Shape(None, [A("otnF", "int"), A("otnL", "int"), A("odet", "bool"), A("owep", "string"), A("owith", "int"), Rest()]) Otch = Shape(None, [A("otnF", "int"), A("otnL", "int"), A("odet", "bool"), A("otch", "string"), A("owith", "int"), Rest()]) Note = Shape("Nts", [A("NtSys", "int"), A("NtTxt", "string"), A("NtTrn", "int"), Rest()]) # Game::ShipRecords. TWO COUNTED SECTIONS, and `srbd` is the second COUNT, not a # field (wire-schema-channel.md §3.4). The recovery lists srd/src/srb/srl/sri # right after `srbd` as loop-body writes, and the saves EXERCISE it: srbd takes the # values 0, 1, 3 and 4 across the players of the four saves we hold, and every # non-zero count is followed by exactly srbd x 5 scalars (e.g. turn3 player 4 has # srbd == 4 and 20 items). So this one is behaviourally confirmed, not inferred. DesignRecord = Seq([ A("srd", "int"), A("src", "int"), A("srb", "int"), A("srl", "int"), A("sri", "int"), ]) ShipRecords = Shape("ShipRecs", [ A("srnc", NArr(Seq([A("srb", "int"), A("srl", "int"), A("srk", "int"), A("sri", "int")])), key="records"), A("srbd", NArr(DesignRecord), key="designRecords"), Rest(), ]) # Game::SpeciesRatios. `nv` is the COUNT of (sp, va2) pairs, not a field # (wire-schema-channel.md §3.3): the recovery marks sp and va2 as loop-body writes # and the saves agree -- nv==1 frames carry one pair, nv==0 frames carry nothing. SpeciesRatios = Shape("spe", [ A("nv", NArr(Seq([A("sp", "int"), A("va2", "int")])), key="ratios"), Rest(), ]) CivilianRatios = Shape("civr", [A("smx", "float"), A("spe", SpeciesRatios), Rest()]) # Design header: on-disk tags are FAIDes/DHide/DWep/DName (R1's camel-case names # were wrong). The typed-dict keys keep R1's spelling because # verify/design-rules/stock_designs.py consumes them. Design = Shape("Des", [ A("FAIDes", "bool", key="faiDes"), A("DHide", "bool", key="dHide"), A("DWep", "int", key="dWep"), A("DName", "string", key="dName"), Rest("sections"), ]) ConMods = Shape("ConMods", [ A("ConMod", "float", key="ConMod0"), A("SavMod", "float", key="SavMod0"), A("ConMod", "float", key="ConMod1"), A("SavMod", "float", key="SavMod1"), A("ConMod", "float", key="ConMod2"), A("SavMod", "float", key="SavMod2"), ]) Player = Shape("Player", [ A("TechTree", "any"), A("HomeSys", "int"), A("PlyrIdx", "int"), A("PlryName", "string"), A("Species", "int"), A("ClrID", PlayerColor), A("Bdg", "string"), A("Avt", "string"), A("Team", "int"), A("Sav", "int"), A("IdealSuit", "float"), A("SuitTol", "float"), A("MaxOH", "float"), A("ResRate", "float"), A("ResMod", "float"), A("ResScl", "float"), A("TRM", "float"), A("TRP", "int"), A("TRA", "int"), A("OutMod", "float"), A("RebOutMod", "float"), A("ScOutMod", "float"), A("IncMod", "float"), Opt("SensMod", "float"), Opt("ExPopSys", "int"), A("PopMod", "float"), A("TerraMod", "float"), A("AMine", "bool"), A("MinPure", "float"), A("MinRate", "float"), A("NGts", "int"), A("PrGtTrf", "int"), A("GTraf", "int"), A("CstR", "float"), A("CstE", "float"), A("CstT", "float"), A("Maint", "int"), A("shrm", "float"), A("Status", "int"), A("Elim", "bool"), A("NPC", "bool"), A("RebAI", "bool"), A("ReqCL", "bool"), A("Team", Alliances, key="Alliances"), A("HasVac", "int"), A("HasImm", "int"), A("NPTrk", "int"), A("HasDisc", "int"), A("HasDiscSp", "int"), A("HasDiscCl", "int"), A("HasEnc", "int"), A("HasEng", "int"), A("Events", "any"), A("FNG", "any"), A("PvSav", "int"), A("PvMA", "bool"), A("AIBn", "bool"), A("CnTrd", "bool"), A("CnRad", "bool"), A("hgs", "bool"), A("hadvs", "bool"), A("harcc", "bool"), A("CnVItl", "bool"), A("pddm", "float"), A("BnkWrn", "int"), A("BnkTrn", "int"), A("BnkPr", "int"), A("BnkEl", "int"), A("ShipRecs", ShipRecords, key="shipRecs"), A("NextPrjID", "int"), A("plcy", "int"), A("pswd", "string"), A("lret", "int"), A("nmeid", "int"), A("cdp", "bool"), A("spy2", "any"), A("civr", CivilianRatios), A("aidf", "int"), A("Srn", "bool"), A("SrnTo", "int"), A("lboid", "int"), Opt("lcid", "int"), A("lcid2", "int"), A("ResTNm", "string"), A("ResErrRoll", "bool"), R("conMods", ConMods, flex=True), A("NumOwn", NArr(A("OwnId", "int")), key="owners"), A("NumDes", NArr(Seq([A("DesID", "int"), A("Des", Design)])), key="designs"), A("NumLeg", NArr(Seq([A("DesID", "int"), A("Des", Design)])), key="legacyDesigns"), A("NumNotes", NArr(A("Nts", Note)), key="notes"), A("NumPR", NArr(Seq([A("PRm", "float"), A("PRBt", "int")])), key="pr"), A("HasAIR", "bool"), If("HasAIR", A("AIR", "any")), A("cta", "bool"), A("AIEnf", "any"), A("NSprj", NArr(Seq([A("SprjT", "int"), A("Sprj", "any")])), key="specialProjects"), A("Nexp", NArr(Seq([A("xid", "int"), A("xmin", "int"), A("xmax", "int"), A("xper", "float")])), key="nexp"), A("NWeapXcl", NArr(A("WeapXcl", "int")), key="weapXcl"), A("Ojvs", "any"), A("dipstats", CArr(DipStat)), A("comms", "any"), A("preps", CArr(Prep)), A("odes", CArr(Odes)), A("owep", CArr(Owep)), A("otch", CArr(Otch)), A("aid", "any"), A("ndeflay", NArr(A("deflay", "any")), key="defLayouts"), A("rdtc", NArr(A("rdt", "any")), key="raidTargets"), A("tnc", "int"), Rest(), ]) # -- fleets & ships (struct-recovery §3, §4) -------------------------------------- NodeRoute = Shape("nrt", [A("nrp", "int"), A("nrf", "int"), A("nrt", "int"), Rest()]) Waypoint = Shape(None, [A("Wpt", "int"), A("Tp", "int"), R("nrt", NodeRoute, flex=True), Rest()]) FlightPlan = Shape("FPlan", [ A("wpts", CArr(Waypoint)), A("FPsp2", "float"), A("FPeta2", "int"), # `wpts` = framed VectorHelper (§10) A("FPogn2", "vec3"), A("FPdpos", "vec3"), A("pnd", "int"), Rest(), ]) PrisonerHold = Shape("PrisH", [ # PrisonerHold::Write @0x0056ec00 A("PrMax", "int"), If("PrMax", A("PrNSp", NArr(Seq([A("PrSp", "int"), A("PrNum", "int")])), key="prisoners"), greater_than=0), Rest(), ]) Ship = Shape("Ship", [ A("DesID", "int"), A("FltID", "int"), A("PlrID", "int"), A("Range", "float"), A("Health", "vec3"), A("ConCap", "int"), A("RefCap", "float"), A("RepCap", "float"), A("MineCap", "int"), A("Plg", "int"), A("Act", "int"), A("Dep", "bool"), A("Atq", "bool"), A("EncID", "int"), A("PrisH", PrisonerHold), A("LCT", "int"), A("tsd", "int"), A("atsp", "int"), A("tblt", "int"), A("hbq", "bool"), If("hbq", A("BQ2", BuildQueue)), A("hsp", "bool"), If("hsp", Seq([A("pop", Population), A("ppop", Population)])), A("NTH", NArr(Seq([A("TH", "float"), A("THM", "float")])), key="thrusters"), Rest(), ]) Fleet = Shape("Flt", [ A("Pos", "vec3"), A("PID", "int"), A("LocID", "int"), Opt("SysID", "int"), Opt("TrdID", "int"), A("HFPlan", "bool"), If("HFPlan", A("FPlan", FlightPlan)), A("FtName", "string"), Opt("Caps", "int"), Opt("GtTrf", "int"), Opt("FtSens", "float"), Opt("FtInc", "int"), A("FtTrans", "int"), A("FtOrig", "vec3"), A("FtFlg", "int"), A("Ftae", "int"), A("Ftpae", "int"), A("FtEnc", "int"), A("FtMS", "int"), A("Perm", "bool"), A("PrvPos", "vec3"), A("HLay", "bool"), If("HLay", A("Lay", "any")), A("NShips", NArr(Seq([A("ShipID", "int"), A("Ship", Ship)])), key="ships"), Rest(), ]) # -- combat reports, node grid (R1 §9, §11) ------------------------------------ Dams = Shape("dams", [R("dams", "int"), R("damp", "int"), R("dami", "int"), R("damt", "int"), Rest()]) Wrep = Shape(None, [R("wep", "string"), R("dams", Dams, flex=True), Rest()]) CrepPrep = Shape(None, [ R("plr", "int"), R("ai", "bool"), R("ally", "int"), R("status", "int"), R("mxeng", "int"), R("mxcls", "int"), R("mxmsl", "int"), Rest(), ]) Crep = Shape("crep", [ R("cid", "int"), R("trn", "int"), R("pos", "vec3"), R("sid", "int"), R("auto", "int"), R("dur", "int"), R("cow", "int"), R("cdst", "int"), R("cpk", "int"), R("cpt", "int"), R("cdt", "int"), R("cdi", "int"), R("prep", CArr(CrepPrep), flex=True), R("wrep", CArr(Wrep), flex=True), Rest(), ]) NodePath = Shape(None, [ R("npt", "int"), R("npid", "int"), R("npfr", "int"), R("npto", "int"), R("npctm", "int"), R("npcby", "int"), R("npdtn", "int"), R("npdtf", "int"), R("npenp", "int"), R("npuse", "int"), R("nptf", "int"), Rest(), ]) # NdGr2: `paths` (framed VectorHelper) then `nextid` -- all lower-case on disk # (R1 had `nextId`; hidden by positional matching, see SAVE_FORMAT.md §10). NodeGrid = Shape("NdGr2", [A("paths", CArr(NodePath)), A("nextid", "int"), Rest()]) # -- turn statistics (schema-gaps-resolved.md §6) -------------------------------- SystemEvent = Shape(None, [ # SystemEvent::Write @0x008189a0, element tag "." A("set", "int"), A("ses", "int"), A("seop", "int"), A("senp", "int"), A("seno2", NArr(A("seot2", "int")), key="others"), Rest(), ]) ClassStats = Seq([A("cls", "int"), A("shpt", "int"), A("shpl", "int"), A("shpk", "int"), A("satt", "int"), A("satl", "int"), A("satk", "int")]) PlayerTurnStats = Shape("stats", [ # PlayerTurnStats::Write @0x0082c290 A("pop", "int64"), A("sacq", CArr(SystemEvent)), A("slost", CArr(SystemEvent)), A("trn", "int"), A("almem", "int"), A("inc", "int"), A("tdinc", "int"), A("sav", "int"), A("col", "int"), A("bat", "int"), A("tch", "int"), # tch: INT (int16 widened) A("ncls", NArr(ClassStats), key="classes"), Rest(), ]) PlayerTurnHistory = Shape("hist", [ # PlayerTurnHistory::Write @0x0082c4a0 A("ply", "int"), Repeat(A("stats", PlayerTurnStats), key="stats"), # no count: until frame END ]) TurnStats = Shape("turnstats", [ # GameTurnHistory::Write @0x0082c5a0 A("nply", NArr(Seq([A("ply", "int"), A("hist", PlayerTurnHistory)])), key="players"), Rest(), ]) # -- Sim block = StrategyServer::Write @0x0079fa70 (schema-gaps-resolved.md §8) ---- Invasion = Seq([A("invs", "int"), A("inve", "int"), A("invt", "int"), A("invtb", "int")]) IdList = NArr(R(".", "int")) # FUN_00794cd0: count, n x "." int Sim = Shape("Sim", [ A("KeyPath", "string"), A("NMSz", "int"), A("NMLc", "int"), A("NMnx", "int"), A("PlayerIDs", IdList, key="playerIds"), A("DesignIDs", IdList, key="designIds"), A("SystemIDs", IdList, key="systemIds"), A("FleetIDs", IdList, key="fleetIds"), A("ShipIDs", IdList, key="shipIds"), A("TradeIDs", IdList, key="tradeIds"), A("ModCount", "int"), A("Frame", "int"), A("GameID", "int"), Opt("AIDifficultyID", "int"), A("Attrib", "any"), Opt("Rand", "int"), A("RNG", "any"), A("GameName", "string"), A("Map", "int"), A("IncMod", "float"), A("ResMod", "float"), Opt("RandEnc", "bool"), A("EnAl", "bool"), A("EnTm", "bool"), A("GOTurn", "int"), A("GOWinPly", CArr("int")), # VectorHelper A("NPCm", "int"), A("NPCo", "int"), A("NPCi", "int"), A("NPCv", "int"), A("NPCa", "int"), Opt("NPC", "int"), A("szadj", "float"), A("rsadj", "float"), A("suadj", "float"), A("sprjs", "any"), A("RandEncAdj", "float"), A("cmbtid", "int"), A("turnstats", TurnStats), A("numcreps", NArr(A("crep", Crep)), key="combatReports"), A("ninv", NArr(Invasion), key="invasions"), A("AllExc", NArr(Seq([A("AllExc", "int"), A("AllExc", "int")])), key="exclusions3"), A("AllExc", NArr(Seq([A("AllExc", "int"), A("AllExc", "int")])), key="exclusions2"), A("AllExcCF", NArr(Seq([A("AllExcCFp", "int"), A("AllExcCFp", "int")])), key="exclusionsCF"), A("NumPlrs", NArr(Seq([A("PlayerID", "int"), A("Player", Player)])), key="players"), Repeat(Seq([A("ISsp", "string"), A("ISsu", "float")]), key="species"), # 7 pairs, no count A("NumSys", NArr(Seq([A("SysID", "int"), A("Sys", Sys)])), key="systems"), A("NdGr2", NodeGrid), A("trdmgr", "any"), A("spymgr", "any"), A("NumFlts", NArr(Seq([A("FltID", "int"), A("Flt", Fleet)])), key="fleets"), A("NumActs", NArr(A("Act", "int")), key="acts"), Opt("SvSctOb", "any"), # only if pointer != NULL A("zdsc", NArr(Seq([A("zdsi", "int"), A("zdst", "int")])), key="zoneDefence"), Rest(), ]) # -- file root (FUN_00877070): Summary, CreateParams, Sim, CDT, then one opaque # CD frame per custom-data id that has a blob (Player..TurnCommands_v5 / .AIAgent). CdTable = Shape("CDT", [A("NumIDs", NArr(A("ID", "string")), key="ids"), Rest()]) ROOT = Seq([ A("Summary", Summary, key="summary"), A("CreateParams", CreateParams, key="createParams"), A("Sim", Sim, key="sim"), A("CDT", CdTable, key="cdTable"), Repeat(A("CD", "any"), key="customData"), ]) # Hand-maintained kinds for names that only occur inside generic ("any") regions. MANUAL_KINDS = { # tech tree / designs (R1 §7) "TNm": "string", "tfc": "bool", "tResCost": "int", "tResDone": "int", "tAcq": "int", # (R1's `dName`/`faiDes`/`dHide` never occur on disk; the real Design tags # FAIDes/DHide/DWep/DName are registered from the Design shape) "tiAcq": "int", "tUnlck": "int", "wfn": "string", "bId": "bool", # objectives / comms / research / encounters (R1 §5, §6, §11) "cmp": "bool", "dsc": "string", "xcsn": "string", "nm": "string", "ntg": "string", "wep": "string", "gmch": "int", "drad": "float", "cst": "float", "aOdd": "float", "aInc": "float", "rMd": "float", "sctSize": "float", "smx": "float", "nPrvVa": "float", "crPce": "bool", "maintHf": "bool", "caps2": "int64", "tRsld": "bool", "tRsldd": "bool", "tRsldc": "bool", "tRsldi": "bool", "tRsldr": "bool", "prm": "float", } def _register(shape, kinds: dict, conflicts: set, shapes: dict, seen: set, weak: bool = False): """Collect global (name -> kind) hints and (name -> shape) frames. weak=False registers A()/R() fields (conflicting kinds are dropped later); weak=True registers Opt() legacy tags only where nothing else claims the name.""" if id(shape) in seen: return seen.add(id(shape)) if isinstance(shape, Shape) and shape.name: shapes.setdefault(shape.name, shape) fields = shape.fields if isinstance(shape, (Shape, Seq)) else [] for f in fields: items, is_opt = [f], False if isinstance(f, Opt): items, is_opt = [f.field], True elif isinstance(f, If): items = f.inner.fields if isinstance(f.inner, Seq) else [f.inner] elif isinstance(f, Repeat): items = f.elem.fields if isinstance(f.elem, Seq) else [f.elem] elif not isinstance(f, Field): continue for fld in items: if not isinstance(fld, Field): continue _register_type(fld.name, fld.type, kinds, conflicts, shapes, seen, weak, is_opt) def _register_type(name, t, kinds, conflicts, shapes, seen, weak=False, is_opt=False): if isinstance(t, str): if t in PRIMITIVES: if weak: if is_opt and name not in conflicts: kinds.setdefault(name, t) elif not is_opt: if name in kinds and kinds[name] != t: conflicts.add(name) kinds.setdefault(name, t) elif isinstance(t, Shape): if name and not t.name: shapes.setdefault(name, t) _register(t, kinds, conflicts, shapes, seen, weak) elif isinstance(t, Seq): _register(t, kinds, conflicts, shapes, seen, weak) elif isinstance(t, (CArr, NArr)): e = t.elem if isinstance(e, Field): _register_type(e.name, e.type, kinds, conflicts, shapes, seen, weak) elif isinstance(e, (Shape, Seq)): _register(e, kinds, conflicts, shapes, seen, weak) if isinstance(t, CArr) and name: shapes.setdefault(name, t) def _build_catalog(): kinds, conflicts, shapes = {}, set(), {} _register(ROOT, kinds, conflicts, shapes, set()) for k in conflicts: # ambiguous across contexts -> no global hint kinds.pop(k, None) _register(ROOT, kinds, conflicts, shapes, set(), weak=True) # legacy Opt() tags, lowest priority kinds.pop(".", None) # the NULL-name tag carries ints, floats, frames -- never hint it for k, v in MANUAL_KINDS.items(): kinds[k] = v return kinds, shapes GLOBAL_KINDS, GLOBAL_SHAPES = _build_catalog() # Frames whose body is an opaque byte blob (read straight to the END marker). RAW_FRAMES = {"RNG"} # --- generic walker ----------------------------------------------------------- _CP1252_UNDEFINED = frozenset((0x81, 0x8d, 0x8f, 0x90, 0x9d)) def _text_plausible(b: bytes) -> bool: """Guard used only while GUESSING that an unknown tag holds a string. Every byte windows-1252 defines counts as text -- including 0x80-0x9f (system names like "Kor\\x92Voth" use 0x92 = right single quote). Never apply this to a tag that is known to be a string.""" if not b: return True ok = sum(1 for c in b if (0x20 <= c < 0x7f) or (c >= 0x80 and c not in _CP1252_UNDEFINED) or c in (9, 10, 13)) return ok * 10 >= len(b) * 9 def _classify_word(raw: bytes): """Unknown 4-byte word -> (kind, value, alt).""" i = struct.unpack(" root Node.""" GUESS_ORDER = ("word", "bool", "string", "int64") def __init__(self, data: bytes, padding: str = "joint", schema=ROOT, kinds: Optional[dict] = None, shapes: Optional[dict] = None): if padding not in ("joint", "split"): raise ValueError("padding must be 'joint' or 'split'") self.d = data self.n = len(data) self.padding = padding self.schema = schema self.kinds = GLOBAL_KINDS if kinds is None else kinds self.shapes = GLOBAL_SHAPES if shapes is None else shapes self.issues: list[Issue] = [] self._eof_limited = False self.stats = {"items": 0, "frames": 0, "resyncs": 0, "hint_failures": 0, "guessed": 0, "raw_bytes": 0, "best_effort": 0} # -- low level --------------------------------------------------------------- def issue(self, level, off, msg, path=""): self.issues.append(Issue(level, path, off, msg)) def u32(self, p: int) -> int: return struct.unpack_from(" bool: return not any(self.d[a:b]) def tag_at(self, p: int, allow_empty: bool = False): """(name, end) if a plausible name tag starts at p, else None.""" if p + 4 > self.n: return None ln = struct.unpack_from(" MAX_TAG_LEN or p + 4 + ln > self.n: return None b = self.d[p + 4:p + 4 + ln] if not all(0x20 <= c < 0x7f for c in b): return None return b.decode("ascii"), p + 4 + ln def value_pos(self, tag_start: int, name_end: int) -> int: if self.padding == "joint": return name_end return tag_start + pad4(name_end - tag_start) def item_end(self, tag_start: int, name_end: int, size: int) -> int: if self.padding == "joint": return tag_start + pad4(name_end - tag_start + size) vp = self.value_pos(tag_start, name_end) return vp + pad4(size) def pads_zero(self, tag_start, name_end, vp, size, end) -> bool: if self.padding == "split" and not self.zero(name_end, vp): return False return self.zero(vp + size, end) def try_scalar(self, tag_start: int, name_end: int, kind: str, known: bool = False): """Read a candidate scalar; (kind, value, raw, end) or None if impossible. known=True means the kind comes from the schema/catalog: a string value is then accepted whatever bytes it holds (only tags must be ASCII).""" d, n = self.d, self.n vp = self.value_pos(tag_start, name_end) if kind == "string": if vp + 4 > n: self._eof_limited = True return None ln = struct.unpack_from(" MAX_STRING_LEN: return None if vp + 4 + ln > n: self._eof_limited = True return None raw = d[vp + 4:vp + 4 + ln] # ln == 0 (empty string) is legal if not known and not _text_plausible(raw): return None size, value = 4 + ln, raw.decode(ENCODING, errors="replace") else: size = 4 if kind == "word" else SIZES[kind] if vp + size > n: self._eof_limited = True return None raw = d[vp:vp + size] if kind == "bool": if raw[0] > 1: return None value = raw[0] == 1 elif kind == "int": value = struct.unpack(" n: self._eof_limited = True return None if not self.pads_zero(tag_start, name_end, vp, size, end): return None return kind, value, raw, end def plausible_item(self, p: int, depth: int, allow_empty: bool = False) -> bool: """Does an item plausibly start at p? Looks `depth` items ahead. Sets self._eof_limited when a verdict was forced by running out of data.""" n = self.n if p == n: return True if p + 4 > n: self._eof_limited = True return False w = self.u32(p) if w == END_MARK or w == BEGIN_MARK: return True t = self.tag_at(p, allow_empty) if t is None: return False name, ne = t a = pad4(ne) if a + 4 <= n and self.u32(a) == BEGIN_MARK and self.zero(ne, a): return True if depth <= 0: return True # a tag the catalog knows to be a string is read as one without any # text test, so a non-ASCII value (cp1252 0x92 in "Kor'Voth") can never # veto the layout of the item that precedes it if self.kinds.get(name) == "string": r = self.try_scalar(p, ne, "string", known=True) if r is not None and self.plausible_item(r[3], depth - 1, allow_empty): return True for kind in self.GUESS_ORDER: r = self.try_scalar(p, ne, kind) if r is None: continue if self.plausible_item(r[3], depth - 1, allow_empty): return True return False def resync(self, p: int, depth: int): """Scan forward for the next END marker or plausible tag. -> (what, q).""" d, n = self.d, self.n q = p + 1 while q + 4 <= n: w = self.u32(q) if w == END_MARK and depth > 0: return "end", q if w == BEGIN_MARK: return "begin", q if self.tag_at(q) is not None and self.plausible_item(q, 2, allow_empty=False): return "tag", q q += 1 return "eof", n # -- hints ------------------------------------------------------------------- def child_hint(self, frame_shape, index: int, name: Optional[str]): """-> (scalar kind or None, sub-shape or None) for child #index named name.""" kind = sub = None if isinstance(frame_shape, Shape): pre = frame_shape.prefix() if index < len(pre): kind, sub = _split_hint(pre[index]) if kind is None and sub is None and name: t = frame_shape.by_name().get(name) or frame_shape.by_name().get(name.lower()) if t is not None: kind, sub = _split_hint(t) elif isinstance(frame_shape, CArr): if index == 0: kind = "int" else: kind, sub = _split_hint(frame_shape.elem) elif frame_shape == "raw": kind = "raw" if kind is None and sub is None and name: # global catalog: exact case only (case-folding was found to # mislabel unknown names, e.g. 'a' vs star-colour 'A') k = self.kinds.get(name) if k: kind = k s = self.shapes.get(name) if s is not None: sub = s if name in RAW_FRAMES: sub = "raw" return kind, sub # -- walking ----------------------------------------------------------------- def walk(self) -> Node: children, end = self.walk_frame(0, None, 0, self.schema, "") root = Node(None, "complex", None, 0, end, children) return root def walk_frame(self, p: int, ctx, depth: int, shape, path: str): """Read items until the frame's END marker (or EOF). -> (children, pos).""" d, n = self.d, self.n children: list[Node] = [] index = 0 self.stats["frames"] += depth > 0 while True: if p >= n: if depth > 0: self.issue("error", p, f"frame {ctx!r} not terminated before EOF", path) return children, p if p + 4 > n: children.append(self._raw_node(None, p, n, "trailing bytes", "warn", path)) if depth > 0: self.issue("error", n, f"frame {ctx!r} not terminated before EOF", path) return children, n w = self.u32(p) if w == END_MARK: if depth > 0: return children, p + 4 self.issue("warn", p, "stray END marker at top level", path) children.append(self._raw_node(None, p, p + 4, "stray END marker", "warn", path)) p += 4 continue node, p = self.read_item(p, ctx, index, depth, shape, path) children.append(node) index += 1 def read_item(self, p: int, ctx, index: int, depth: int, frame_shape, path: str): d, n = self.d, self.n self.stats["items"] += 1 w = self.u32(p) if w == BEGIN_MARK: # tagless frame _, sub = self.child_hint(frame_shape, index, None) children, end = self.walk_frame(p + 4, None, depth + 1, sub, f"{path}/<{index}>") return Node(None, "complex", None, p, end - p, children), end t = self.tag_at(p, allow_empty=True) if t is None: return self._unreadable(p, depth, path) name, ne = t kind, sub = self.child_hint(frame_shape, index, name) a = pad4(ne) if a + 4 <= n and self.u32(a) == BEGIN_MARK and self.zero(ne, a): children, end = self.walk_frame(a + 4, name, depth + 1, sub, f"{path}/{name}") return Node(name, "complex", None, p, end - p, children, hinted=sub is not None), end # scalar: hinted kind first, then the guess order if kind == "raw": vp = self.value_pos(p, ne) q = d.find(END_BYTES, vp) if depth > 0 else -1 if q < 0: q = n node = Node(name, "raw", {"len": q - vp, "hex": d[vp:min(q, vp + 32)].hex()}, p, q - p, raw=d[vp:q], hinted=True) self.stats["raw_bytes"] += q - vp return node, q order = [] if kind in ("int", "float"): order.append("word") elif kind in ("bool", "string", "int64"): order.append(kind) order += [k for k in self.GUESS_ORDER if k not in order] # pass 1: continuation must be a non-empty tag / marker; pass 2 tolerates # empty ("") tags; pass 3 (only when lookahead was defeated by EOF) takes # whatever fits the remaining bytes. self._eof_limited = False chosen = None for mode in ("strict", "empty-ok", "eof"): if mode == "eof" and not self._eof_limited: break for i, cand in enumerate(order): r = self.try_scalar(p, ne, cand, known=(i == 0 and kind is not None)) if r is None: continue if mode == "eof" or self.plausible_item(r[3], 2, allow_empty=(mode == "empty-ok")): chosen = (i, r, mode) break if chosen: break if chosen: i, (ckind, value, raw, end), mode = chosen alt = None if ckind == "word": if kind in ("int", "float"): ckind = kind value = struct.unpack(" 0: q = p while q + 4 <= n and q - p <= SMALL_RAW: if self.u32(q) == END_MARK: return self._raw_node(None, p, q, "unnamed payload", "info", path), q q += 4 what, q = self.resync(p, depth) self.stats["resyncs"] += 1 return self._raw_node(None, p, q, f"unreadable; resynced to {what}", "warn", path), q def _raw_node(self, name, p, q, why, level, path) -> Node: d = self.d self.issue(level, p, f"{why}: {q - p} raw bytes", path) self.stats["raw_bytes"] += q - p return Node(name, "raw", {"len": q - p, "hex": d[p:min(q, p + 32)].hex()}, p, q - p, raw=d[p:q]) def _split_hint(t): if isinstance(t, str): if t in PRIMITIVES or t == "raw": return t, None return None, None # "vec3", "any" if isinstance(t, Field): return _split_hint(t.type) if isinstance(t, (Shape, CArr)): return None, t return None, None # --- generic tree helpers ----------------------------------------------------- def plain(node: Node): """Node -> JSON-able generic representation.""" if node.is_complex: return {"_name": node.name, "_off": node.offset, "_items": [plain(c) for c in node.children]} d = {"name": node.name, "kind": node.kind, "value": node.value, "off": node.offset} if node.alt is not None: d["alt"] = node.alt if not node.hinted and node.kind != "raw": d["guessed"] = True return d def dump_tree(node: Node, indent: int = 0, out=None, max_depth: int = 999) -> list[str]: lines = [] if out is None else out pad = " " * indent for c in node.children: nm = c.name if c.name is not None else "" if c.is_complex: lines.append(f"@{c.offset:08x} {pad}{nm} {{ # {len(c.children)} items, {c.size} bytes") if indent < max_depth: dump_tree(c, indent + 1, lines, max_depth) lines.append(f"@{c.offset + c.size - 4:08x} {pad}}}") elif c.kind == "raw": v = c.value lines.append(f"@{c.offset:08x} {pad}{nm} raw[{v['len']}] {v['hex']}{'...' if v['len'] > 32 else ''}") else: q = "" if c.hinted else "?" val = json.dumps(c.value) if isinstance(c.value, str) else c.value alt = "" if c.alt is not None and not c.hinted: alt = f" (alt {c.alt!r})" lines.append(f"@{c.offset:08x} {pad}{nm} {c.kind}{q} {val}{alt}") return lines # --- schema application ------------------------------------------------------- class _Cursor: __slots__ = ("nodes", "i") def __init__(self, nodes): self.nodes, self.i = nodes, 0 def peek(self): return self.nodes[self.i] if self.i < len(self.nodes) else None def take(self): n = self.nodes[self.i] self.i += 1 return n def find(self, name, ci=False): for j in range(self.i, len(self.nodes)): nm = self.nodes[j].name if nm == name or (ci and nm is not None and nm.lower() == name.lower()): return j return None @property def done(self): return self.i >= len(self.nodes) class Applier: def __init__(self): self.issues: list[Issue] = [] def issue(self, level, path, node, msg): off = node.offset if isinstance(node, Node) else (node or 0) self.issues.append(Issue(level, path, off, msg)) # -- entry ------------------------------------------------------------------- def apply(self, schema, root: Node) -> dict: out = {} cur = _Cursor(root.children) self.apply_fields(schema.fields if isinstance(schema, (Shape, Seq)) else [], cur, out, "") if not cur.done: out["_extra"] = [plain(n) for n in cur.nodes[cur.i:]] self.issue("warn", "", cur.peek(), f"{len(cur.nodes) - cur.i} unexpected trailing items") return out # -- fields ------------------------------------------------------------------ def apply_fields(self, fields, cur: _Cursor, out: dict, path: str): for f in fields: if isinstance(f, Field): self.apply_field(f, cur, out, path) elif isinstance(f, Opt): nxt = cur.peek() if nxt is not None and nxt.name is not None and nxt.name.lower() == f.field.name.lower(): self.apply_field(f.field, cur, out, path) elif isinstance(f, If): if f.holds(out): inner = f.inner if isinstance(inner, Seq): self.apply_fields(inner.fields, cur, out, path) else: self.apply_field(inner, cur, out, path) elif isinstance(f, Repeat): items = [] while not cur.done and cur.peek().name == f.lead: before = cur.i items.append(self.convert_elem(f.elem, cur, f"{path}/{f.key}[{len(items)}]")) if cur.i == before: break out[f.key] = items elif isinstance(f, Until): items = [] while not cur.done: nxt = cur.peek() if nxt.name is not None and nxt.name.lower() == f.stop.lower(): break items.append(plain(cur.take())) out[f.key] = items elif isinstance(f, Rest): if not cur.done: out[f.key] = [plain(n) for n in cur.nodes[cur.i:]] cur.i = len(cur.nodes) else: raise TypeError(f"bad schema element {f!r}") def apply_field(self, f: Field, cur: _Cursor, out: dict, path: str): fpath = f"{path}/{f.key}" t = f.type if isinstance(t, NArr): cnt_node = self.take_named(f, cur, out, fpath) if cnt_node is None: return count = self.coerce(cnt_node, "int", fpath) items = [] remaining = len(cur.nodes) - cur.i if not isinstance(count, int) or count < 0 or count > remaining: self.issue("error", fpath, cnt_node, f"array count {count!r} exceeds the {remaining} item(s) left in the frame") count = max(0, min(count if isinstance(count, int) else 0, remaining)) for i in range(count): if cur.done: self.issue("error", fpath, cur.nodes[-1] if cur.nodes else None, f"array truncated: {i} of {count} elements present") break before = cur.i items.append(self.convert_elem(t.elem, cur, f"{fpath}[{i}]")) if cur.i == before: self.issue("error", fpath, cur.peek(), f"array element {i} consumed nothing; stopping") break out[f.key] = items return if isinstance(t, Seq): self.apply_fields(t.fields, cur, out, path) return if isinstance(t, (Shape, CArr)) and f.flex: nxt = cur.peek() if nxt is not None and not nxt.is_complex: # inline (unframed) variant if isinstance(t, Shape): sub = {} self.apply_fields(t.fields, cur, sub, fpath) out[f.key] = sub else: self.apply_field(Field(f.name, NArr(t.elem), f.key, f.auth), cur, out, path) return node = self.take_named(f, cur, out, fpath) if node is None: return out[f.key] = self.convert(node, t, fpath) def take_named(self, f: Field, cur: _Cursor, out: dict, fpath: str): node = cur.peek() if node is None: self.issue("error" if f.auth else "warn", fpath, None, f"missing field {f.name!r} (frame ended)") out[f.key] = None return None if f.auth: if node.name != f.name: j = cur.find(f.name) if j is None: self.issue("error", fpath, node, f"expected {f.name!r}, found {node.name!r}; field missing") out[f.key] = None return None skipped = cur.nodes[cur.i:j] out.setdefault("_unexpected", []).extend(plain(n) for n in skipped) self.issue("warn", fpath, node, f"{len(skipped)} unexpected item(s) before {f.name!r}: " + ", ".join(repr(n.name) for n in skipped[:6])) cur.i = j node = cur.peek() elif node.name is not None and node.name.lower() != f.name.lower(): self.issue("info", fpath, node, f"tag {node.name!r} read positionally as {f.name!r}") return cur.take() # -- conversion -------------------------------------------------------------- def convert_elem(self, elem, cur: _Cursor, path: str): if isinstance(elem, Seq): sub = {} self.apply_fields(elem.fields, cur, sub, path) return sub if isinstance(elem, Field): sub = {} self.apply_field(elem, cur, sub, path) return sub.get(elem.key) node = cur.take() return self.convert(node, elem, path) def convert(self, node: Node, t, path: str): if isinstance(t, str): if t in PRIMITIVES: if node.is_complex: self.issue("error", path, node, f"expected {t}, found frame {node.name!r}") return plain(node) return self.coerce(node, t, path) if t == "vec3": return self.vec3(node, path) return plain(node) # "any" / "raw" if isinstance(t, Shape): if not node.is_complex: self.issue("error", path, node, f"expected frame, found {node.kind} {node.name!r}") return plain(node) sub = {"_off": node.offset} cur = _Cursor(node.children) self.apply_fields(t.fields, cur, sub, path) if not cur.done: sub["_extra"] = [plain(n) for n in cur.nodes[cur.i:]] self.issue("warn", path, cur.peek(), f"{len(cur.nodes) - cur.i} unexpected item(s) at end of frame") return sub if isinstance(t, CArr): if not node.is_complex: self.issue("error", path, node, f"expected framed array, found {node.kind} {node.name!r}") return plain(node) cur = _Cursor(node.children) if cur.done: return [] count = self.coerce(cur.take(), "int", path) items = [] remaining = len(cur.nodes) - cur.i if not isinstance(count, int) or count < 0 or count > remaining: self.issue("error", path, node, f"framed-array count {count!r} exceeds {remaining} item(s)") count = max(0, min(count if isinstance(count, int) else 0, remaining)) for i in range(count): if cur.done: self.issue("error", path, node, f"framed array truncated: {i} of {count}") break before = cur.i items.append(self.convert_elem(t.elem, cur, f"{path}[{i}]")) if cur.i == before: self.issue("error", path, cur.peek(), f"array element {i} consumed nothing; stopping") break if not cur.done: self.issue("warn", path, cur.peek(), f"{len(cur.nodes) - cur.i} item(s) after framed array elements") items.append({"_extra": [plain(n) for n in cur.nodes[cur.i:]]}) return items if isinstance(t, Seq): raise TypeError("Seq cannot be converted from a single node") raise TypeError(f"unknown field type {t!r}") def coerce(self, node: Node, kind: str, path: str): k, raw = node.kind, node.raw if k == kind: return node.value if kind == "int": if k in ("float",) and len(raw) == 4: return struct.unpack(" 1: self.issue("warn", path, node, f"bool expected, word {raw.hex()} read") return raw[0] != 0 elif kind == "int64": if k in ("int", "float") and len(raw) == 4: self.issue("warn", path, node, "int64 expected, 4-byte word read (width mismatch)") return struct.unpack(" (typed dict, issues).""" ap = Applier() return ap.apply(schema, root), ap.issues # --- top-level API ------------------------------------------------------------ class SaveResult: def __init__(self, inflated, padding, tree, typed, issues, stats): self.inflated = inflated self.padding = padding self.tree = tree self.typed = typed self.issues = issues self.stats = stats def count(self, level: str) -> int: return sum(1 for i in self.issues if i.level == level) def _walk_mode(data: bytes, padding: str, schema): w = Walker(data, padding, schema) tree = w.walk() return w, tree def read_bytes(data: bytes, padding: str = "auto", strict: bool = False, schema=ROOT) -> SaveResult: """Parse a .sav (or already-inflated) byte string.""" inflated = inflate(data) if len(inflated) < 8: raise SaveFormatError("stream too short to be a save", 0) if padding == "auto": best = None for mode in ("joint", "split"): w, tree = _walk_mode(inflated, mode, schema) score = (w.stats["resyncs"], w.stats["hint_failures"], w.stats["raw_bytes"], sum(1 for i in w.issues if i.level != "info")) if best is None or score < best[0]: best = (score, mode, w, tree) _, padding, w, tree = best else: w, tree = _walk_mode(inflated, padding, schema) typed, sissues = apply_schema(tree, schema) issues = w.issues + sissues res = SaveResult(inflated, padding, tree, typed, issues, dict(w.stats)) if strict: bad = [i for i in issues if i.level in ("error", "warn")] if bad: head = "\n ".join(repr(i) for i in bad[:20]) raise SaveFormatError(f"strict: {len(bad)} schema/framing problem(s):\n {head}", bad[0].offset) return res def read_save(path: str, **kw) -> SaveResult: with open(path, "rb") as f: return read_bytes(f.read(), **kw) # --- CLI ------------------------------------------------------------------------ def _summary_lines(res: SaveResult, path: str) -> list[str]: t = res.typed st = res.stats lines = [f"file: {path} inflated: {len(res.inflated)} bytes padding: {res.padding}", f"items: {st['items']} frames: {st['frames']} guessed: {st['guessed']} " f"hint-failures: {st['hint_failures']} resyncs: {st['resyncs']} raw-bytes: {st['raw_bytes']}", f"issues: {res.count('error')} error, {res.count('warn')} warn, {res.count('info')} info"] s = t.get("summary") or {} if isinstance(s, dict): lines.append(f"summary: game={s.get('GameName')!r} turn={s.get('Turn')} numSys={s.get('NumSys')} " f"players={len(s.get('Players') or [])} scenario={s.get('Scenario')!r}") sim = t.get("sim") or {} if isinstance(sim, dict): lines.append(f"sim: gameName={sim.get('GameName')!r} players={len(sim.get('players') or [])} " f"systems={len(sim.get('systems') or [])} fleets={len(sim.get('fleets') or [])}") for p in sim.get("players") or []: pl = p.get("Player") if isinstance(p, dict) else None if isinstance(pl, dict): lines.append(f" player {p.get('PlayerID')}: {pl.get('PlryName')!r} species={pl.get('Species')} " f"home={pl.get('HomeSys')} sav={pl.get('Sav')} designs={len(pl.get('designs') or [])}") return lines def main(argv=None) -> int: ap = argparse.ArgumentParser(description="Sword of the Stars (2006) save reader") ap.add_argument("save", help=".sav (gzip) or already-inflated stream") ap.add_argument("--dump", action="store_true", help="print the generic tree (text, or JSON with --json)") ap.add_argument("--json", action="store_true", help="emit JSON (typed view, or tree with --dump)") ap.add_argument("--strict", action="store_true", help="fail on any schema/framing deviation") ap.add_argument("--padding", choices=("auto", "joint", "split"), default="auto") ap.add_argument("--out", help="write output here instead of stdout") ap.add_argument("--inflate", metavar="FILE", help="also write the inflated stream to FILE") ap.add_argument("--issues", type=int, default=30, help="max issues to print (default 30)") ap.add_argument("--max-depth", type=int, default=999, help="tree dump depth limit") args = ap.parse_args(argv) try: res = read_save(args.save, padding=args.padding, strict=args.strict) except SaveFormatError as e: print(f"error: {e}", file=sys.stderr) return 2 if args.inflate: with open(args.inflate, "wb") as f: f.write(res.inflated) if args.dump and args.json: text = json.dumps({"padding": res.padding, "stats": res.stats, "issues": [i.as_dict() for i in res.issues], "tree": plain(res.tree)["_items"]}, indent=1) elif args.dump: text = "\n".join(dump_tree(res.tree, max_depth=args.max_depth)) elif args.json: text = json.dumps({"padding": res.padding, "stats": res.stats, "issues": [i.as_dict() for i in res.issues], "data": res.typed}, indent=1) else: lines = _summary_lines(res, args.save) shown = [i for i in res.issues if i.level != "info"][:args.issues] if shown: lines.append("issues (errors/warnings, first %d):" % args.issues) lines += [" " + repr(i) for i in shown] text = "\n".join(lines) if args.out: with open(args.out, "w", encoding="utf-8") as f: f.write(text + "\n") else: sys.stdout.write(text + "\n") return 1 if res.count("error") else 0 if __name__ == "__main__": sys.exit(main())