"""Run the guarded ANSYS 2024 R2 v2 acceptance suite on a dedicated Windows node.""" from __future__ import annotations import argparse import csv import hashlib import json import re import shutil import subprocess import sys import time from datetime import datetime, timezone from pathlib import Path from uuid import uuid4 WORKER = Path(__file__).with_name("worker.py") BUILTIN_GEOMETRY = Path(__file__).with_name("acceptance-coupon.step") PROCESS_NAMES = {"ansys.exe", "ansyswbu.exe", "mechanical.exe", "mapdl.exe"} def _request(geometry_id: str) -> dict: return { "schema_version": 2, "inputs": [{"key": "geometry", "artifact_id": geometry_id}], "operation": { "analysis": { "type": "static_structural", "title": "PyMechanical 2024 R2 acceptance coupon", "unit_system": "mm_kg_s", "geometry": {"input": "geometry"}, "material": "structural_steel", "mesh": {"global_size": 2.0}, "boundary_conditions": [ { "type": "fixed_support", "name": "fixed_face", "scope": { "type": "planar_faces", "normal": [0, 0, 1], "offset": -1.5, "tolerance": 0.001, "expected_count": 1, }, } ], "loads": [ { "type": "force", "name": "load_face", "scope": { "type": "planar_faces", "normal": [0, 0, 1], "offset": 0.425, "tolerance": 0.001, "expected_count": 1, }, "components": [0, 0, -1000], } ], "results": [ "total_deformation", "equivalent_stress", "reaction_force", ], } }, "outputs": [ {"key": "project", "format": "mechdat"}, {"key": "stress_image", "format": "png"}, {"key": "deformation_image", "format": "png"}, {"key": "solver_log", "format": "txt"}, ], } def _stage(root: Path, geometry: Path, label: str) -> Path: job_dir = root / label request_dir = job_dir / "request" input_dir = job_dir / "input" / "geometry" request_dir.mkdir(parents=True) input_dir.mkdir(parents=True) shutil.copyfile(geometry, input_dir / geometry.name) request = _request(str(uuid4())) normalized = json.dumps(request, ensure_ascii=False, sort_keys=True, separators=(",", ":")) record = { "job_id": str(uuid4()), "lease_id": str(uuid4()), "request_digest": hashlib.sha256(normalized.encode("utf-8")).hexdigest(), "request": request, } (request_dir / "request.json").write_text( json.dumps(record, ensure_ascii=False, indent=2), encoding="utf-8" ) return job_dir def _mechanical_processes() -> dict[int, str]: completed = subprocess.run( ["tasklist.exe", "/fo", "csv", "/nh"], check=True, capture_output=True, text=True, encoding="utf-8", errors="replace", ) processes: dict[int, str] = {} for row in csv.reader(completed.stdout.splitlines()): name = row[0].casefold() if row else "" if len(row) >= 2 and (name in PROCESS_NAMES or re.fullmatch(r"ansys\d{3}\.exe", name)): processes[int(row[1])] = row[0] return processes def _wait_for_release(baseline: dict[int, str], timeout_seconds: int) -> dict[int, str]: deadline = time.monotonic() + timeout_seconds while True: remaining = { pid: name for pid, name in _mechanical_processes().items() if pid not in baseline } if not remaining: return {} if time.monotonic() >= deadline: return remaining time.sleep(2) def _run_success(job_dir: Path, baseline: dict[int, str], release_wait: int) -> dict: started = time.monotonic() completed = subprocess.run( [sys.executable, str(WORKER), "--acceptance", str(job_dir)], capture_output=True, text=True, encoding="utf-8", errors="replace", check=False, ) elapsed = time.monotonic() - started terminal = json.loads((job_dir / "terminal.json").read_text(encoding="utf-8")) remaining = _wait_for_release(baseline, release_wait) if completed.returncode != 0 or terminal["status"] != "succeeded": raise RuntimeError(f"BENCHMARK_FAILED:{completed.stderr[-500:]}") if remaining: raise RuntimeError(f"MECHANICAL_PROCESS_REMAINS:{remaining}") expected = { "project", "summary", "result_table", "selection_preview", "stress_image", "deformation_image", "solver_log", "analysis_spec", "provenance", } actual = {item["artifact_id"] for item in terminal["artifact_manifest"]} if actual != expected: raise RuntimeError(f"ARTIFACT_MANIFEST_MISMATCH:{sorted(actual)}") return {"elapsed_seconds": elapsed, "terminal": terminal} def _run_cancellation( job_dir: Path, baseline: dict[int, str], cancel_after: int, release_wait: int ) -> dict: process = subprocess.Popen( [sys.executable, str(WORKER), "--acceptance", str(job_dir)], stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True, encoding="utf-8", errors="replace", ) time.sleep(cancel_after) if process.poll() is not None: raise RuntimeError("CANCELLATION_JOB_FINISHED_BEFORE_CANCEL") killed = subprocess.run( ["taskkill.exe", "/pid", str(process.pid), "/t", "/f"], capture_output=True, text=True, encoding="utf-8", errors="replace", check=False, ) process.wait(timeout=30) remaining = _wait_for_release(baseline, release_wait) if killed.returncode != 0: raise RuntimeError(f"PROCESS_TREE_CANCEL_FAILED:{killed.stderr[-500:]}") if remaining: raise RuntimeError(f"CANCELLED_PROCESS_REMAINS:{remaining}") return {"worker_pid": process.pid, "exit_code": process.returncode} def main() -> int: parser = argparse.ArgumentParser() parser.add_argument("geometry", type=Path, nargs="?", default=BUILTIN_GEOMETRY) parser.add_argument("--work-root", type=Path, required=True) parser.add_argument("--repeat", type=int, default=20) parser.add_argument("--cancel-after", type=int, default=10) parser.add_argument("--release-wait", type=int, default=60) args = parser.parse_args() if sys.platform != "win32": raise RuntimeError("Acceptance requires Windows") geometry = args.geometry.resolve(strict=True) if geometry.suffix.casefold() not in {".step", ".stp", ".x_t", ".x_b", ".iges", ".igs"}: raise ValueError("Unsupported benchmark geometry") if args.repeat < 1: raise ValueError("repeat must be positive") root = args.work_root.resolve() root.mkdir(parents=True, exist_ok=False) baseline = _mechanical_processes() report = { "started_at": datetime.now(timezone.utc).isoformat(), "geometry": str(geometry), "baseline_processes": baseline, "runs": [], } report_path = root / "acceptance-report.json" try: cancel_job = _stage(root, geometry, "cancel") report["cancellation"] = _run_cancellation( cancel_job, baseline, args.cancel_after, args.release_wait ) print("[OK] Process-tree cancellation completed and released.") for index in range(1, args.repeat + 1): job_dir = _stage(root, geometry, f"run-{index:02d}") report["runs"].append(_run_success(job_dir, baseline, args.release_wait)) print(f"[OK] Acceptance solve {index}/{args.repeat} completed and released.") report["passed"] = True except Exception as exc: report["passed"] = False report["failure"] = {"type": type(exc).__name__, "detail": str(exc)[:1000]} raise finally: report["completed_at"] = datetime.now(timezone.utc).isoformat() report_path.write_text( json.dumps(report, ensure_ascii=False, indent=2), encoding="utf-8" ) print(f"[OK] Acceptance passed. Report: {report_path}") return 0 if __name__ == "__main__": raise SystemExit(main())