feat: add compatibility table endpoint and rendering for game libraries
This commit is contained in:
Patryk Gensch
336
ams/compat.py
Normal file
336
ams/compat.py
Normal file
@@ -0,0 +1,336 @@
|
||||
"""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
|
||||
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 _vs_reference(reference: Snapshot, variant: Snapshot) -> dict[str, Any]:
|
||||
"""How a divergent variant differs from its library's reference variant: per-axis +/-/~ counts
|
||||
plus the pooled surface-overlap score, reusing the existing diff/similarity engines."""
|
||||
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"]),
|
||||
}
|
||||
|
||||
|
||||
# --- table builder -----------------------------------------------------------------------------
|
||||
def _library_name(snap: Snapshot) -> str:
|
||||
return 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,
|
||||
"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 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`). "
|
||||
"Szczegóły rozbieżności w sekcji *Warianty* niżej."
|
||||
)
|
||||
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))
|
||||
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())
|
||||
Reference in New Issue
Block a user