450 lines
16 KiB
Python
450 lines
16 KiB
Python
"""Shared, declarative geometry helpers for the ANSYS Mechanical adapters."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import hashlib
|
|
import json
|
|
import math
|
|
import os
|
|
import sys
|
|
from datetime import datetime, timezone
|
|
from importlib.metadata import PackageNotFoundError, version
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
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(adapter_version: str) -> 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 atomic_json(path: Path, value: Any) -> None:
|
|
temporary = path.with_name(path.name + ".tmp-" + os.urandom(8).hex())
|
|
try:
|
|
with temporary.open("w", encoding="utf-8", newline="\n") as handle:
|
|
json.dump(value, handle, ensure_ascii=False, indent=2)
|
|
handle.flush()
|
|
os.fsync(handle.fileno())
|
|
os.replace(temporary, path)
|
|
finally:
|
|
temporary.unlink(missing_ok=True)
|
|
|
|
|
|
def file_sha256(path: Path) -> str:
|
|
digest = hashlib.sha256()
|
|
with path.open("rb") as handle:
|
|
for chunk in iter(lambda: handle.read(1024 * 1024), b""):
|
|
digest.update(chunk)
|
|
return digest.hexdigest()
|
|
|
|
|
|
def input_file(job_dir: Path, key: str) -> Path:
|
|
directory = job_dir / "input" / key
|
|
files = [
|
|
path for path in directory.iterdir()
|
|
if path.is_file() and not path.name.startswith(".")
|
|
]
|
|
if len(files) != 1:
|
|
raise ValueError(f"INPUT_FILE_COUNT_INVALID:{key}")
|
|
return files[0]
|
|
|
|
|
|
def artifact_manifest(path: Path, artifacts: dict[str, tuple[str, str]]) -> dict[str, Any]:
|
|
artifact_id, media_type = artifacts[path.name]
|
|
return {
|
|
"artifact_id": artifact_id,
|
|
"filename": path.name,
|
|
"media_type": media_type,
|
|
"size_bytes": path.stat().st_size,
|
|
"sha256": file_sha256(path),
|
|
}
|
|
|
|
|
|
def validate_artifact(path: Path) -> None:
|
|
if not path.is_file() or path.stat().st_size == 0:
|
|
raise RuntimeError(f"OUTPUT_EXPORT_EMPTY:{path.name}")
|
|
if path.suffix.lower() == ".png" and path.read_bytes()[:8] != b"\x89PNG\r\n\x1a\n":
|
|
raise RuntimeError(f"PNG_EXPORT_INVALID:{path.name}")
|
|
if path.suffix.lower() == ".mechdat" and path.stat().st_size < 64:
|
|
raise RuntimeError("MECHDAT_EXPORT_INVALID")
|
|
|
|
|
|
def terminal(
|
|
record: dict[str, Any],
|
|
status: str,
|
|
artifacts: list[dict[str, Any]],
|
|
error: dict[str, str],
|
|
) -> dict[str, Any]:
|
|
return {
|
|
"job_id": record.get("job_id", ""),
|
|
"lease_id": record.get("lease_id", ""),
|
|
"request_digest": record.get("request_digest", ""),
|
|
"status": status,
|
|
"error": error,
|
|
"artifact_manifest": artifacts,
|
|
"terminal_at": datetime.now(timezone.utc).isoformat(),
|
|
}
|
|
|
|
|
|
def import_geometry(app: Any, api: dict[str, Any], geometry: Path) -> Any:
|
|
model = app.Model
|
|
geometry_import = model.GeometryImportGroup.AddGeometryImport()
|
|
preferences = api["Ansys"].ACT.Mechanical.Utilities.GeometryImportPreferences()
|
|
preferences.ProcessNamedSelections = True
|
|
preferences.NamedSelectionKey = ""
|
|
preferences.ProcessMaterialProperties = False
|
|
preferences.ProcessCoordinateSystems = False
|
|
geometry_import.Import(
|
|
str(geometry),
|
|
api["Ansys"].Mechanical.DataModel.Enums.GeometryImportPreference.Format.Automatic,
|
|
preferences,
|
|
)
|
|
return model
|
|
|
|
|
|
def _xyz(value: Any) -> list[float]:
|
|
try:
|
|
result = [float(value[index]) for index in range(3)]
|
|
except (IndexError, KeyError, TypeError):
|
|
names = ("X", "Y", "Z") if hasattr(value, "X") else ("x", "y", "z")
|
|
result = [float(getattr(value, name)) for name in names]
|
|
if not all(math.isfinite(item) for item in result):
|
|
raise ValueError("GEOMETRY_COORDINATE_NOT_FINITE")
|
|
return result
|
|
|
|
|
|
def _normal(face: Any, centroid: list[float]) -> list[float] | None:
|
|
try:
|
|
uv = face.ParamAtPoint(centroid)
|
|
raw = face.NormalAtParam(float(uv[0]), float(uv[1]))
|
|
normal = _xyz(raw)
|
|
magnitude = math.sqrt(sum(item * item for item in normal))
|
|
if magnitude <= 1e-12:
|
|
return None
|
|
return [item / magnitude for item in normal]
|
|
except Exception: # noqa: BLE001 - vendor faces can reject a centroid parameterization
|
|
return None
|
|
|
|
|
|
def _surface_type(value: Any) -> str:
|
|
text = str(value or "unknown").split(".")[-1]
|
|
if text.casefold().startswith("geosurface"):
|
|
text = text[len("GeoSurface"):]
|
|
return text.casefold() or "unknown"
|
|
|
|
|
|
def _entity_id(value: Any) -> int:
|
|
return int(value.Id)
|
|
|
|
|
|
def _optional_number(value: Any, name: str) -> float | None:
|
|
try:
|
|
number = float(getattr(value, name))
|
|
except (AttributeError, TypeError, ValueError):
|
|
return None
|
|
return number if math.isfinite(number) else None
|
|
|
|
|
|
def geometry_inventory(app: Any) -> tuple[dict[str, Any], dict[int, Any]]:
|
|
geo_data = app.ExtAPI.DataModel.GeoData
|
|
unit = str(geo_data.Unit)
|
|
assemblies_out: list[dict[str, Any]] = []
|
|
bodies_out: list[dict[str, Any]] = []
|
|
faces_out: list[dict[str, Any]] = []
|
|
face_entities: dict[int, Any] = {}
|
|
points: list[list[float]] = []
|
|
|
|
for assembly_index, assembly in enumerate(geo_data.Assemblies):
|
|
assembly_name = str(getattr(assembly, "Name", f"assembly-{assembly_index + 1}"))
|
|
parts = getattr(assembly, "Parts", None) or getattr(assembly, "AllParts", ())
|
|
part_count = 0
|
|
for part_index, part in enumerate(parts):
|
|
part_count += 1
|
|
part_name = str(getattr(part, "Name", f"part-{part_index + 1}"))
|
|
for body_index, body in enumerate(part.Bodies):
|
|
body_id = _entity_id(body)
|
|
body_name = str(getattr(body, "Name", f"body-{body_index + 1}"))
|
|
body_face_ids: list[int] = []
|
|
for vertex in getattr(body, "Vertices", ()):
|
|
recorded = False
|
|
for attribute in ("Point", "Coordinates", "XYZ"):
|
|
if hasattr(vertex, attribute):
|
|
try:
|
|
points.append(_xyz(getattr(vertex, attribute)))
|
|
recorded = True
|
|
except (AttributeError, TypeError, ValueError):
|
|
pass
|
|
break
|
|
if not recorded:
|
|
try:
|
|
points.append(_xyz(vertex))
|
|
except (AttributeError, TypeError, ValueError):
|
|
pass
|
|
for face in body.Faces:
|
|
face_id = _entity_id(face)
|
|
centroid = _xyz(face.Centroid)
|
|
points.append(centroid)
|
|
body_face_ids.append(face_id)
|
|
face_entities[face_id] = face
|
|
faces_out.append(
|
|
{
|
|
"id": face_id,
|
|
"assembly": assembly_name,
|
|
"part": part_name,
|
|
"body_id": body_id,
|
|
"body": body_name,
|
|
"surface_type": _surface_type(getattr(face, "SurfaceType", None)),
|
|
"centroid": centroid,
|
|
"normal": _normal(face, centroid),
|
|
"area": _optional_number(face, "Area"),
|
|
}
|
|
)
|
|
bodies_out.append(
|
|
{
|
|
"id": body_id,
|
|
"assembly": assembly_name,
|
|
"part": part_name,
|
|
"name": body_name,
|
|
"volume": _optional_number(body, "Volume"),
|
|
"face_ids": body_face_ids,
|
|
}
|
|
)
|
|
assemblies_out.append({"name": assembly_name, "part_count": part_count})
|
|
|
|
if not bodies_out:
|
|
raise RuntimeError("IMPORTED_GEOMETRY_HAS_NO_BODIES")
|
|
if not faces_out:
|
|
raise RuntimeError("IMPORTED_GEOMETRY_HAS_NO_FACES")
|
|
|
|
named_selections: list[dict[str, Any]] = []
|
|
container = app.Model.NamedSelections
|
|
for item in () if container is None else (container.Children or ()):
|
|
location = getattr(item, "Location", None)
|
|
ids = [] if location is None else [int(value) for value in (location.Ids or ())]
|
|
named_selections.append({"name": str(item.Name), "entity_ids": ids})
|
|
|
|
bounds = {
|
|
"min": [min(point[index] for point in points) for index in range(3)],
|
|
"max": [max(point[index] for point in points) for index in range(3)],
|
|
}
|
|
return (
|
|
{
|
|
"geometry_unit": unit,
|
|
"counts": {
|
|
"assemblies": len(assemblies_out),
|
|
"bodies": len(bodies_out),
|
|
"faces": len(faces_out),
|
|
"named_selections": len(named_selections),
|
|
},
|
|
"bounding_box": bounds,
|
|
"assemblies": assemblies_out,
|
|
"bodies": bodies_out,
|
|
"faces": faces_out,
|
|
"named_selections": named_selections,
|
|
},
|
|
face_entities,
|
|
)
|
|
|
|
|
|
def export_geometry_image(
|
|
app: Any,
|
|
api: dict[str, Any],
|
|
path: Path,
|
|
selection: Any | None = None,
|
|
) -> None:
|
|
if selection is not None:
|
|
app.ExtAPI.SelectionManager.NewSelection(selection)
|
|
app.Graphics.Camera.SetFit()
|
|
settings = api["Ansys"].Mechanical.Graphics.GraphicsImageExportSettings()
|
|
settings.Resolution = api["GraphicsResolutionType"].EnhancedResolution
|
|
settings.Width = 1600
|
|
settings.Height = 1200
|
|
app.Graphics.ExportImage(str(path), api["GraphicsImageExportFormat"].PNG, settings)
|
|
validate_artifact(path)
|
|
if selection is not None:
|
|
app.ExtAPI.SelectionManager.ClearSelection()
|
|
|
|
|
|
def _convert_length(api: dict[str, Any], value: float, source_unit: str, target_unit: str) -> float:
|
|
if source_unit.casefold() == target_unit.casefold():
|
|
return float(value)
|
|
quantity = api["Quantity"](float(value), source_unit)
|
|
converted = quantity.ConvertUnit(target_unit)
|
|
return float((converted if converted is not None else quantity).Value)
|
|
|
|
|
|
def _expected_count(scope: dict[str, Any], face_ids: list[int]) -> None:
|
|
expected = scope.get("expected_count")
|
|
if expected is not None and len(face_ids) != expected:
|
|
raise RuntimeError(
|
|
f"GEOMETRY_SCOPE_COUNT_MISMATCH:expected={expected},actual={len(face_ids)}"
|
|
)
|
|
|
|
|
|
def _axis_index(axis: str) -> int:
|
|
return {"x": 0, "y": 1, "z": 2}[axis]
|
|
|
|
|
|
def _planar_faces(inventory: dict[str, Any]) -> list[dict[str, Any]]:
|
|
return [
|
|
face for face in inventory["faces"]
|
|
if "plane" in face["surface_type"] and face["normal"] is not None
|
|
]
|
|
|
|
|
|
def _scope_face_ids(
|
|
api: dict[str, Any],
|
|
inventory: dict[str, Any],
|
|
scope: dict[str, Any],
|
|
request_length_unit: str,
|
|
) -> list[int]:
|
|
planar = _planar_faces(inventory)
|
|
if not planar:
|
|
raise RuntimeError("GEOMETRY_SCOPE_NO_PLANAR_FACES")
|
|
geo_unit = inventory["geometry_unit"]
|
|
angle = math.radians(float(scope.get("angle_tolerance_degrees", 5.0)))
|
|
cosine = math.cos(angle)
|
|
|
|
if scope["type"] == "extreme_face":
|
|
axis = _axis_index(scope["axis"])
|
|
aligned = [face for face in planar if abs(face["normal"][axis]) >= cosine]
|
|
if not aligned:
|
|
raise RuntimeError("GEOMETRY_SCOPE_NO_AXIS_ALIGNED_FACES")
|
|
coordinates = [face["centroid"][axis] for face in aligned]
|
|
extreme = min(coordinates) if scope["side"] == "min" else max(coordinates)
|
|
span = inventory["bounding_box"]["max"][axis] - inventory["bounding_box"]["min"][axis]
|
|
tolerance = scope.get("tolerance")
|
|
tolerance = (
|
|
max(abs(span) * 1e-7, 1e-12)
|
|
if tolerance is None
|
|
else _convert_length(api, tolerance, request_length_unit, geo_unit)
|
|
)
|
|
ids = [
|
|
face["id"] for face in aligned
|
|
if abs(face["centroid"][axis] - extreme) <= tolerance
|
|
]
|
|
elif scope["type"] == "planar_faces":
|
|
raw_normal = [float(item) for item in scope["normal"]]
|
|
magnitude = math.sqrt(sum(item * item for item in raw_normal))
|
|
if magnitude <= 1e-12:
|
|
raise ValueError("PLANE_NORMAL_MUST_BE_NONZERO")
|
|
normal = [item / magnitude for item in raw_normal]
|
|
offset = _convert_length(api, scope["offset"], request_length_unit, geo_unit)
|
|
tolerance = _convert_length(api, scope["tolerance"], request_length_unit, geo_unit)
|
|
ids = []
|
|
for face in planar:
|
|
alignment = abs(sum(a * b for a, b in zip(normal, face["normal"], strict=True)))
|
|
position = sum(a * b for a, b in zip(normal, face["centroid"], strict=True))
|
|
if alignment >= cosine and abs(position - offset) <= tolerance:
|
|
ids.append(face["id"])
|
|
else:
|
|
raise ValueError("GEOMETRY_SCOPE_TYPE_UNSUPPORTED")
|
|
|
|
ids = sorted(set(ids))
|
|
if not ids:
|
|
raise RuntimeError(f"GEOMETRY_SCOPE_EMPTY:{scope['type']}")
|
|
_expected_count(scope, ids)
|
|
return ids
|
|
|
|
|
|
def resolve_scope(
|
|
app: Any,
|
|
api: dict[str, Any],
|
|
inventory: dict[str, Any],
|
|
face_entities: dict[int, Any],
|
|
scope: dict[str, Any],
|
|
request_length_unit: str,
|
|
generated_name: str,
|
|
) -> dict[str, Any]:
|
|
if scope["type"] == "named_selection":
|
|
matches = [
|
|
item for item in (() if app.Model.NamedSelections is None else app.Model.NamedSelections.Children)
|
|
if str(item.Name).casefold() == scope["name"].casefold()
|
|
]
|
|
if len(matches) != 1:
|
|
code = "NOT_FOUND" if not matches else "AMBIGUOUS"
|
|
raise RuntimeError(f"NAMED_SELECTION_{code}:{scope['name']}")
|
|
location = matches[0].Location
|
|
ids = [int(value) for value in (location.Ids or ())]
|
|
if not ids:
|
|
raise RuntimeError(f"NAMED_SELECTION_EMPTY:{scope['name']}")
|
|
_expected_count(scope, ids)
|
|
return {
|
|
"name": str(matches[0].Name),
|
|
"source": "imported_named_selection",
|
|
"scope": scope,
|
|
"entity_ids": ids,
|
|
"location": matches[0],
|
|
}
|
|
|
|
face_ids = _scope_face_ids(api, inventory, scope, request_length_unit)
|
|
if any(face_id not in face_entities for face_id in face_ids):
|
|
raise RuntimeError("GEOMETRY_SCOPE_ENTITY_NOT_FOUND")
|
|
conflicts = [
|
|
item for item in (() if app.Model.NamedSelections is None else app.Model.NamedSelections.Children)
|
|
if str(item.Name).casefold() == generated_name.casefold()
|
|
]
|
|
if conflicts:
|
|
raise RuntimeError(f"GENERATED_SELECTION_NAME_CONFLICT:{generated_name}")
|
|
location = app.ExtAPI.SelectionManager.CreateSelectionInfo(
|
|
api["SelectionTypeEnum"].GeometryEntities
|
|
)
|
|
location.Ids = face_ids
|
|
named_selection = app.Model.AddNamedSelection()
|
|
named_selection.Name = generated_name
|
|
named_selection.Location = location
|
|
return {
|
|
"name": generated_name,
|
|
"source": "geometry_query",
|
|
"scope": scope,
|
|
"entity_ids": face_ids,
|
|
"location": named_selection,
|
|
}
|
|
|
|
|
|
def selected_face_records(inventory: dict[str, Any], ids: list[int]) -> list[dict[str, Any]]:
|
|
selected = set(ids)
|
|
return [face for face in inventory["faces"] if face["id"] in selected]
|