Files
airdef/tools/airdef.py
T
2026-11-04 15:55:45 +03:00

456 lines
23 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
#!/usr/bin/env python3
"""AirDef: сборка, установка и запуск на Windows, macOS и Linux. Только стандартная библиотека Python.
python3 tools/airdef.py fork # инициализировать закреплённый сабмодуль Horizon (horizon/)
python3 tools/airdef.py game # собрать игру (форк) — нужен Rust
python3 tools/airdef.py mod # собрать мод airdef.hmod — нужен Rust с wasm32
python3 tools/airdef.py install # положить мод в папку игры и включить
python3 tools/airdef.py play [аргументы игры] # запустить форк на папке игры
python3 tools/airdef.py all # всё по порядку: fork, game, mod, install
python3 tools/airdef.py check # совпадают ли форк, упаковщик, уровень API мода и ресурсы
python3 tools/airdef.py test-wasm # собрать и прогнать настоящий .hmod в песочнице без GPU
python3 tools/airdef.py test-budgets # тяжёлый WASM budget сценарий + JSON provenance
Переменные окружения: HORIZON_HOME (папка игры), HORIZON_RESOURCES (подготовленные ресурсы
Minecraft 26.3), AIRDEF_GAME (готовый horizon-client вместо собранного).
"""
import glob
import hashlib
import json
import os
import platform
import re
import shutil
import subprocess
import sys
import tempfile
import zipfile
from pathlib import Path
ROOT = Path(__file__).resolve().parent.parent
FORK = ROOT / "horizon"
EXE = ".exe" if os.name == "nt" else ""
def data_dir() -> Path:
"""Как `dirs::data_dir()` в Horizon."""
system = platform.system()
if system == "Windows":
return Path(os.environ["APPDATA"])
if system == "Darwin":
return Path.home() / "Library" / "Application Support"
return Path(os.environ.get("XDG_DATA_HOME", Path.home() / ".local" / "share"))
def game_home() -> Path:
return Path(os.environ.get("HORIZON_HOME") or data_dir() / "horizon")
def run(cmd, cwd=None, env=None):
print("+", " ".join(str(c) for c in cmd), flush=True)
subprocess.run([str(c) for c in cmd], cwd=cwd, env=env, check=True)
def cargo() -> str:
found = shutil.which("cargo") or str(Path.home() / ".cargo" / "bin" / ("cargo" + EXE))
if not Path(found).exists() and not shutil.which("cargo"):
sys.exit("Нужен Rust: https://rustup.rs (затем: rustup target add wasm32-unknown-unknown)")
return found
def engine_commit() -> str:
"""The index pins the engine, including a deliberate not-yet-committed gitlink update."""
row = subprocess.check_output(["git", "ls-files", "--stage", "--", "horizon"], cwd=ROOT, text=True).strip().split()
if len(row) < 2 or row[0] != "160000":
sys.exit("Нет закреплённого сабмодуля horizon/: проверьте checkout AirDef")
return row[1]
def engine_identity() -> str:
commit = subprocess.check_output(["git", "rev-parse", "HEAD"], cwd=FORK, text=True).strip()
diff = subprocess.check_output(["git", "diff", "--binary", "HEAD"], cwd=FORK)
return commit + ("-" + hashlib.sha256(diff).hexdigest()[:16] if diff else "")
def fork():
"""Initialize the exact gitlink; never replace a modified developer checkout."""
expected = engine_commit()
matched = False
if (FORK / ".git").exists():
current = subprocess.check_output(["git", "rev-parse", "HEAD"], cwd=FORK, text=True).strip()
matched = current == expected
dirty = subprocess.check_output(["git", "status", "--porcelain"], cwd=FORK, text=True).strip()
if not matched and dirty:
sys.exit("horizon/: есть изменения при другом закреплённом SHA; сохраните их перед обновлением сабмодуля")
if (FORK / ".git").is_dir():
# Migrate the old standalone clone without deleting it or any local branches.
run(["git", "submodule", "absorbgitdirs", "--", "horizon"], cwd=ROOT)
url = subprocess.check_output(["git", "config", "-f", ".gitmodules", "--get", "submodule.horizon.url"], cwd=ROOT, text=True).strip()
run(["git", "remote", "set-url", "origin", url], cwd=FORK)
elif FORK.exists() and any(FORK.iterdir()):
sys.exit("horizon/: непустой каталог без Git; сохраните его перед инициализацией сабмодуля")
# Existing clones cache the previous upstream URL in local Git configuration.
# Synchronize even when HEAD already matches: the next pin may exist only in our fork.
run(["git", "submodule", "sync", "--recursive", "--", "horizon"], cwd=ROOT)
if matched:
print(f"horizon/: закреплённый коммит {expected}")
return
run(["git", "submodule", "update", "--init", "--recursive", "--", "horizon"], cwd=ROOT)
current = subprocess.check_output(["git", "rev-parse", "HEAD"], cwd=FORK, text=True).strip()
if current != expected:
sys.exit(f"horizon/: ожидался {expected}, получен {current}")
def game():
fork()
run([cargo(), "build", "--profile", "fast", "-p", "horizon-client", "--features", "mods"], cwd=FORK)
def built_game() -> Path:
if os.environ.get("AIRDEF_GAME"):
return Path(os.environ["AIRDEF_GAME"])
for name in (FORK / "target" / "fast" / ("horizon-client" + EXE), ROOT / ("horizon-client" + EXE)):
if name.exists():
return name
sys.exit("Нет собранной игры: python3 tools/airdef.py game (или AIRDEF_GAME=путь к horizon-client)")
def packer_path() -> Path:
return ROOT / ".horizon-cache" / "packer" / engine_identity() / "bin" / ("cargo-horizon-mod" + EXE)
def mod():
fork()
tool = packer_path()
if not tool.exists():
run([cargo(), "install", "--locked", "--no-default-features", "--target-dir", FORK / "target", "--root", tool.parent.parent,
"--path", FORK / "crates" / "cargo-horizon-mod"])
# Invoke the pinned executable directly; PATH may contain an unrelated old packer.
build_env = dict(os.environ)
build_env["PATH"] = str(Path(cargo()).parent) + os.pathsep + build_env.get("PATH", "")
run([tool, "horizon-mod", "build"], cwd=ROOT / "airdef-mod", env=build_env)
artifact = hmod()
identity = {"airdef": subprocess.check_output(["git", "rev-parse", "HEAD"], cwd=ROOT, text=True).strip(),
"engine": engine_identity(), "engine_pin": engine_commit(),
"wit_sha256": hashlib.sha256((FORK / "crates/horizon-mod-sdk/wit/horizon.wit").read_bytes()).hexdigest(),
"artifact_sha256": hashlib.sha256(artifact.read_bytes()).hexdigest(),
"airdef_dirty": bool(subprocess.check_output(["git", "status", "--porcelain", "--untracked-files=no"], cwd=ROOT)),
"airdef_diff_sha256": hashlib.sha256(subprocess.check_output(["git", "diff", "--binary", "HEAD"], cwd=ROOT)).hexdigest()}
artifact.with_name("build-identity.json").write_text(json.dumps(identity, indent=2) + "\n")
def hmod() -> Path:
for p in (ROOT / "airdef-mod" / "target" / "horizon-mod" / "airdef.hmod", ROOT / "airdef.hmod"):
if p.exists():
return p
sys.exit("Нет airdef.hmod: python3 tools/airdef.py mod")
def test_wasm():
"""The real component and host ABI, with the engine's normal fuel/memory limits."""
mod()
build_env = dict(os.environ)
build_env["PATH"] = str(Path(cargo()).parent) + os.pathsep + build_env.get("PATH", "")
# `full` is the packer's default feature; the lightweight release packer has no runtime.
run([cargo(), "run", "--locked", "--manifest-path", FORK / "Cargo.toml",
"-p", "cargo-horizon-mod", "--", "horizon-mod", "test", "--hmod", hmod()],
cwd=ROOT / "airdef-mod", env=build_env)
BUDGET_SCENARIO = "tests/budgets/wasm_flat_budget.toml"
BUDGET_ASSERTED_SHA256 = "bcc0359289ab20ab3de608bcc130f72d2cf88f9f6fbc470d7b6846e9db5d11df"
def fuel_measurements(output):
"""Read the pinned host's callback counters; never infer them from source caps."""
samples = []
for mod_id, callback, calls, maximum, budget, failures in re.findall(
r"(?m)^fuel (\S+) (\S+): calls=(\d+) max=(\d+) budget=(\d+) failures=(\d+)\s*$", output):
samples.append({"mod": mod_id, "callback": callback, "calls": int(calls),
"max_used": int(maximum), "budget": int(budget), "failures": int(failures)})
return samples
def test_budgets():
"""Run the opt-in heavy scenario and always write a machine-readable report."""
report_path = Path(os.environ.get("AIRDEF_BUDGET_REPORT") or
ROOT / ".horizon-cache" / "wasm-budget-report.json")
report = {
"schema": 1,
"command": "test-budgets",
"scenario": BUDGET_SCENARIO,
"scenario_render_world": None,
"status": "failed",
"exit_code": None,
"scenario_counts": {"scripts": 1, "steps": None, "raid_requests": None,
"raid_units_requested": None, "air_defense_units": None,
"layout_unit_limit": None, "frames": None, "ticks": None,
"accepted_raids_asserted": None, "snapshot_airborne_asserted": None,
"snapshot_tick": None, "peak_airborne_observed": None},
"runtime_limits": {"source": "horizon/crates/horizon-mods/src/runtime.rs",
"metering": "source_configured"},
"measured_fuel": {"available": False, "callbacks": [],
"scope": "guest callback fuel; excludes host CPU, memory and GPU time"},
}
def git(*args, text=True):
return subprocess.check_output(["git", *args], cwd=ROOT,
text=text, stderr=subprocess.DEVNULL)
try:
report["airdef"] = {
"sha": git("rev-parse", "HEAD").strip(),
"dirty": bool(git("status", "--porcelain", "--untracked-files=no").strip()),
"diff_sha256": hashlib.sha256(git("diff", "--binary", "HEAD", text=False)).hexdigest(),
}
report["engine"] = {"pin": engine_commit(), "identity": engine_identity()}
wit = FORK / "crates" / "horizon-mod-sdk" / "wit" / "horizon.wit"
report["api_level"] = fork_api()[0] if fork_api() else None
report["wit_sha256"] = hashlib.sha256(wit.read_bytes()).hexdigest() if wit.is_file() else None
runtime = (FORK / "crates" / "horizon-mods" / "src" / "runtime.rs").read_text()
layout = (ROOT / "airdef-mod" / "src" / "layout.rs").read_text()
limit = re.search(r"pub const MAX_UNITS: usize = (\d+);", layout)
report["scenario_counts"]["layout_unit_limit"] = int(limit.group(1)) if limit else None
memory = re.search(r"const MEMORY: usize = (\d+) << (\d+);", runtime)
report["runtime_limits"].update({
"memory_bytes": int(memory.group(1)) << int(memory.group(2)) if memory else None,
"fuel_per_call": {name.lower(): int(value) for name, value in
re.findall(r"const FUEL_(INIT|TICK|FRAME|EVENT): u64 = ([\d_]+);", runtime)
for value in [value.replace("_", "")]},
"wasmtime_component_model": bool(re.search(r"wasm_component_model\(true\)", runtime)),
"wasmtime_fuel_metering": bool(re.search(r"consume_fuel\(true\)", runtime)),
})
lock_text = (ROOT / "horizon" / "Cargo.lock").read_text()
wasmtime = re.search(r'\[\[package\]\]\s*name = "wasmtime"\s*version = "([^"]+)"', lock_text)
report["wasmtime_version"] = wasmtime.group(1) if wasmtime else None
fixture = ROOT / "airdef-mod" / BUDGET_SCENARIO
fixture_bytes = fixture.read_bytes()
fixture_text = fixture_bytes.decode("utf-8")
flat_heights = list(dict.fromkeys(re.findall(r"(?m)^render-ground-y\s*=\s*(-?\d+)\s*$", fixture_text)))
if flat_heights:
report["scenario_render_world"] = "flat at y=" + ",".join(flat_heights)
counts = report["scenario_counts"]
counts["steps"] = len(re.findall(r"(?m)^\[\[step\]\]$", fixture_text))
counts["air_defense_units"] = len(re.findall(r'(?m)^command = "airdef deploy [^"]+"', fixture_text))
requests = re.findall(r'(?m)^command = "airdef wave \S+ (\d+)(?: [^"]*)?"', fixture_text)
counts["raid_requests"] = len(requests)
counts["raid_units_requested"] = sum(map(int, requests))
for field in ("ticks", "frames"):
counts[field] = sum(map(int, re.findall(rf"(?m)^{field}\s*=\s*(\d+)\s*$", fixture_text)))
report["scenario_sha256"] = hashlib.sha256(fixture_bytes).hexdigest()
mod()
artifact = hmod()
report["artifact_sha256"] = hashlib.sha256(artifact.read_bytes()).hexdigest()
identity_path = artifact.with_name("build-identity.json")
identity = json.loads(identity_path.read_text())
identity_checks = {
"engine_pin": identity.get("engine_pin") == report["engine"]["pin"],
"engine_identity": identity.get("engine") == report["engine"]["identity"],
"airdef_sha": identity.get("airdef") == report["airdef"]["sha"],
"airdef_dirty": identity.get("airdef_dirty") == report["airdef"]["dirty"],
"airdef_diff_sha256": identity.get("airdef_diff_sha256") == report["airdef"]["diff_sha256"],
"artifact_sha256": identity.get("artifact_sha256") == report["artifact_sha256"],
"wit_sha256": identity.get("wit_sha256") == report["wit_sha256"],
}
report["build_identity"] = {"path": str(identity_path), "checks": identity_checks,
"matches": all(identity_checks.values())}
if not report["build_identity"]["matches"]:
raise RuntimeError("build identity is stale or does not match the tested artifact: " +
", ".join(k for k, ok in identity_checks.items() if not ok))
build_env = dict(os.environ)
build_env["PATH"] = str(Path(cargo()).parent) + os.pathsep + build_env.get("PATH", "")
with tempfile.TemporaryDirectory(prefix="airdef-wasm-budget-") as temp:
isolated = Path(temp)
(isolated / "tests").mkdir()
(isolated / "tests" / "wasm_budget.toml").write_bytes(fixture_bytes)
command = [cargo(), "run", "--locked", "--manifest-path", FORK / "Cargo.toml",
"-p", "cargo-horizon-mod", "--", "horizon-mod", "test", "--hmod", artifact,
"--fuel-profile", "budget"]
print("+", " ".join(str(c) for c in command), flush=True)
result = subprocess.run([str(c) for c in command], cwd=isolated, env=build_env,
text=True, stdout=subprocess.PIPE, stderr=subprocess.STDOUT)
print(result.stdout, end="", flush=True)
report["exit_code"] = result.returncode
report["status"] = "passed" if result.returncode == 0 else "failed"
report["output"] = result.stdout[-12000:]
samples = fuel_measurements(result.stdout)
report["measured_fuel"].update({"available": bool(samples), "callbacks": samples})
if result.returncode == 0 and not samples:
report["status"] = "failed"
raise RuntimeError("pinned runtime did not return requested callback fuel counters")
# These assertions belong to this exact fixture, not to arbitrary edited scripts.
if result.returncode == 0 and report["scenario_sha256"] == BUDGET_ASSERTED_SHA256:
counts["accepted_raids_asserted"] = 2
counts["snapshot_airborne_asserted"] = 184
counts["snapshot_tick"] = 2102
except Exception as exc:
report["error"] = f"{type(exc).__name__}: {exc}"
except SystemExit as exc:
report["error"] = f"SystemExit: {exc.code}"
finally:
report_path.parent.mkdir(parents=True, exist_ok=True)
report_path.write_text(json.dumps(report, indent=2, sort_keys=True) + "\n")
print(f"Budget report: {report_path}", flush=True)
if report["status"] != "passed":
sys.exit(f"WASM budget scenario failed; see {report_path}")
def install():
home = game_home()
mods = home / "horizon-mods"
mods.mkdir(parents=True, exist_ok=True)
shutil.rmtree(mods / "airdef", ignore_errors=True) # первая версия мода была папкой
shutil.copy2(hmod(), mods / "airdef.hmod")
listing = home / "horizon-mods.txt"
lines = listing.read_text().splitlines() if listing.exists() else []
lines = [l for l in lines if not l.startswith("mod.airdef:")] + ["mod.airdef:true"]
listing.write_text("\n".join(lines) + "\n")
print(f"Мод установлен в {mods} и включён")
def resources() -> Path:
if os.environ.get("HORIZON_RESOURCES"):
return Path(os.environ["HORIZON_RESOURCES"])
found = sorted(glob.glob(str(data_dir() / "horizon-launcher" / "versions" / "*" / ".local" / "horizon-client" / "26.3")))
local = FORK / ".local" / "horizon-client" / "26.3"
if local.exists():
found.append(str(local))
if not found:
sys.exit("Не найдены ресурсы игры. Установи Horizon через лаунчер и запусти его один раз "
"(он подготовит ресурсы из твоего Minecraft 26.3), либо задай HORIZON_RESOURCES.")
return Path(found[-1])
def play(args):
exe = built_game().resolve()
home = game_home()
home.mkdir(parents=True, exist_ok=True)
# Игра хранит настройки, моды и миры в текущей папке.
run([exe, "--resources", resources(), *args], cwd=home)
def fork_api():
"""Новейший уровень API форка и его таблица функций (horizon-mods/src/api_level.rs); None без форка."""
try:
text = (FORK / "crates" / "horizon-mods" / "src" / "api_level.rs").read_text(encoding="utf-8")
except OSError:
return None
try:
level = int(re.search(r"pub const LEVEL: u32 = (\d+);", text).group(1))
table = text[text.index("pub const FUNCTIONS"):]
table = table[:table.index("];")]
except (AttributeError, ValueError):
return None
return level, [(name, int(lvl)) for name, lvl in re.findall(r'\("([a-z.\-]+)", (\d+)\)', table)]
def needed_level(wasm: bytes, table) -> tuple:
"""Уровень, который нужен модулю: самая новая функция, которую он импортирует (как api_level::needed).
В компоненте имя функции записано как 0x00, длина, UTF-8 имя; по этим байтам и ищем."""
need = (0, None)
for name, level in table:
part = name.split(".", 1)[1].encode()
if level > need[0] and b"\x00" + bytes([len(part)]) + part in wasm:
need = (level, name)
return need
def game_level():
"""Уровень, который игра записала в себя (метка horizon-mod-api-level:NNNN;); None, если игры нет.
Игра без метки считается уровня 0, как в api_level::read_mark."""
try:
exe = built_game()
except SystemExit:
return None
if not exe.exists():
return None
found = re.search(rb"horizon-mod-api-level:(\d{4});", exe.read_bytes())
return int(found.group(1)) if found else 0
def check():
"""Говорит прямо, совпадают ли форк, упаковщик, уровень API мода с игрой и ресурсы. Код 0 — да, 1 — нет."""
commit = engine_commit()
problems = []
def line(name, ok, text):
print(f"[{'ok' if ok else 'нет'}] {name}: {text}")
if not ok:
problems.append(name)
current = subprocess.run(["git", "rev-parse", "HEAD"], cwd=FORK, capture_output=True, text=True).stdout.strip() if (FORK / ".git").exists() else ""
fork_ok = current == commit
line("форк", fork_ok, f"gitlink {commit}, checkout {current or 'не инициализирован'}")
packer_ok = fork_ok and packer_path().exists()
line("упаковщик", packer_ok, "repo-local, соответствует engine identity" if packer_ok else
"нет упаковщика этого форка: python3 tools/airdef.py mod")
api = fork_api()
try:
mod_file = hmod()
except SystemExit:
mod_file = None
if api is None:
line("уровень API", False, "нет или не читается horizon/crates/horizon-mods/src/api_level.rs: python3 tools/airdef.py fork")
elif mod_file is None:
line("уровень API", False, "нет airdef.hmod: python3 tools/airdef.py mod")
else:
level, table = api
with zipfile.ZipFile(mod_file) as z:
wasm = z.read("mod.wasm") if "mod.wasm" in z.namelist() else b""
need, function = needed_level(wasm, table)
game = game_level()
text = f"мод нужен {need} ({function}), форк {level}, игра " + ("не собрана" if game is None else str(game))
if game is None:
line("уровень API", False, text + ": python3 tools/airdef.py game")
else:
# Сравнивается уровень, не код: одинаковый уровень не доказывает одинаковые патчи.
supported = need <= game and game == level
line("уровень API", supported, text if supported else
text + ": игра не поддерживает мод или собрана из другого форка")
try:
res = resources()
except SystemExit as e:
res = None
line("ресурсы", False, str(e.code))
if res is not None:
try:
preparer = (FORK / "crates" / "horizon-prepare" / "src" / "lib.rs").read_text(encoding="utf-8")
schema = int(re.search(r"pub const SCHEMA: i64 = (\d+);", preparer).group(1))
except (OSError, AttributeError):
schema = None
try:
manifest = json.loads((res / "horizon-resources.json").read_text(encoding="utf-8"))
except (OSError, ValueError):
manifest = {}
ok = schema is not None and manifest.get("schema") == schema and manifest.get("minecraft") == "26.3"
line("ресурсы", ok, f"{res}, схема {schema}, Minecraft 26.3" if ok else
f"{res}: схема {manifest.get('schema')}, Minecraft {manifest.get('minecraft')}; нужны схема {schema} и Minecraft 26.3")
if problems:
sys.exit("Не совпадает: " + ", ".join(problems) + ".")
print("Всё совпадает: форк, упаковщик, уровень API мода и ресурсы.")
def main():
if len(sys.argv) < 2:
sys.exit(__doc__)
cmd, rest = sys.argv[1], sys.argv[2:]
steps = {"fork": fork, "game": game, "mod": mod, "install": install, "check": check, "test-wasm": test_wasm, "test-budgets": test_budgets}
if cmd == "play":
play(rest)
elif cmd == "all":
for step in ("fork", "game", "mod", "install"):
steps[step]()
elif cmd in steps:
steps[cmd]()
else:
sys.exit(__doc__)
if __name__ == "__main__":
main()