XFE.SpaceEngineers.AgentBridge

"""Record actual AMS flight state and motion without losing case-sensitive minified field names."""
import argparse
import datetime as dt
import json
from pathlib import Path
import subprocess
import sys
import time

# Both exports have been checked against the minifier output: the fields below
# are identical. Do not accept an unreviewed hash after rebuilding AMS.
EXPECTED = {
    "2cc105a720ecd1552791816a0edf106cfab91b501a1e8048421b7f23382d333a",
    "b191430cf9ab7fe2b21569344ebcdc927bfa52847b84632440bdc68f3e5abc73",
}
STATES = {"ï": "Docked", "ð": "Servicing", "ò": "Departing", "ó": "Transit",
          "ô": "Survey", "ù": "Returning", "ú": "Holding", "û": "DockAlign",
          "ü": "DockApproach"}


def main():
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--block", required=True)
    parser.add_argument("--grid", required=True)
    parser.add_argument("--seconds", type=float, default=30)
    parser.add_argument("--interval", type=float, default=1)
    parser.add_argument("--out", type=Path, required=True)
    parser.add_argument("--pipe")
    args = parser.parse_args()
    if not 0 < args.seconds <= 600 or not .1 <= args.interval <= 30:
        parser.error("seconds must be 0..600 and interval must be 0.1..30")
    cli = Path(__file__).resolve().parents[1] / "src/XFE.SeAgent.Cli/bin/Release/net10.0/xfe-se.exe"

    def call(method, parameters):
        command = [str(cli), "call", method, "--params", json.dumps(parameters)]
        if args.pipe:
            command += ["--pipe", args.pipe]
        process = subprocess.run(command, capture_output=True, encoding="utf-8", timeout=35)
        response = json.loads(process.stdout)
        if process.returncode or "error" in response:
            raise RuntimeError(response.get("error", process.stderr))
        return response["result"]

    program = call("pb.read", {"entityId": args.block})
    if program["sha256"] not in EXPECTED:
        raise RuntimeError("Unrecognized AMS export " + program["sha256"] + "; derive a new field mapping for this source before tracing.")
    args.out.parent.mkdir(parents=True, exist_ok=True)
    started = time.monotonic()
    samples, previous = 0, None
    with args.out.open("x", encoding="utf-8") as output:
        while time.monotonic() - started < args.seconds:
            fields = call("pb.inspect", {"entityId": args.block, "fields": ["E", "H"], "depth": 3})["fields"]
            grid = call("grids.get", {"entityId": args.grid})
            role = fields["H"]
            flight = role["أ"]
            sample = {"utc": dt.datetime.now(dt.timezone.utc).isoformat(), "time": fields["E"],
                      "state": role["Ԙ"], "stateName": STATES.get(role["Ԙ"], role["Ԙ"]),
                      "sourceSha256": program["sha256"], "reason": role["l"], "position": grid["pose"]["position"],
                      "actualSpeed": grid["speed"], "commandSpeed": flight["ψ"], "clearance": flight["φ"],
                      "attitudeError": flight["χ"], "scanIndex": flight["Ͼ"], "scanBlocked": flight["ϭ"],
                      "problem": flight["ſ"], "camera": flight["Ѕ"], "cameraDetail": flight["І"],
                      "obstacle": flight["Ї"], "fields": fields}
            output.write(json.dumps(sample, ensure_ascii=False, separators=(",", ":")) + "\n")
            output.flush()
            state = (sample["state"], sample["reason"])
            if state != previous:
                print(json.dumps({k: sample[k] for k in ("time", "state", "reason", "actualSpeed", "commandSpeed", "scanIndex")}, ensure_ascii=False), flush=True)
                previous = state
            samples += 1
            time.sleep(args.interval)
    print(json.dumps({"samples": samples, "output": str(args.out.resolve())}), flush=True)


if __name__ == "__main__":
    sys.stdout.reconfigure(encoding="utf-8")
    main()