#!/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, CreateParameters, Sim -> Player / Sys / Flt / Ship ...) applied on top of the generic tree to produce typed dicts. Fields marked A() carry binary-confirmed on-disk names (struct-recovery.md) and are matched by name; fields marked R() carry only the community editor's C# name and are matched by position. Padding convention: the community reference is ambiguous about whether the NUL padding is computed over the whole item ("joint": [len][name][value][pad]) or separately after the name and after the value ("split"). Both are implemented; --padding auto (default) walks the file both ways and keeps the cleaner parse. REAL-SAVE VALIDATION IS PENDING -- see SAVE_FORMAT.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", "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).""" __slots__ = ("key", "inner", "equals") def __init__(self, key, inner, equals=True): self.key, self.inner, self.equals = key, inner, 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 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) # --- 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). NestedInt = Shape(None, [R("value", "int"), Rest()]) PlayerColor = Shape("ClrID", [ R("idx", "int"), If("idx", Seq([R("r", "int"), R("g", "int"), R("b", "int")]), equals=-1), Rest(), ]) # -- Summary (R1 §3) ----------------------------------------------------------- PlayerSettings = Shape("Settings", [ R("initialTreasury", "int"), R("initialColonies", "int"), R("initialTechnologies", "int"), R("difficulty", "int"), Rest(), ]) PlayerSlot = Shape("Slot", [ R("isPlay", "bool"), R("isDead", "bool"), R("isReq", "bool"), R("isRec", "bool"), R("isFxNm", "bool"), R("fxNm", "string"), R("isFxSp", "bool"), R("fxSp", "int"), R("isFxCr", "bool"), R("fxCrId", PlayerColor), R("isFxBd", "bool"), R("fxBd", "string"), R("isFxAv", "bool"), R("fxAv", "string"), R("tag", "int"), R("pwd", "int"), R("team", "int"), R("settings", PlayerSettings), Rest(), ]) PlayerSlotWrapper = Shape("PlayerSlotWrapper", [R("slot", PlayerSlot), R("rank", "int"), Rest()]) Tmrs = Shape("Tmrs", [R("tstl", "int"), R("tctl", "int"), R("tqtl", "int"), R("tqtle", "int"), Rest()]) Session = Shape("Session", [R("tmrs", Tmrs), Rest()]) Summary = Shape("Summary", [ R("gameName", "string"), R("turn", "int"), R("numSys", "int"), R("checkSum", "int"), R("players", CArr(PlayerSlotWrapper)), R("session", Session), R("mapShape", "int"), R("incMod", "int"), R("resMod", "int"), R("alliances", "bool"), R("teams", "bool"), R("encounters", "bool"), R("scenario", "string"), Rest(), ]) # -- CreateParameters (R1 §4) --------------------------------------------------- Planet = Shape("Planet", [ R("coordinates", "vec3"), R("unknown1", "int"), R("unknown2", "int"), R("unknown3", "int"), R("unknown4", "int"), Rest(), ]) MapPNpc = Shape("Npc", [R("unknown1", "int"), R("unknown2", "int"), Rest()]) MapP = Shape("MapP", [ R("unknown1", "int"), R("planetArray", CArr(Planet)), R("players", NArr(CArr("int"))), R("npcArray", CArr(MapPNpc)), Rest(), ]) Scrp = Shape("Scrp", [R("spc", "int"), Rest()]) CreateParams = Shape("CreateParameters", [ R("name", "string"), R("id", "int"), R("rSeed", "int"), R("aid", "int"), R("key", "string"), R("mapP", MapP), R("mapS", "int"), R("mapF", "int"), R("nSys", "int"), R("rEnc", "float"), R("sDist", "int"), R("sSize", "float"), R("sRes", "float"), R("sSuit", "int"), R("maxP", "int"), R("aSpec", "int"), R("bAlly", "bool"), R("nTeam", "int"), R("tmgrp", "bool"), R("pSav", "int"), R("pCol", "int"), R("pTech", "int"), R("incM", "float"), R("resM", "float"), R("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", [R("ords", CArr(BuildOrder), flex=True), Rest()]) 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"), A("NVO", NArr(Seq([ A("PID", "int"), A("TShn", "int"), A("OID", "int"), A("isind", "bool"), If("isind", 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(), ]) Odes = Shape(None, [R("ontF", "int"), R("otnL", "int"), R("odid", "int"), R("opid", "int"), Rest()]) Owep = Shape(None, [R("ontF", "int"), R("otnL", "int"), R("odet", "int"), R("owep", "string"), R("owith", "int"), Rest()]) Otch = Shape(None, [R("ontF", "int"), R("otnL", "int"), R("odet", "int"), R("otch", "string"), R("owith", "int"), Rest()]) Note = Shape("Nts", [A("NtSys", "int"), A("NtTxt", "string"), A("NtTrn", "int"), Rest()]) Design = Shape("Des", [ R("faiDes", "bool"), R("dHide", "bool"), R("dWep", "int"), R("dName", "string"), 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", "any"), A("NextPrjID", "int"), A("plcy", "int"), A("pswd", "string"), A("lret", "int"), A("nmeid", "int"), A("cdp", "bool"), A("spy2", "any"), A("civr", "any"), 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", [ R("wpts", CArr(Waypoint), flex=True), A("FPsp2", "float"), A("FPeta2", "int"), A("FPogn2", "vec3"), A("FPdpos", "vec3"), A("pnd", "int"), Rest(), ]) PrisonerHold = Shape("PrisH", [ A("PrMax", "int"), A("PrNSp", NArr(Seq([A("PrSp", "int"), A("PrNum", "int")])), key="prisoners"), 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(), ]) NodeGrid = Shape("NdGr2", [R("paths", CArr(NodePath), flex=True), R("nextId", "int"), Rest()]) # -- Sim block (struct-recovery §5 tag order + R1 §5) ---------------------------- Sim = Shape("Sim", [ A("KeyPath", "string"), A("NMSz", "int"), A("NMLc", "int"), A("NMnx", "int"), Until("idLists", "ModCount"), # PlayerIDs/DesignIDs/SystemIDs/FleetIDs/ShipIDs/TradeIDs A("ModCount", "int"), A("Frame", "int"), A("GameID", "int"), Opt("AIDifficultyID", "int"), A("Attrib", "any"), A("RNG", "any"), A("GameName", "string"), A("Map", "int"), A("IncMod", "int"), A("ResMod", "int"), A("EnAl", "int"), A("EnTm", "int"), A("GOTurn", "int"), A("GOWinPly", NestedInt), A("NPCm", "int"), A("NPCo", "int"), A("NPCi", "int"), A("NPCv", "int"), A("NPCa", "int"), A("szadj", "int"), A("rsadj", "int"), A("suadj", "int"), A("sprjs", "any"), A("RandEncAdj", "float"), A("cmbtid", "int"), A("turnstats", "any"), A("numcreps", NArr(A("crep", Crep)), key="combatReports"), A("ninv", "int"), Until("invasionsAndExclusions", "NumPlrs"), # invs/inve/invt/invtb, AllExc x6, AllExcCF, AllExcCFp x2 A("NumPlrs", NArr(Seq([A("PlayerID", "int"), A("Player", Player)])), key="players"), Until("species", "NumSys"), # ISsp / ISsu 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"), A("SvSctOb", "any"), A("zdsc", "int"), A("zdsi", "int"), A("zdst", "int"), Rest(), ]) # -- file root: Summary, CreateParameters, Sim, then the unframed CdTable --------- ROOT = Seq([ R("summary", Summary), R("createParams", CreateParams), R("sim", Sim), Rest("cdTable"), ]) # 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", "tiAcq": "int", "tUnlck": "int", "dName": "string", "wfn": "string", "bId": "bool", "faiDes": "bool", "dHide": "bool", # objectives / comms / research / encounters (R1 §5, §6, §11) "cmp": "bool", "dsc": "string", "xcsn": "string", "nm": "string", "ntg": "string", "tch": "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): 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 = [f] if isinstance(f, Opt): items = [f.field] elif isinstance(f, If): items = f.inner.fields if isinstance(f.inner, Seq) else [f.inner] 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) def _register_type(name, t, kinds, conflicts, shapes, seen): if isinstance(t, str): if t in PRIMITIVES: 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) elif isinstance(t, Seq): _register(t, kinds, conflicts, shapes, seen) elif isinstance(t, (CArr, NArr)): e = t.elem if isinstance(e, Field): _register_type(e.name, e.type, kinds, conflicts, shapes, seen) elif isinstance(e, (Shape, Seq)): _register(e, kinds, conflicts, shapes, seen) 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) 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 ----------------------------------------------------------- def _text_plausible(b: bytes) -> bool: if not b: return True ok = sum(1 for c in b if 0x20 <= c < 0x7f or c >= 0xa0 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): """Read a candidate scalar; (kind, value, raw, end) or None if impossible.""" 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] if not _text_plausible(raw): return None size, value = 4 + ln, raw.decode(ENCODING) 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 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) 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 out.get(f.key) == f.equals: 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, 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())