433 lines
17 KiB
Python
433 lines
17 KiB
Python
"""Cross-binary **compatibility table**, grouped by library (DLL) name.
|
||
|
||
A documentation aid: take a set of catalogued snapshots and lay out, per engine library
|
||
(``PIKLIB8.dll``, ``bloomoodll.dll`` …), which games ship it. The catch the user cares about:
|
||
the *same* library name can appear in several games yet not actually be the same engine —
|
||
``PIKLIB8.dll`` in game A and game B may diverge. So within each library name we **cluster
|
||
variants by engine surface** (the compiler-agnostic identity sets that `ams.similarity` /
|
||
`ams.diff` already key on) and flag every non-reference variant with an **asterisk**, carrying a
|
||
``vs_reference`` delta so the footnote can say *what* differs.
|
||
|
||
The builder is pure (lists of snapshots in, JSON-able dict out); `render_markdown` /
|
||
`render_json` turn that into the downloadable file. Both the CLI (`python -m ams.compat`) and the
|
||
`/compat` API endpoint sit on top of this.
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import hashlib
|
||
import json
|
||
import re
|
||
from collections import Counter
|
||
from dataclasses import dataclass
|
||
from datetime import datetime, timezone
|
||
from typing import Any, Iterable
|
||
|
||
from .diff import compute_diff
|
||
from .similarity import _AXES as _SIM_AXES # axis -> (accessor, identity-key fn)
|
||
from .similarity import _keys, surface_similarity
|
||
from .snapshot import Snapshot
|
||
|
||
# The four script-facing axes we cluster on (same set similarity scores over).
|
||
_AXES = list(_SIM_AXES)
|
||
|
||
|
||
@dataclass
|
||
class Entry:
|
||
"""One catalogued binary: the column label (game) plus its surface snapshot."""
|
||
|
||
game: str
|
||
snapshot: Snapshot
|
||
|
||
|
||
# --- surface signature -------------------------------------------------------------------------
|
||
def surface_signature(snap: Snapshot) -> str:
|
||
"""A stable hash of the engine surface: the sorted identity sets of every axis.
|
||
|
||
Two snapshots with the same signature expose the same script-facing API — i.e. they are
|
||
*compatible* even across recompiles/compilers, so they collapse into one variant. Any
|
||
difference in the sets (a method added, a field retyped, a type gone) yields a new signature
|
||
and thus an asterisk."""
|
||
payload = {axis: sorted(list(k) for k in _keys(snap, axis)) for axis in _AXES}
|
||
blob = json.dumps(payload, sort_keys=True, ensure_ascii=False).encode("utf-8")
|
||
return hashlib.sha1(blob).hexdigest()
|
||
|
||
|
||
def _surface_counts(snap: Snapshot) -> dict[str, int]:
|
||
"""Identity-set size per axis (mirrors the signature's notion of 'how big is the surface')."""
|
||
return {axis: len(_keys(snap, axis)) for axis in _AXES}
|
||
|
||
|
||
def _axis_counts(block: dict[str, Any]) -> dict[str, int]:
|
||
return {
|
||
"added": len(block["added"]),
|
||
"removed": len(block["removed"]),
|
||
"changed": len(block["changed"]),
|
||
}
|
||
|
||
|
||
def _changes_str(changes: dict[str, list]) -> str:
|
||
return ", ".join("{0} {1}→{2}".format(f, v[0], v[1]) for f, v in sorted(changes.items()))
|
||
|
||
|
||
def _types_named(block: dict[str, Any]) -> dict[str, list[str]]:
|
||
"""Type-axis changes as display strings (types are global, not owner-scoped)."""
|
||
return {
|
||
"added": sorted("{0} → {1} (size {2})".format(
|
||
t["script_name"], t.get("cpp_class") or "?", t.get("object_size")) for t in block["added"]),
|
||
"removed": sorted(t["script_name"] for t in block["removed"]),
|
||
"changed": sorted("{0} ({1})".format(c["item"]["script_name"], _changes_str(c["changes"]))
|
||
for c in block["changed"]),
|
||
}
|
||
|
||
|
||
def _owner_named(block: dict[str, Any]) -> dict[str, Any]:
|
||
"""Owner-scoped axis (methods/events/fields) grouped by class, so the diff reads as
|
||
'which class gained/lost/changed which member' — the bit that was missing from the counts."""
|
||
by: dict[str, dict[str, list[str]]] = {}
|
||
|
||
def slot(owner: str) -> dict[str, list[str]]:
|
||
return by.setdefault(owner or "?", {"added": [], "removed": [], "changed": []})
|
||
|
||
for it in block["added"]:
|
||
slot(it["owner"])["added"].append(it["name"])
|
||
for it in block["removed"]:
|
||
slot(it["owner"])["removed"].append(it["name"])
|
||
for c in block["changed"]:
|
||
it = c["item"]
|
||
slot(it["owner"])["changed"].append("{0} ({1})".format(it["name"], _changes_str(c["changes"])))
|
||
|
||
return {"by_owner": {o: {k: sorted(v) for k, v in d.items()} for o, d in sorted(by.items())}}
|
||
|
||
|
||
def _vs_reference(reference: Snapshot, variant: Snapshot) -> dict[str, Any]:
|
||
"""How a divergent variant differs from its library's reference variant: per-axis +/-/~ counts,
|
||
the pooled surface-overlap score, and a `detail` block naming *which* classes/methods/events/
|
||
fields changed (so the table can say 'CMC_Animo lost SHOW', not just 'methods −1')."""
|
||
diff = compute_diff(reference, variant)
|
||
return {
|
||
"overall": surface_similarity(reference, variant)["overall"],
|
||
"axes": {axis: _axis_counts(diff[axis]) for axis in _AXES},
|
||
"moved_methods": len(diff["moved_methods"]),
|
||
"detail": {
|
||
"types": _types_named(diff["types"]),
|
||
"methods": _owner_named(diff["methods"]),
|
||
"events": _owner_named(diff["events"]),
|
||
"fields": _owner_named(diff["fields"]),
|
||
},
|
||
}
|
||
|
||
|
||
# --- table builder -----------------------------------------------------------------------------
|
||
# Stored binaries are prefixed with the first 8 hex of their sha256 to keep filenames unique on
|
||
# disk (e.g. ``333ab0c4_PIKLIB8.dll``). That prefix is per-file, so it must be stripped before
|
||
# grouping — otherwise every copy of PIKLIB8.dll lands in its own row and nothing ever clusters.
|
||
_HASH_PREFIX = re.compile(r"^[0-9a-fA-F]{8}_")
|
||
|
||
|
||
def library_name(raw: str) -> str:
|
||
"""The display/grouping name: the stored binary name with any ``<sha8>_`` prefix removed."""
|
||
return _HASH_PREFIX.sub("", raw)
|
||
|
||
|
||
def _library_name(snap: Snapshot) -> str:
|
||
return library_name(snap.binary.get("name") or "?")
|
||
|
||
|
||
def _variant_id(index: int) -> str:
|
||
"""0 -> A, 1 -> B, …, 26 -> AA (spreadsheet-style), so cells read 'A', 'B*', 'C*'."""
|
||
out = ""
|
||
index += 1
|
||
while index:
|
||
index, rem = divmod(index - 1, 26)
|
||
out = chr(ord("A") + rem) + out
|
||
return out
|
||
|
||
|
||
def _build_library(name: str, entries: list[Entry]) -> dict[str, Any]:
|
||
"""Cluster one library name's entries into variants and assemble its table row."""
|
||
# group by surface signature -> the entries that share it
|
||
by_sig: dict[str, list[Entry]] = {}
|
||
for e in entries:
|
||
by_sig.setdefault(surface_signature(e.snapshot), []).append(e)
|
||
|
||
# deterministic order: biggest cluster first, ties broken by earliest game name. The first
|
||
# is the reference variant (no asterisk); the rest diverge.
|
||
clusters = sorted(
|
||
by_sig.values(),
|
||
key=lambda es: (-len(es), min(e.game for e in es)),
|
||
)
|
||
reference_snap = clusters[0][0].snapshot
|
||
|
||
variants: list[dict[str, Any]] = []
|
||
variant_of_game: dict[str, list[str]] = {}
|
||
for i, cluster in enumerate(clusters):
|
||
vid = _variant_id(i)
|
||
is_ref = i == 0
|
||
sample = cluster[0].snapshot
|
||
members = [
|
||
{
|
||
"game": e.game,
|
||
"file": e.snapshot.binary.get("name"), # stored name, incl. the <sha8>_ prefix
|
||
"engine": e.snapshot.binary.get("engine"),
|
||
"compiler": e.snapshot.binary.get("compiler"),
|
||
"sha256": e.snapshot.binary.get("sha256"),
|
||
}
|
||
for e in sorted(cluster, key=lambda e: e.game)
|
||
]
|
||
for e in cluster:
|
||
variant_of_game.setdefault(e.game, []).append(vid)
|
||
variants.append(
|
||
{
|
||
"id": vid,
|
||
"reference": is_ref,
|
||
"surface": _surface_counts(sample),
|
||
"members": members,
|
||
"vs_reference": None if is_ref else _vs_reference(reference_snap, sample),
|
||
}
|
||
)
|
||
|
||
cells: dict[str, dict[str, Any]] = {}
|
||
for game, vids in variant_of_game.items():
|
||
uniq = sorted(set(vids))
|
||
cells[game] = {
|
||
"present": True,
|
||
"variants": uniq,
|
||
"diverges": any(v != "A" for v in uniq),
|
||
}
|
||
|
||
return {
|
||
"name": name,
|
||
"variant_count": len(variants),
|
||
"diverges": len(variants) > 1,
|
||
"cells": cells,
|
||
"variants": variants,
|
||
}
|
||
|
||
|
||
def compatibility_table(entries: Iterable[Entry]) -> dict[str, Any]:
|
||
"""Build the full compatibility table from catalogued entries.
|
||
|
||
Groups by library name (case-insensitive, keeping the most common spelling for display),
|
||
clusters each group into surface variants, and returns a JSON-able dict: a `games` column
|
||
list and a `libraries` list of rows. Libraries with >1 variant are the ones carrying
|
||
asterisks."""
|
||
entries = list(entries)
|
||
|
||
by_lib: dict[str, list[Entry]] = {}
|
||
spellings: dict[str, Counter] = {}
|
||
for e in entries:
|
||
name = _library_name(e.snapshot)
|
||
key = name.casefold()
|
||
by_lib.setdefault(key, []).append(e)
|
||
spellings.setdefault(key, Counter())[name] += 1
|
||
|
||
libraries = [
|
||
_build_library(spellings[key].most_common(1)[0][0], es)
|
||
for key, es in by_lib.items()
|
||
]
|
||
libraries.sort(key=lambda lib: lib["name"].casefold())
|
||
|
||
games = sorted({e.game for e in entries})
|
||
return {
|
||
"schema": "compat/1",
|
||
"generated_at": datetime.now(timezone.utc).isoformat(timespec="seconds"),
|
||
"games": games,
|
||
"libraries": libraries,
|
||
}
|
||
|
||
|
||
# --- renderers ---------------------------------------------------------------------------------
|
||
def render_json(table: dict[str, Any]) -> str:
|
||
return json.dumps(table, indent=2, sort_keys=True, ensure_ascii=False)
|
||
|
||
|
||
def _cell_marker(lib: dict[str, Any], game: str) -> str:
|
||
"""Matrix cell: empty if absent; ✓ for a single-variant library; else the variant letter(s)
|
||
with an asterisk on every non-reference (divergent) variant."""
|
||
cell = lib["cells"].get(game)
|
||
if not cell:
|
||
return ""
|
||
if lib["variant_count"] == 1:
|
||
return "✓"
|
||
return " / ".join(v + ("" if v == "A" else "\\*") for v in cell["variants"])
|
||
|
||
|
||
def _short(sha: str | None) -> str:
|
||
return (sha or "?")[:8]
|
||
|
||
|
||
def _signed(sign: str, items: list[str], cap: int = 15) -> str:
|
||
head = [sign + it for it in items[:cap]]
|
||
if len(items) > cap:
|
||
head.append("…+{0}".format(len(items) - cap))
|
||
return ", ".join(head)
|
||
|
||
|
||
def _emit_detail(out: list[str], vs: dict[str, Any]) -> None:
|
||
"""Nested, named breakdown of a divergent variant: which types changed, and per class which
|
||
methods/events/fields were added (`+`), removed (`−`) or changed (`~`)."""
|
||
det = vs.get("detail") or {}
|
||
|
||
t = det.get("types") or {}
|
||
toks = []
|
||
if t.get("added"):
|
||
toks.append(_signed("+", t["added"]))
|
||
if t.get("removed"):
|
||
toks.append(_signed("−", t["removed"]))
|
||
if t.get("changed"):
|
||
toks.append(_signed("~", t["changed"]))
|
||
if toks:
|
||
out.append(" - typy: {0}".format(" ".join(toks)))
|
||
|
||
for axis, label in (("methods", "metody"), ("events", "eventy"), ("fields", "pola")):
|
||
by = (det.get(axis) or {}).get("by_owner") or {}
|
||
if not by:
|
||
continue
|
||
out.append(" - {0} (klasy ze zmianami):".format(label))
|
||
for owner in by: # already sorted at build time
|
||
d = by[owner]
|
||
parts = []
|
||
if d.get("added"):
|
||
parts.append(_signed("+", d["added"]))
|
||
if d.get("removed"):
|
||
parts.append(_signed("−", d["removed"]))
|
||
if d.get("changed"):
|
||
parts.append(_signed("~", d["changed"]))
|
||
out.append(" - **{0}**: {1}".format(owner, " ".join(parts)))
|
||
|
||
|
||
def render_markdown(table: dict[str, Any]) -> str:
|
||
games = table["games"]
|
||
libs = table["libraries"]
|
||
out: list[str] = ["# Tablica kompatybilności silników", ""]
|
||
out.append(
|
||
"Grupowane po nazwie biblioteki. Komórka: pusta = brak; `✓` = obecna (jedyny wariant); "
|
||
"litera wariantu (`A`, `B\\*`, …) gdy ta sama nazwa biblioteki ma **różne powierzchnie "
|
||
"silnika** — `\\*` oznacza rozbieżność względem wariantu referencyjnego (`A`). "
|
||
"Sekcja *Warianty* niżej wymienia per klasa **które** metody/eventy/pola doszły (`+`), "
|
||
"zniknęły (`−`) lub się zmieniły (`~`)."
|
||
)
|
||
out.append("")
|
||
out.append("_Wygenerowano: {0} · gier: {1} · bibliotek: {2}_".format(
|
||
table.get("generated_at", "?"), len(games), len(libs)))
|
||
out.append("")
|
||
|
||
if not libs:
|
||
out.append("_(katalog pusty)_")
|
||
return "\n".join(out)
|
||
|
||
# --- matrix: library × game ---
|
||
header = "| Biblioteka | " + " | ".join(games) + " |"
|
||
sep = "|" + "---|" * (len(games) + 1)
|
||
out.append(header)
|
||
out.append(sep)
|
||
for lib in libs:
|
||
mark = "" if lib["variant_count"] == 1 else " \\*"
|
||
row = ["**{0}**{1}".format(lib["name"], mark)]
|
||
row += [_cell_marker(lib, g) for g in games]
|
||
out.append("| " + " | ".join(row) + " |")
|
||
out.append("")
|
||
|
||
# --- variant breakdown: only the interesting (multi-variant) libraries ---
|
||
divergent = [lib for lib in libs if lib["variant_count"] > 1]
|
||
if divergent:
|
||
out.append("## Warianty (\\* = rozbieżność powierzchni)")
|
||
out.append("")
|
||
for lib in divergent:
|
||
out.append("### {0} — {1} warianty".format(lib["name"], lib["variant_count"]))
|
||
for v in lib["variants"]:
|
||
s = v["surface"]
|
||
tag = "referencyjny" if v["reference"] else "rozbieżny \\*"
|
||
games_str = ", ".join(
|
||
"{0} ({1}/{2}, {3})".format(
|
||
m["game"], m["engine"] or "?", m["compiler"] or "?", _short(m["sha256"]))
|
||
for m in v["members"]
|
||
)
|
||
out.append(
|
||
"- **{0}** ({1}) — typy {2} / metody {3} / eventy {4} / pola {5}".format(
|
||
v["id"], tag, s["types"], s["methods"], s["events"], s["fields"]))
|
||
out.append(" - gry: {0}".format(games_str))
|
||
vs = v["vs_reference"]
|
||
if vs:
|
||
ax = vs["axes"]
|
||
deltas = ", ".join(
|
||
"{0} +{1}/-{2}/~{3}".format(a, ax[a]["added"], ax[a]["removed"], ax[a]["changed"])
|
||
for a in _AXES
|
||
)
|
||
moved = " · przeniesionych metod {0}".format(vs["moved_methods"]) if vs["moved_methods"] else ""
|
||
out.append(
|
||
" - vs **A**: overlap {0}% · {1}{2}".format(vs["overall"], deltas, moved))
|
||
_emit_detail(out, vs)
|
||
out.append("")
|
||
|
||
return "\n".join(out).rstrip() + "\n"
|
||
|
||
|
||
# --- CLI: build the table from snapshot.json files (no DB/server) -------------------------------
|
||
def _label_and_path(token: str) -> tuple[str | None, str]:
|
||
"""Parse a CLI token: 'Game Name=path.json' pins the column label, a bare 'path.json' lets the
|
||
label default (acquisition source, else the file's basename). The '=' must precede a real
|
||
file so paths that happen to contain '=' still work."""
|
||
import os
|
||
|
||
if "=" in token:
|
||
lhs, rhs = token.split("=", 1)
|
||
if os.path.exists(rhs):
|
||
return lhs.strip(), rhs
|
||
return None, token
|
||
|
||
|
||
def _default_label(path: str, snap: Snapshot) -> str:
|
||
import os
|
||
|
||
acq = snap.binary.get("acquisition") or {}
|
||
if acq.get("source"):
|
||
return str(acq["source"])
|
||
base = os.path.basename(path)
|
||
for suffix in (".snapshot.json", ".json"):
|
||
if base.endswith(suffix):
|
||
return base[: -len(suffix)]
|
||
return base
|
||
|
||
|
||
def main(argv: list[str] | None = None) -> int:
|
||
import argparse
|
||
|
||
p = argparse.ArgumentParser(
|
||
prog="ams.compat",
|
||
description="Build a cross-binary compatibility table grouped by library name.",
|
||
)
|
||
p.add_argument(
|
||
"snapshots",
|
||
nargs="+",
|
||
help="snapshot.json files; prefix with 'Label=' to set the game/column label",
|
||
)
|
||
p.add_argument("--format", choices=["md", "json"], default="md", help="output format")
|
||
p.add_argument("-o", "--out", help="write here instead of stdout")
|
||
args = p.parse_args(argv)
|
||
|
||
entries: list[Entry] = []
|
||
for token in args.snapshots:
|
||
label, path = _label_and_path(token)
|
||
snap = Snapshot.load(path)
|
||
entries.append(Entry(game=label or _default_label(path, snap), snapshot=snap))
|
||
|
||
table = compatibility_table(entries)
|
||
text = render_json(table) if args.format == "json" else render_markdown(table)
|
||
|
||
if args.out:
|
||
with open(args.out, "w", encoding="utf-8") as fh:
|
||
fh.write(text if text.endswith("\n") else text + "\n")
|
||
print("[+] {0} libraries, {1} games -> {2}".format(
|
||
len(table["libraries"]), len(table["games"]), args.out))
|
||
else:
|
||
print(text)
|
||
return 0
|
||
|
||
|
||
if __name__ == "__main__":
|
||
import sys
|
||
|
||
sys.exit(main())
|