SOLIDWORKS Bridge

#!/usr/bin/env python3
"""cswa-check.py - the CSWA benchmark on REAL SOLIDWORKS (0.11.0).

Every spec in resources/cswa/*.json is one exam-style part: the published answer key (mass in grams,
sometimes volume / centre of mass / surface area), the material, and the swx_run_script that builds
it. Each part is built through the bridge, weighed with swx_mass_properties, and compared with the
key within the spec's tolerance (CSWA accepts about 1 %). Nothing is trusted that was not measured.

    python3 tools/cswa-check.py [--target adom_drew2] [--report auto|<path>] [--only cswa-00-lego-2x4] [--fresh] [--keep]
    python3 tools/cswa-check.py --derivations      # re-derive every derived key (no bridge call)

Same rules as tools/dsl-live-check.py: every creation says purpose:"bridge"; by default the parts are
built in ONE bridge-made part (reuseDocument: its leading new_part becomes clear_part), so N parts
cost one document; the run STOPS on a user_present / low_commit_memory / dialog refusal and never
creates another part after one. The report (every reply, every expectation, pass/fail) is written
to cswa-report.json (--report auto: tools/cswa/reports/cswa-<date>-sw<year>-<version>.json);
tools/release.sh --require-live <report> refuses to publish without a passing one newer than the last code
commit. The report also says WHICH BUILD ran (`bridge`: the installed exe's sha256, read back through ab), and
the publish ships only the canary zip holding that exe (README, "Live validation").
"""
import argparse, base64, glob, hashlib, json, math, os, re, subprocess, sys, time

TARGET = "adom_drew2"
THREAD = "SolidWorks Bridge"
ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
STOP_CODES = ("user_present", "low_commit_memory", "stopped_by_dialog", "dialog_open", "user_present_refusal")


def ascii_reason(args):
    """A non-ASCII `reason` makes ab fail the call before it reaches the bridge (measured 2026-10-13)."""
    if isinstance(args, dict) and isinstance(args.get("reason"), str):
        args = dict(args, reason=args["reason"].encode("ascii", "replace").decode("ascii"))
    return args


def call(verb, args, timeout=300):
    args = ascii_reason(args)
    try:
        r = subprocess.run(["adom-bridge", "--target", TARGET, "--ai-thread", THREAD, verb, json.dumps(args)],
                           capture_output=True, text=True, timeout=timeout + 60)
    except subprocess.TimeoutExpired:
        return {"status": "error", "errorCode": "no_answer", "error": f"{verb} did not return in {timeout + 60}s"}
    t = r.stdout
    i = t.find("{")
    if i < 0:
        return {"status": "error", "errorCode": "not_json", "error": (t + r.stderr)[:400]}
    try:
        d = json.loads(t[i:])
    except ValueError:
        return {"status": "error", "errorCode": "not_json", "error": t[i:i + 400]}
    return d["result"] if isinstance(d, dict) and isinstance(d.get("result"), dict) else d


# ── spec variables: the SAME grammar as src/cswa.rs (eval / expand) ──────────
FUNCS1 = {"sqrt": math.sqrt, "sin": math.sin, "cos": math.cos, "tan": math.tan, "asin": math.asin, "acos": math.acos,
          "atan": math.atan, "abs": abs, "rad": math.radians, "deg": math.degrees}
FUNCS2 = {"atan2": math.atan2, "min": min, "max": max}


def evaluate(expr, vars_):
    """+ - * / ^ (right-assoc), unary minus, parentheses, pi, the functions above. No eval()."""
    s, i = expr, [0]

    def ws():
        while i[0] < len(s) and s[i[0]].isspace():
            i[0] += 1

    def peek():
        ws()
        return s[i[0]] if i[0] < len(s) else None

    def e():
        v = t()
        while peek() in ("+", "-"):
            op = s[i[0]]; i[0] += 1
            v = v + t() if op == "+" else v - t()
        return v

    def t():
        v = pw()
        while peek() in ("*", "/"):
            op = s[i[0]]; i[0] += 1
            v = v * pw() if op == "*" else v / pw()
        return v

    def pw():
        b = un()
        if peek() == "^":
            i[0] += 1
            return b ** pw()
        return b

    def un():
        c = peek()
        if c == "-":
            i[0] += 1
            return -un()
        if c == "+":
            i[0] += 1
            return un()
        return atom()

    def atom():
        c = peek()
        if c == "(":
            i[0] += 1
            v = e()
            if peek() != ")":
                raise ValueError(f"missing ) in {expr!r}")
            i[0] += 1
            return v
        if c is not None and (c.isdigit() or c == "."):
            st = i[0]
            while i[0] < len(s) and (s[i[0]].isdigit() or s[i[0]] in ".e" or (s[i[0]] in "+-" and i[0] > st and s[i[0] - 1] == "e")):
                i[0] += 1
            return float(s[st:i[0]])
        if c is not None and (c.isalpha() or c == "_"):
            st = i[0]
            while i[0] < len(s) and (s[i[0]].isalnum() or s[i[0]] == "_"):
                i[0] += 1
            name = s[st:i[0]]
            if peek() == "(":
                i[0] += 1
                args = [e()]
                while peek() == ",":
                    i[0] += 1
                    args.append(e())
                if peek() != ")":
                    raise ValueError(f"missing ) after {name}( in {expr!r}")
                i[0] += 1
                if name in FUNCS1 and len(args) == 1:
                    return float(FUNCS1[name](args[0]))
                if name in FUNCS2 and len(args) == 2:
                    return float(FUNCS2[name](args[0], args[1]))
                raise ValueError(f"unknown function {name}/{len(args)} in {expr!r}")
            if name == "pi":
                return math.pi
            if name not in vars_:
                raise ValueError(f"unknown variable {name} in {expr!r}")
            return vars_[name]
        raise ValueError(f"unexpected {c!r} in {expr!r}")

    v = e()
    if peek() is not None:
        raise ValueError(f"unexpected text in {expr!r}")
    if not math.isfinite(v):
        raise ValueError(f"{expr!r} is not finite")
    return v


def resolve_extends(spec, by_id, depth=0):
    base_id = spec.get("extends")
    if not base_id:
        return json.loads(json.dumps(spec))
    if depth > 8:
        raise ValueError(f"{spec['id']}: extends goes more than 8 deep")
    merged = resolve_extends(by_id[base_id], by_id, depth + 1)
    # What is true of the BASE part only is not inherited (src/cswa.rs::resolve_extends).
    for k in ("needsOps", "derivation", "simulatorSkips"):
        merged.pop(k, None)
    variables = dict(merged.get("variables") or {})
    variables.update(spec.get("variables") or {})
    for k, v in spec.items():
        if k not in ("extends", "variables", "scriptAppend", "derivedAppend"):
            merged[k] = v
    merged["variables"] = variables
    # A derivedAppend name the base already derives would shadow it (src/cswa.rs::resolve_extends).
    taken = {p[0] for p in merged.get("derived") or []}
    for name, _ in spec.get("derivedAppend") or []:
        if name in taken:
            raise ValueError(f"{spec['id']}: derivedAppend redefines {name!r}, which {base_id} already derives - name it differently")
    for key, into in (("scriptAppend", "script"), ("derivedAppend", "derived")):
        if key in spec:
            merged.setdefault(into, []).extend(spec[key])
    return merged


def expand(spec, by_id):
    spec = resolve_extends(spec, by_id)
    vars_ = {k: float(v) for k, v in (spec.get("variables") or {}).items()}
    for name, ex in spec.get("derived") or []:
        vars_[name] = evaluate(ex, vars_)

    def sub(v):
        if isinstance(v, str) and v.startswith("="):
            return evaluate(v[1:], vars_)
        if isinstance(v, list):
            return [sub(x) for x in v]
        if isinstance(v, dict):
            return {k: sub(x) for k, x in v.items()}
        return v

    for key in ("script", "expected"):
        if key in spec:
            spec[key] = sub(spec[key])
    return spec


def spec_vars(spec):
    """The spec's variables and derived values, by the spec grammar (as expand() computes them)."""
    vars_ = {k: float(v) for k, v in (spec.get("variables") or {}).items()}
    for name, ex in spec.get("derived") or []:
        vars_[name] = evaluate(ex, vars_)
    return vars_


QUANTITIES = ("massG", "volumeMm3", "centerOfMassMm", "surfaceAreaMm2")


def check_derivation(spec, raw):
    """0.11.4: a DERIVED key must be reproducible. Evaluate the spec's `derivation` ([name, Python
    expression] pairs, in order; math's functions and the spec's variables in scope) and compare every
    quantity it yields with the key: within 1e-6 (relative, or absolute below 1). Returns (values, problems)."""
    import math as _m
    ns = {k: getattr(_m, k) for k in dir(_m) if not k.startswith("_")}
    ns.update({"abs": abs, "min": min, "max": max, "sum": sum, "len": len, "round": round, "range": range})
    ns.update(spec_vars(raw))
    problems, values = [], {}
    for name, ex in spec.get("derivation") or []:
        try:
            ns[name] = eval(ex, {"__builtins__": {}}, ns)  # noqa: S307 - the repo's own spec files, a reviewer's tool
        except Exception as e:  # pragma: no cover - reported
            problems.append(f"{name} = {ex}: {type(e).__name__}: {e}")
            return values, problems
        if name in QUANTITIES:
            values[name] = ns[name]
    exp = spec.get("expected") or {}
    if "massG" not in values:
        problems.append("the derivation does not yield massG")
    for q in QUANTITIES:
        if exp.get(q) is None:
            continue
        if q not in values:
            problems.append(f"the key has {q} but the derivation does not yield it")
            continue
        got, want = values[q], exp[q]
        pairs = list(zip(got, want)) if isinstance(want, list) else [(got, want)]
        if isinstance(want, list) and (not isinstance(got, (list, tuple)) or len(got) != len(want)):
            problems.append(f"{q}: the derivation gives {got}, the key {want}")
            continue
        for g, w in pairs:
            if abs(g - w) > 1e-6 * max(1.0, abs(w)):
                problems.append(f"{q}: the derivation gives {got}, the key says {want}")
                break
    return values, problems


# ── WHICH BUILD RAN (the canary gate) ─────────────────────────────────────────
# tools/release.sh --publish ships the exact zip a live run tested (the canary), and refuses a report that does
# not say which build it ran. The bridge cannot hash the zip it came in (ab unzips it), so the facts come from
# ab, read-only: bridge_info (the install path), the INSTALLED exe read back through read_file (base64, in
# ranges) and hashed here, and - best effort - the manifest ab installed it from (its zip sha256, which ab
# checked before extracting).
WIKI = "https://wiki.adom.inc"


def remote_sha256(path, chunk=1_000_000):
    """sha256 of a file on the target, read back through ab's read_file in ranges. (hex, None) or (None, why)."""
    h, off, total = hashlib.sha256(), 0, None
    for _ in range(64):
        r = call("read_file", {"path": path, "encoding": "base64", "offset": off, "length": chunk}, timeout=120)
        if not isinstance(r, dict) or r.get("success") is False or r.get("content") is None:
            return None, f"read_file {path}: {str(r)[:200]}"
        data = base64.b64decode(r["content"])
        h.update(data)
        total = r.get("totalSize", r.get("size"))
        off = r.get("nextOffset", off + len(data))
        if r.get("eof") or not data or (total is not None and off >= total):
            return h.hexdigest(), None
    return None, f"read_file {path}: more than 64 ranges"


def installed_from():
    """The manifest URL ab installed the bridge from (~/.adom/bridge-install-urls.json on the target), or None."""
    r = call("read_file", {"path": "~/.adom/bridge-install-urls.json"}, timeout=60)
    try:
        d = json.loads(r.get("content") or "")
    except (AttributeError, ValueError):
        return None
    if isinstance(d, dict):
        for k in ("bridges", "urls"):
            if isinstance(d.get(k), (dict, list)):
                d = d[k]
                break
    entries = d.items() if isinstance(d, dict) else [((e or {}).get("name"), e) for e in d if isinstance(e, dict)] if isinstance(d, list) else []
    for name, v in entries:
        if name == "solidworks":
            return v if isinstance(v, str) else (v or {}).get("manifestUrl") or (v or {}).get("url")
    return None


def wiki_json(url):
    """A wiki page file as JSON: https://wiki.adom.inc/api/v1/pages/<slug>/files/<path>, or the page form
    https://wiki.adom.inc/<owner>/<slug>/files/<path>, read with adom-wiki api (authenticated)."""
    m = re.match(r"^https://wiki\.adom\.inc(/api/v1/pages/[^/]+/files/.+)$", url or "")
    if not m:
        m2 = re.match(r"^https://wiki\.adom\.inc/[^/]+/([^/]+)/files/(.+)$", url or "")
        if not m2:
            return None
        api = f"/api/v1/pages/{m2.group(1)}/files/{m2.group(2)}"
    else:
        api = m.group(1)
    try:
        out = subprocess.run(["adom-wiki", "api", api], capture_output=True, text=True, timeout=60).stdout
        return json.loads(out)
    except (OSError, subprocess.TimeoutExpired, ValueError):
        return None


def bridge_facts():
    facts = {"source": "ab bridge_info + read_file (read-only)"}
    info = call("bridge_info", {"name": "solidworks"}, timeout=60)
    if not isinstance(info, dict) or info.get("success") is False or info.get("status") == "error":
        facts["why"] = f"bridge_info failed: {str(info)[:200]}"
        return facts
    facts.update({"version": info.get("version"), "installPath": info.get("installPath"), "manifestUrl": info.get("manifestUrl"),
                  "installSource": info.get("source")})
    path = (info.get("installPath") or "").rstrip("\\/")
    if path:
        facts["exeSha256"], why = remote_sha256(path + "\\swx-bridge.exe")
        if why:
            facts["exeWhy"] = why
        facts["bridgeJsonSha256"], _ = remote_sha256(path + "\\bridge.json")
    else:
        facts["exeSha256"], facts["exeWhy"] = None, "bridge_info gave no installPath"
    src = installed_from()
    facts["installedFrom"] = src
    m = wiki_json(src) if src else None
    if isinstance(m, dict):
        facts["zipSha256"] = m.get("sha256")
        facts["installedManifestVersion"] = m.get("version")
        if m.get("canary"):
            facts["canary"] = m.get("canary")
    return facts


def report_path(arg, st, sw):
    """--report auto: tools/cswa/reports/cswa-<UTC date>-sw<year>-<bridge version>.json (never overwriting one)."""
    if arg != "auto":
        return arg
    year = (sw.get("versionSupport") or {}).get("year") or (sw.get("majorVersion") or 0) + 1992
    base = os.path.join(ROOT, "tools", "cswa", "reports", f"cswa-{time.strftime('%Y-%m-%d', time.gmtime())}-sw{year}-{st.get('bridgeVersion')}")
    path, n = base + ".json", 2
    while os.path.exists(path):
        path, n = f"{base}-{n}.json", n + 1
    return path


def ok(r):
    return isinstance(r, dict) and r.get("success") is not False and r.get("status") != "error"


def pct_off(got, want):
    if want == 0:
        return 0.0 if got == 0 else float("inf")
    return abs(got - want) / abs(want) * 100.0


def compare(spec, mp):
    """The same rules as src/cswa.rs::compare (the self-test's): mass always; volume, surface area and
    centre of mass when the key has them; a coordinate passes within tol % or 0.05 mm."""
    exp = spec["expected"]
    tol = exp.get("tolerancePct", 1.0)
    checks = []
    mass = mp.get("massG")
    if mass is None and mp.get("massKg") is not None:
        mass = mp["massKg"] * 1000.0

    def one(q, got, want):
        off = None if got is None else pct_off(got, want)
        checks.append({"quantity": q, "expected": want, "measured": got, "offPct": off, "pass": off is not None and off <= tol})

    one("massG", mass, exp["massG"])
    if exp.get("volumeMm3") is not None:
        v = mp.get("volumeMm3")
        if v is None and mp.get("volumeM3") is not None:
            v = mp["volumeM3"] * 1e9
        one("volumeMm3", v, exp["volumeMm3"])
    if exp.get("surfaceAreaMm2") is not None:
        a = mp.get("surfaceAreaMm2")
        if a is None and mp.get("surfaceAreaM2") is not None:
            a = mp["surfaceAreaM2"] * 1e6
        one("surfaceAreaMm2", a, exp["surfaceAreaMm2"])
    if exp.get("centerOfMassMm") is not None:
        got = mp.get("centerOfMassMm") or []
        want = exp["centerOfMassMm"]
        good = len(got) == 3 and all(abs(g - w) <= max(abs(w) * tol / 100.0, 0.05) for g, w in zip(got, want))
        checks.append({"quantity": "centerOfMassMm", "expected": want, "measured": got, "pass": good})
    return all(c["pass"] for c in checks), checks


def main():
    global TARGET
    ap = argparse.ArgumentParser()
    ap.add_argument("--target", default=TARGET)
    ap.add_argument("--only", default="")
    ap.add_argument("--fresh", action="store_true", help="a fresh part per spec instead of one reused part")
    ap.add_argument("--keep", action="store_true", help="leave the parts open")
    ap.add_argument("--report", default="cswa-report.json", help="the report path, or auto: tools/cswa/reports/cswa-<date>-sw<year>-<version>.json")
    ap.add_argument("--expand-only", action="store_true", help="print every spec's script with its variables substituted, and stop (no bridge call)")
    ap.add_argument("--derivations", action="store_true", help="re-derive every derived key from its spec's `derivation` and compare (no bridge call)")
    a = ap.parse_args()
    TARGET = a.target
    raw = [json.load(open(p)) for p in sorted(glob.glob(os.path.join(ROOT, "resources", "cswa", "*.json")))]
    by_id = {s["id"]: s for s in raw}
    specs = [expand(s, by_id) for s in raw]
    if a.expand_only:
        print(json.dumps({s["id"]: {"script": s.get("script"), "expected": s.get("expected"), "needsOps": s.get("needsOps") or []} for s in specs}))
        return 0
    if a.derivations:
        out, bad = {}, 0
        for spec in specs:
            if not spec.get("derivation"):
                continue
            values, problems = check_derivation(spec, resolve_extends(by_id[spec["id"]], by_id))
            out[spec["id"]] = {"derived": values, "expected": spec.get("expected"), "problems": problems, "ok": not problems}
            bad += bool(problems)
        print(json.dumps(out, indent=1))
        print(f"{len(out) - bad} of {len(out)} derived keys reproduced", file=sys.stderr)
        return 1 if bad else 0
    only = {x.strip() for x in a.only.split(",") if x.strip()}
    st = call("swx_status", {"reason": "cswa-check: is SOLIDWORKS ready"})
    sw = st.get("solidworks") or {}
    print(f"bridge {st.get('bridgeVersion')} | SOLIDWORKS running={sw.get('running')} ready={sw.get('ready')} "
          f"version={sw.get('majorVersion')} edition={sw.get('edition')}")
    if not sw.get("running"):
        print("SOLIDWORKS is not running on the target: start it (swx_start) and run again. Nothing was done.")
        return 2
    a.report = report_path(a.report, st, sw)
    facts = bridge_facts()
    print(f"bridge build: exe sha256 {(facts.get('exeSha256') or 'UNREAD - ' + str(facts.get('exeWhy') or facts.get('why')))[:76]}"
          f" | installed from {facts.get('installedFrom') or '?'}")
    shared = None
    if not a.fresh:
        nd = call("swx_new_document", {"type": "part", "purpose": "bridge", "reason": "cswa-check: the one part every CSWA spec is built in"})
        if nd.get("errorCode") in STOP_CODES:
            print(f"STOP  the bridge refused to create the shared part ({nd.get('errorCode')}). Nothing was built.")
            return 3
        shared = (nd.get("documentIdentity") or {}).get("path") or (nd.get("created") or {}).get("path") or (nd.get("created") or {}).get("title")
    results, stopped = [], None
    for spec in specs:
        sid = spec["id"]
        if only and sid not in only:
            continue
        rec = {"id": sid, "title": spec.get("title"), "official": spec.get("official"), "source": spec.get("source"),
               "expected": spec["expected"], "material": spec.get("material")}
        if spec.get("needsOps"):
            rec["skipped"] = f"needs ops the bridge does not have yet: {', '.join(spec['needsOps'])}"
            results.append(rec)
            print(f"SKIP  {sid}  {rec['skipped']}")
            continue
        t0 = time.time()
        req = {"script": spec["script"], "purpose": "bridge", "reason": f"cswa-check {sid}"}
        if shared:
            req["reuseDocument"] = shared
        r = call("swx_run_script", req)
        rec["runScript"] = r
        doc = r.get("documentName")
        if not ok(r):
            rec["pass"] = False
            rec["fails"] = [f"swx_run_script failed: {r.get('errorCode')} {str(r.get('error'))[:300]}"]
            if r.get("errorCode") in STOP_CODES:
                stopped = r.get("errorCode")
        else:
            mp = call("swx_mass_properties", {"targetDocument": doc, "reason": f"cswa-check {sid}: weigh it"})
            rec["massProperties"] = mp
            rec["pass"], rec["checks"] = compare(spec, mp)
            rec["fails"] = [f"{c['quantity']}: {c.get('measured')} vs {c['expected']} ({c.get('offPct')} % off)" for c in rec["checks"] if not c["pass"]]
        rec["seconds"] = round(time.time() - t0, 1)
        results.append(rec)
        print(f"{'PASS' if rec['pass'] else 'FAIL'}  {sid}  {spec.get('title')}  ({rec['seconds']} s)")
        for f in rec.get("fails", []):
            print(f"        - {f}")
        if stopped:
            print(f"STOP  the bridge refused ({stopped}); no more parts will be built. Re-run when the user is idle / memory is back / the dialog is answered.")
            break
        if not shared and not a.keep and doc:
            rw = call("swx_close_doc", {"targetDocument": doc, "save": False, "reason": f"cswa-check: close {sid}"})
            if not ok(rw):
                stopped = rw.get("errorCode") or "close_refused"
                print(f"STOP  the close after {sid} was refused ({stopped}); no more parts will be built.")
                break
    if shared and not a.keep:
        call("swx_close_doc", {"path" if ("\\" in shared or "/" in shared) else "targetDocument": shared, "save": False, "reason": "cswa-check: close the shared part"})
    ran = [r for r in results if "skipped" not in r]
    passed = sum(1 for r in ran if r.get("pass"))
    report = {"tool": "cswa-check", "when": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()), "whenEpoch": int(time.time()),
              "target": TARGET, "bridgeVersion": st.get("bridgeVersion"), "bridge": facts,
              "solidworks": {k: sw.get(k) for k in ("majorVersion", "edition", "versionSupport")},
              "summary": f"{passed} of {len(ran)} CSWA parts within tolerance", "passed": passed, "ran": len(ran),
              "skipped": len(results) - len(ran), "stopped": stopped, "allPassed": bool(ran) and passed == len(ran) and not stopped,
              "parts": results}
    json.dump(report, open(a.report, "w"), indent=1)
    print(f"\n{report['summary']}{' (STOPPED: ' + stopped + ')' if stopped else ''}. Full evidence: {a.report}")
    return 0 if report["allPassed"] else 1


if __name__ == "__main__":
    sys.exit(main())