125 lines
4.2 KiB
Python
125 lines
4.2 KiB
Python
"""Guarded ANSYS Mechanical 2024 R2 adapter bootstrap.
|
|
|
|
The first implementation phase validates the fixed declarative job protocol and
|
|
probes the managed v242 runtime. Execution remains mechanically disabled until a
|
|
dedicated Windows node passes the real-license acceptance suite. The request can
|
|
never provide Python, APDL, journal text, executables, URLs, or local paths.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import math
|
|
import os
|
|
import sys
|
|
from importlib.metadata import PackageNotFoundError, version
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
ADAPTER_VERSION = "0.1.0"
|
|
ANSYS_REVISION = 242
|
|
ANSYS_RELEASE = "2024 R2"
|
|
EXECUTION_GATE = "ZCBOT_ANSYS_242_VALIDATED"
|
|
INSTALL_ROOT = Path(os.environ.get("AWP_ROOT242", r"C:\Program Files\ANSYS Inc\v242"))
|
|
|
|
|
|
def _probe() -> int:
|
|
health = "unavailable"
|
|
software_version = None
|
|
details: list[str] = []
|
|
try:
|
|
if sys.platform != "win32":
|
|
raise RuntimeError("ANSYS adapter requires Windows")
|
|
if not INSTALL_ROOT.is_dir():
|
|
raise RuntimeError("ANSYS Mechanical 2024 R2 installation was not found")
|
|
from ansys.mechanical.core import App # noqa: F401
|
|
|
|
package_version = version("ansys-mechanical-core")
|
|
software_version = ANSYS_RELEASE
|
|
details.append(f"PyMechanical {package_version} and v242 installation detected")
|
|
if os.environ.get(EXECUTION_GATE) != "1":
|
|
details.append("real-license acceptance is pending")
|
|
else:
|
|
health = "ready"
|
|
details.append("real-license acceptance gate is enabled")
|
|
except (
|
|
FileNotFoundError,
|
|
ImportError,
|
|
OSError,
|
|
PackageNotFoundError,
|
|
RuntimeError,
|
|
) as exc:
|
|
details = [str(exc)]
|
|
print(
|
|
json.dumps(
|
|
{
|
|
"adapter_version": ADAPTER_VERSION,
|
|
"software": "ANSYS Mechanical",
|
|
"software_version": software_version,
|
|
"health": health,
|
|
"detail": "; ".join(details),
|
|
},
|
|
ensure_ascii=False,
|
|
)
|
|
)
|
|
return 0
|
|
|
|
|
|
def _validate_semantics(request: dict[str, Any]) -> None:
|
|
analysis = request["operation"]["analysis"]
|
|
input_keys = {item["key"] for item in request["inputs"]}
|
|
if analysis["geometry"]["input"] not in input_keys:
|
|
raise ValueError("GEOMETRY_INPUT_NOT_BOUND")
|
|
|
|
selections = [
|
|
item["named_selection"]
|
|
for item in analysis["boundary_conditions"] + analysis["loads"]
|
|
]
|
|
if len({item.casefold() for item in selections}) != len(selections):
|
|
raise ValueError("NAMED_SELECTION_TARGETS_MUST_BE_UNIQUE")
|
|
|
|
for load in analysis["loads"]:
|
|
values = load.get("components", [load.get("magnitude")])
|
|
if not all(isinstance(item, (int, float)) and math.isfinite(item) for item in values):
|
|
raise ValueError("LOAD_VALUES_MUST_BE_FINITE")
|
|
if load["type"] == "force" and not any(item != 0 for item in values):
|
|
raise ValueError("FORCE_VECTOR_MUST_BE_NONZERO")
|
|
if load["type"] == "pressure" and load["magnitude"] == 0:
|
|
raise ValueError("PRESSURE_MUST_BE_NONZERO")
|
|
|
|
output_keys = [item["key"] for item in request["outputs"]]
|
|
if len(set(output_keys)) != len(output_keys):
|
|
raise ValueError("OUTPUT_KEYS_MUST_BE_UNIQUE")
|
|
|
|
|
|
def _load_request(job_dir: Path) -> dict[str, Any]:
|
|
record = json.loads(
|
|
(job_dir / "request" / "request.json").read_text(encoding="utf-8")
|
|
)
|
|
request = record.get("request")
|
|
if not isinstance(request, dict):
|
|
raise ValueError("JOB_REQUEST_MISSING")
|
|
_validate_semantics(request)
|
|
return record
|
|
|
|
|
|
def main() -> int:
|
|
if sys.argv[1:] == ["--probe"]:
|
|
return _probe()
|
|
if len(sys.argv) != 2:
|
|
print("[ERR] Usage: worker.py <job-directory>", file=sys.stderr)
|
|
return 2
|
|
|
|
try:
|
|
_load_request(Path(sys.argv[1]))
|
|
if os.environ.get(EXECUTION_GATE) != "1":
|
|
raise RuntimeError("ANSYS_242_REAL_LICENSE_ACCEPTANCE_REQUIRED")
|
|
raise RuntimeError("ANSYS_242_SOLVER_IMPLEMENTATION_PENDING")
|
|
except (KeyError, OSError, ValueError, json.JSONDecodeError, RuntimeError) as exc:
|
|
print(f"[ERR] {exc}", file=sys.stderr)
|
|
return 1
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|