a4aba6a9fb
Blocker #4 on lane Y's path to a byte-matching turn. HOW DESIGNS PERSIST, correcting a published finding. Game::ShipDesign::Write is 0x008325e0 and makes four stream calls. The recorded address 0x008747a0 is in NO vftable at all (lane V2's inversion), so "ShipDesign::Write makes no stream call" was a misattribution, not a fact about the class. Game::ShipDesign derives from Game::ShipDesignDef and inherits IStreamable second, so its writer is reached through an adjustor thunk -- which is what the slot-indexed serializer sweep found instead. A design persists as two serializers, base and derived: ShipDesignDef::Write 0x00827390 emits FAIDes/DHide/DWep/DName then exactly three DSec frames (+0x4c command, +0x24 mission, +0x74 engine), ShipDesign::Write appends Dtc, the Dwgv flag and a conditional Dwg frame. THREE sections, not five. The campaign's "slots 3-4 reserved and always empty" was save_reader.py's trailing Rest("sections") sweeping Dtc and Dwgv into the section list, and stock_designs.py decoding them as two empty sections -- rule 8 in its exact form, reader and consumer agreeing with each other and both wrong. Two independent enumerations say three: the writer's straight-line body, and the ctor's eh_vector_constructor_iterator(this+0x24, 0x28, 3) closing at 0x9c = sizeof(ShipDesignDef). DWep and Dwgv are BOOLs, not ints -- byte-indistinguishable from ints at a four-character tag, the same class of defect as ObservedTech.odet. THE 0x400 FLAG IS `defence_platform`, read off the .shipsection parser's own bit setter at 0x005749b7. NOT lane B5's 0x400: that one is a fleet flag, on the wire as FtFlg. The full role-flag table is in the finding. HULL SIZE is section_class through a three-name stricmp table (Destroyer/Cruiser/Dreadnought -> 0/1/2), absent or unrecognised meaning 0 with a log line rather than an error. Both words are recomputed from the data files by ShipDesign::UpdateDerivedStats 0x0087e7c0 and neither is on the wire. Corroborated by the default hull-health table the same bit picks: 500/3000/15000 without it, 100/500/1000 with. MEASURED: the census rebuilt from each save's own state matches the record the game archived, 480 leaves / 0 mismatched over 11 saves and 503 designs, computed independently in Python and in C++. COVERAGE REPORTED AS LOUDLY: only 32 of the 480 leaves are nonzero, and three of the six census leaves (both cruiser rows, dreadnought platforms) are unexercised by every save in the corpus. Closed 0 / regressed 0 against the standalone's divergence list, reported separately: the census leaves live in src/app's turn record, which lane A2 holds this cycle, so this lane evaluated and reported rather than writing. Oracles fixed openly (rule 12): save_reader.py's Des shape, 49/49 with three corrected tests and one added that pins "exactly three DSec" against real saves; stock_designs.json regenerated, whose diff is only raw_slots 5->3 and dWep int->bool across all 127 designs with every other field identical; test_design_rules.py still 32/32 with the same ground truth. 19 addresses in ghidra/addresses.d/lane-d2.json, no collision; the generated header was validated to a scratch path, never written in place.
1612 lines
73 KiB
Python
1612 lines
73 KiB
Python
#!/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("<I", BEGIN_MARK)
|
|
END_BYTES = struct.pack("<I", END_MARK)
|
|
ENCODING = "cp1252"
|
|
MAX_TAG_LEN = 64 # field names are short identifiers
|
|
MAX_STRING_LEN = 1 << 20 # sanity bound for string payloads
|
|
SMALL_RAW = 64 # unnamed payload <= this before an END is "info" not "warn"
|
|
SIZES = {"int": 4, "float": 4, "bool": 1, "int64": 8}
|
|
PRIMITIVES = ("int", "float", "bool", "int64", "string")
|
|
WORD_INT_ABS = 100_000 # |int| <= this -> 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<SystemParams>
|
|
R("players", NArr(CArr("int"))), # "." count, n x "." frame{ "." count, n x "." int }
|
|
R("nodePaths", CArr("any")), # VectorHelper<SimpleNodePath>, 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.
|
|
#
|
|
# TWO writers, a base and a derived one (lane D2): the base emits the four
|
|
# scalars then EXACTLY THREE `DSec` frames -- command, mission, engine, in that
|
|
# order -- and the derived one appends `Dtc`, the `Dwgv` flag and, only when
|
|
# that flag is set, the `Dwg` weapon-group frame.
|
|
#
|
|
# `DWep` and `Dwgv` are BOOLs, not ints: both writers call the bool primitive.
|
|
# With four-character tags a bool item and an int item are the same size on the
|
|
# wire and the values 0/1 are the same bytes, so no save can tell them apart --
|
|
# this is the same class of defect as `ObservedTech.odet`, and it is byte-neutral.
|
|
#
|
|
# CORRECTION (lane D2): this shape used to end in `Rest("sections")`, which swept
|
|
# `Dtc` and `Dwgv` into the section list. That is where the campaign's "five
|
|
# section slots, of which slots 3 and 4 are reserved and always empty" claim came
|
|
# from -- there are three sections and no reserved slots.
|
|
Design = Shape("Des", [
|
|
A("FAIDes", "bool", key="faiDes"), A("DHide", "bool", key="dHide"),
|
|
A("DWep", "bool", key="dWep"), A("DName", "string", key="dName"),
|
|
Repeat(A("DSec", "any"), key="sections"),
|
|
A("Dtc", "int", key="dtc"), A("Dwgv", "bool", key="dwgv"),
|
|
If("dwgv", A("Dwg", "any", key="weaponGroups")), # never set in any save available
|
|
Rest(),
|
|
])
|
|
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<uint>
|
|
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.<id>.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("<i", raw)[0]
|
|
f = struct.unpack("<f", raw)[0]
|
|
if -WORD_INT_ABS <= i <= WORD_INT_ABS:
|
|
return "int", i, f if math.isfinite(f) else None
|
|
if math.isfinite(f) and 1e-6 <= abs(f) < 1e12:
|
|
return "float", f, i
|
|
return "int", i, f if math.isfinite(f) else None
|
|
|
|
|
|
class Walker:
|
|
"""Generic tokenizer over the inflated stream. walk() -> 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("<I", self.d, p)[0]
|
|
|
|
def zero(self, a: int, b: int) -> 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("<i", self.d, p)[0]
|
|
if ln == 0:
|
|
return ("", p + 4) if allow_empty else None
|
|
if ln < 0 or ln > 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("<i", d, vp)[0]
|
|
if ln < 0 or ln > 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("<i", raw)[0]
|
|
elif kind == "float":
|
|
value = struct.unpack("<f", raw)[0]
|
|
elif kind == "int64":
|
|
value = struct.unpack("<q", raw)[0]
|
|
else:
|
|
value = None
|
|
end = self.item_end(tag_start, name_end, size)
|
|
if end > 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("<i" if kind == "int" else "<f", raw)[0]
|
|
alt = struct.unpack("<f" if kind == "int" else "<i", raw)[0]
|
|
if isinstance(alt, float) and not math.isfinite(alt):
|
|
alt = None
|
|
else:
|
|
ckind, value, alt = _classify_word(raw)
|
|
hinted = kind is not None and i == 0
|
|
if kind is not None and i != 0:
|
|
self.stats["hint_failures"] += 1
|
|
self.issue("warn", p, f"{name!r}: hinted type {kind} not plausible, read as {ckind}", path)
|
|
elif kind is None:
|
|
self.stats["guessed"] += 1
|
|
if mode == "eof":
|
|
self.stats["best_effort"] += 1
|
|
self.issue("warn", p, f"{name!r}: stream ends before layout can be confirmed; read as {ckind}", path)
|
|
return Node(name, ckind, value, p, end - p, raw=raw, alt=alt, hinted=hinted), end
|
|
|
|
# tag looked fine but no value layout fits: skip to the next sync point
|
|
what, q = self.resync(ne, depth)
|
|
self.stats["resyncs"] += 1
|
|
self.issue("warn", p, f"{name!r}: no value layout fits; skipped {q - ne} bytes to {what}", path)
|
|
node = Node(name, "raw", {"len": q - ne, "hex": d[ne:min(q, ne + 32)].hex()}, p, q - p, raw=d[ne:q])
|
|
self.stats["raw_bytes"] += q - ne
|
|
return node, q
|
|
|
|
def _unreadable(self, p: int, depth: int, path: str):
|
|
"""No tag at p. Small unnamed payload before END (e.g. Vector3's three
|
|
floats) is normal; anything longer is a resync event."""
|
|
d, n = self.d, self.n
|
|
if depth > 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 "<tagless>"
|
|
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("<i", raw)[0]
|
|
if k == "bool":
|
|
return int(node.value)
|
|
if k == "int64":
|
|
self.issue("warn", path, node, "int expected, int64 read")
|
|
return node.value
|
|
elif kind == "float":
|
|
if k in ("int",) and len(raw) == 4:
|
|
return struct.unpack("<f", raw)[0]
|
|
if k == "bool":
|
|
return float(node.value)
|
|
elif kind == "bool":
|
|
if k in ("int", "float") and len(raw) == 4:
|
|
if raw[1:4] != b"\0\0\0" or raw[0] > 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("<i", raw)[0]
|
|
elif kind == "string":
|
|
# an empty string is "len 0" = 4 zero bytes, byte-identical to int 0;
|
|
# an unhinted walker reads it as a word -- that IS the empty string
|
|
if k in ("int", "float") and raw == b"\0\0\0\0":
|
|
return ""
|
|
self.issue("error", path, node, f"expected {kind}, read {k}")
|
|
return node.value
|
|
|
|
def vec3(self, node: Node, path: str):
|
|
if node.is_complex:
|
|
ch = node.children
|
|
if len(ch) == 1 and ch[0].kind == "raw" and len(ch[0].raw) == 12:
|
|
return list(struct.unpack("<3f", ch[0].raw))
|
|
if len(ch) == 3 and all(len(c.raw) == 4 and not c.is_complex for c in ch):
|
|
return [struct.unpack("<f", c.raw)[0] for c in ch]
|
|
self.issue("error", path, node, f"vec3 frame has unexpected body ({len(ch)} items)")
|
|
return plain(node)
|
|
if node.kind == "raw" and len(node.raw) == 12:
|
|
return list(struct.unpack("<3f", node.raw))
|
|
self.issue("error", path, node, f"expected vec3, found {node.kind}")
|
|
return plain(node)
|
|
|
|
|
|
def apply_schema(root: Node, schema=ROOT):
|
|
"""-> (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())
|