309 lines
11 KiB
Python
309 lines
11 KiB
Python
|
from __future__ import annotations
|
||
|
|
|
||
|
|
import json
|
||
|
|
import os
|
||
|
|
import subprocess
|
||
|
|
from pathlib import Path
|
||
|
|
from typing import Iterable
|
||
|
|
|
||
|
|
|
||
|
|
ASITUS_RECOGNIZE_HOLES_ENV = "STEP_EDITOR_ASITUS_RECOGNIZE_HOLES"
|
||
|
|
ASITUS_DISABLE_ENV = "STEP_EDITOR_DISABLE_ASITUS"
|
||
|
|
ASITUS_TIMEOUT_ENV = "STEP_EDITOR_ASITUS_TIMEOUT"
|
||
|
|
|
||
|
|
|
||
|
|
def _project_root(project_root: Path | None = None) -> Path:
|
||
|
|
return project_root or Path(__file__).resolve().parent.parent
|
||
|
|
|
||
|
|
|
||
|
|
def default_asitus_recognize_holes_path(project_root: Path | None = None) -> Path | None:
|
||
|
|
env_path = os.environ.get(ASITUS_RECOGNIZE_HOLES_ENV, "").strip()
|
||
|
|
if env_path:
|
||
|
|
path = Path(env_path).expanduser()
|
||
|
|
return path if path.is_file() else None
|
||
|
|
if os.environ.get(ASITUS_DISABLE_ENV, "").strip().lower() in {"1", "true", "yes", "on"}:
|
||
|
|
return None
|
||
|
|
|
||
|
|
root = _project_root(project_root)
|
||
|
|
candidates = (
|
||
|
|
root / "third_party" / "asitus_probe_tools_build" / "Release" / "recognize_holes.exe",
|
||
|
|
root / "third_party" / "asitus_probe_tools_build" / "RelWithDebInfo" / "recognize_holes.exe",
|
||
|
|
root / "third_party" / "asitus_probe_tools_build" / "Debug" / "recognize_holes.exe",
|
||
|
|
root / "third_party" / "asitus_probe_build" / "Release" / "recognize_holes.exe",
|
||
|
|
root / "third_party" / "asitus_probe_build" / "RelWithDebInfo" / "recognize_holes.exe",
|
||
|
|
root / "third_party" / "asitus_probe_build" / "Debug" / "recognize_holes.exe",
|
||
|
|
)
|
||
|
|
for candidate in candidates:
|
||
|
|
if candidate.is_file():
|
||
|
|
return candidate
|
||
|
|
return None
|
||
|
|
|
||
|
|
|
||
|
|
def asitus_runtime_path_entries(project_root: Path | None = None) -> list[Path]:
|
||
|
|
root = _project_root(project_root)
|
||
|
|
third_party = root / "third_party" / "3rdparty"
|
||
|
|
candidates = (
|
||
|
|
root / "third_party" / "AnalysisSitus_build_algo_occt77" / "win64" / "vc14" / "bin",
|
||
|
|
third_party / "OCCT" / "win64" / "vc14" / "bin",
|
||
|
|
third_party / "freeimage-3.17.0-vc14-64" / "bin",
|
||
|
|
third_party / "freetype-2.5.5-vc14-64" / "bin",
|
||
|
|
third_party / "tbb_2021.5-vc14-64" / "bin",
|
||
|
|
third_party / "tcltk-86-64" / "bin",
|
||
|
|
third_party / "ffmpeg-3.3.4-64" / "bin",
|
||
|
|
third_party / "openvr-1.14.15-64" / "bin" / "win64",
|
||
|
|
third_party / "3rdparty-vc14-64" / "freeimage-3.18.0-x64" / "bin",
|
||
|
|
third_party / "3rdparty-vc14-64" / "freetype-2.13.3-x64" / "bin",
|
||
|
|
third_party / "3rdparty-vc14-64" / "tbb-2021.13.0-x64" / "bin",
|
||
|
|
third_party / "3rdparty-vc14-64" / "tcltk-8.6.15-x64" / "bin",
|
||
|
|
third_party / "3rdparty-vc14-64" / "ffmpeg-3.3.4-64" / "bin",
|
||
|
|
third_party / "3rdparty-vc14-64" / "openvr-1.14.15-64" / "bin" / "win64",
|
||
|
|
)
|
||
|
|
return [path for path in candidates if path.is_dir()]
|
||
|
|
|
||
|
|
|
||
|
|
def parse_asitus_hole_groups(payload: object) -> list[tuple[int, ...]]:
|
||
|
|
payload = _json_payload(payload)
|
||
|
|
if not isinstance(payload, dict):
|
||
|
|
return []
|
||
|
|
|
||
|
|
groups: list[tuple[int, ...]] = []
|
||
|
|
holes = payload.get("holes")
|
||
|
|
if isinstance(holes, list):
|
||
|
|
for item in holes:
|
||
|
|
if not isinstance(item, dict):
|
||
|
|
continue
|
||
|
|
group = _int_tuple(item.get("faceIds"))
|
||
|
|
if group:
|
||
|
|
groups.append(group)
|
||
|
|
if groups:
|
||
|
|
return _dedupe_groups(groups)
|
||
|
|
|
||
|
|
flat_ids = _int_tuple(payload.get("holeFaceIds"))
|
||
|
|
return [flat_ids] if flat_ids else []
|
||
|
|
|
||
|
|
|
||
|
|
def parse_asitus_probe_payload(payload: object) -> dict[str, object]:
|
||
|
|
data = _json_payload(payload)
|
||
|
|
if not isinstance(data, dict):
|
||
|
|
return {
|
||
|
|
"groups": (),
|
||
|
|
"faces": (),
|
||
|
|
"adjacency": (),
|
||
|
|
"geometric_relations": (),
|
||
|
|
"surface_summary": {},
|
||
|
|
"angle_summary": {},
|
||
|
|
"geometric_relation_summary": {},
|
||
|
|
"geometric_relation_mode": "",
|
||
|
|
}
|
||
|
|
|
||
|
|
faces: list[dict[str, object]] = []
|
||
|
|
raw_faces = data.get("faces")
|
||
|
|
if isinstance(raw_faces, list):
|
||
|
|
for item in raw_faces:
|
||
|
|
if not isinstance(item, dict):
|
||
|
|
continue
|
||
|
|
face_id = _int_or_none(item.get("id"))
|
||
|
|
if face_id is None:
|
||
|
|
continue
|
||
|
|
faces.append(
|
||
|
|
{
|
||
|
|
"id": face_id,
|
||
|
|
"surface": str(item.get("surface") or ""),
|
||
|
|
"neighbor_ids": _int_tuple(item.get("neighbors")),
|
||
|
|
}
|
||
|
|
)
|
||
|
|
|
||
|
|
adjacency: list[dict[str, object]] = []
|
||
|
|
raw_adjacency = data.get("adjacency")
|
||
|
|
if isinstance(raw_adjacency, list):
|
||
|
|
for item in raw_adjacency:
|
||
|
|
if not isinstance(item, dict):
|
||
|
|
continue
|
||
|
|
face_ids = _int_tuple(item.get("faceIds"))
|
||
|
|
if len(face_ids) != 2:
|
||
|
|
continue
|
||
|
|
adjacency.append(
|
||
|
|
{
|
||
|
|
"face_ids": face_ids,
|
||
|
|
"angle_type": str(item.get("angleType") or item.get("type") or ""),
|
||
|
|
"angle_rad": _float_or_none(item.get("angleRad")),
|
||
|
|
"edge_ids": _int_tuple(item.get("edgeIds")),
|
||
|
|
}
|
||
|
|
)
|
||
|
|
|
||
|
|
geometric_relations: list[dict[str, object]] = []
|
||
|
|
raw_geometric_relations = data.get("geometricRelations")
|
||
|
|
if isinstance(raw_geometric_relations, list):
|
||
|
|
for item in raw_geometric_relations:
|
||
|
|
if not isinstance(item, dict):
|
||
|
|
continue
|
||
|
|
face_ids = _int_tuple(item.get("faceIds"))
|
||
|
|
if len(face_ids) != 2:
|
||
|
|
continue
|
||
|
|
geometric_relations.append(
|
||
|
|
{
|
||
|
|
"face_ids": face_ids,
|
||
|
|
"relation_type": str(item.get("type") or item.get("relationType") or ""),
|
||
|
|
"residual": _float_or_none(item.get("residual")),
|
||
|
|
"source": str(item.get("source") or "analysis-situs-probe"),
|
||
|
|
}
|
||
|
|
)
|
||
|
|
|
||
|
|
return {
|
||
|
|
"groups": tuple(parse_asitus_hole_groups(data)),
|
||
|
|
"valid_brep": data.get("validBreP"),
|
||
|
|
"face_count": _int_or_none(data.get("faceCount")),
|
||
|
|
"aag_node_count": _int_or_none(data.get("aagNodeCount")),
|
||
|
|
"faces": tuple(faces),
|
||
|
|
"adjacency": tuple(adjacency),
|
||
|
|
"geometric_relations": tuple(geometric_relations),
|
||
|
|
"surface_summary": _str_int_dict(data.get("surfaceSummary")),
|
||
|
|
"angle_summary": _str_int_dict(data.get("angleSummary")),
|
||
|
|
"geometric_relation_summary": _str_int_dict(data.get("geometricRelationSummary")),
|
||
|
|
"geometric_relation_mode": str(data.get("geometricRelationMode") or ""),
|
||
|
|
}
|
||
|
|
|
||
|
|
|
||
|
|
def _json_payload(payload: object) -> object:
|
||
|
|
if not isinstance(payload, str):
|
||
|
|
return payload
|
||
|
|
text = payload.strip()
|
||
|
|
json_start = text.find("{")
|
||
|
|
if json_start > 0:
|
||
|
|
text = text[json_start:]
|
||
|
|
return json.loads(text)
|
||
|
|
|
||
|
|
|
||
|
|
def _int_or_none(value: object) -> int | None:
|
||
|
|
try:
|
||
|
|
return int(value)
|
||
|
|
except (TypeError, ValueError):
|
||
|
|
return None
|
||
|
|
|
||
|
|
|
||
|
|
def _float_or_none(value: object) -> float | None:
|
||
|
|
try:
|
||
|
|
return float(value)
|
||
|
|
except (TypeError, ValueError):
|
||
|
|
return None
|
||
|
|
|
||
|
|
|
||
|
|
def _str_int_dict(value: object) -> dict[str, int]:
|
||
|
|
if not isinstance(value, dict):
|
||
|
|
return {}
|
||
|
|
result: dict[str, int] = {}
|
||
|
|
for key, item in value.items():
|
||
|
|
try:
|
||
|
|
result[str(key)] = int(item)
|
||
|
|
except (TypeError, ValueError):
|
||
|
|
continue
|
||
|
|
return result
|
||
|
|
|
||
|
|
|
||
|
|
def _int_tuple(values: object) -> tuple[int, ...]:
|
||
|
|
if values is None:
|
||
|
|
return ()
|
||
|
|
if isinstance(values, (str, bytes)):
|
||
|
|
return ()
|
||
|
|
try:
|
||
|
|
items = list(values) # type: ignore[arg-type]
|
||
|
|
except TypeError:
|
||
|
|
return ()
|
||
|
|
result: list[int] = []
|
||
|
|
for item in items:
|
||
|
|
try:
|
||
|
|
result.append(int(item))
|
||
|
|
except (TypeError, ValueError):
|
||
|
|
continue
|
||
|
|
return tuple(sorted(set(result)))
|
||
|
|
|
||
|
|
|
||
|
|
def _dedupe_groups(groups: Iterable[tuple[int, ...]]) -> list[tuple[int, ...]]:
|
||
|
|
result: list[tuple[int, ...]] = []
|
||
|
|
seen: set[tuple[int, ...]] = set()
|
||
|
|
for group in groups:
|
||
|
|
if not group or group in seen:
|
||
|
|
continue
|
||
|
|
seen.add(group)
|
||
|
|
result.append(group)
|
||
|
|
return result
|
||
|
|
|
||
|
|
|
||
|
|
def run_asitus_hole_recognition(
|
||
|
|
step_path: str | Path,
|
||
|
|
*,
|
||
|
|
cli_path: str | Path | None = None,
|
||
|
|
project_root: Path | None = None,
|
||
|
|
timeout_seconds: float | None = None,
|
||
|
|
) -> dict[str, object]:
|
||
|
|
source = Path(step_path).expanduser()
|
||
|
|
if not source.is_file():
|
||
|
|
return {"ok": False, "reason": "missing-step", "groups": (), "message": f"STEP file not found: {source}"}
|
||
|
|
|
||
|
|
cli = Path(cli_path).expanduser() if cli_path else default_asitus_recognize_holes_path(project_root)
|
||
|
|
if cli is None or not cli.is_file():
|
||
|
|
return {"ok": False, "reason": "missing-cli", "groups": (), "message": "Analysis Situs recognize_holes CLI is not available."}
|
||
|
|
|
||
|
|
if timeout_seconds is None:
|
||
|
|
try:
|
||
|
|
timeout_seconds = float(os.environ.get(ASITUS_TIMEOUT_ENV, "") or 3.0)
|
||
|
|
except ValueError:
|
||
|
|
timeout_seconds = 3.0
|
||
|
|
|
||
|
|
env = os.environ.copy()
|
||
|
|
path_entries = [str(path) for path in asitus_runtime_path_entries(project_root)]
|
||
|
|
env["PATH"] = os.pathsep.join([*path_entries, env.get("PATH", "")])
|
||
|
|
creationflags = getattr(subprocess, "CREATE_NO_WINDOW", 0)
|
||
|
|
try:
|
||
|
|
completed = subprocess.run(
|
||
|
|
[str(cli), str(source)],
|
||
|
|
cwd=str(_project_root(project_root)),
|
||
|
|
env=env,
|
||
|
|
capture_output=True,
|
||
|
|
text=True,
|
||
|
|
encoding="utf-8",
|
||
|
|
errors="replace",
|
||
|
|
timeout=max(float(timeout_seconds), 0.1),
|
||
|
|
creationflags=creationflags,
|
||
|
|
check=False,
|
||
|
|
)
|
||
|
|
except subprocess.TimeoutExpired:
|
||
|
|
return {"ok": False, "reason": "timeout", "groups": (), "message": "Analysis Situs hole recognition timed out."}
|
||
|
|
except OSError as exc:
|
||
|
|
return {"ok": False, "reason": "launch-failed", "groups": (), "message": str(exc)}
|
||
|
|
|
||
|
|
if completed.returncode != 0:
|
||
|
|
message = (completed.stderr or completed.stdout or "").strip()
|
||
|
|
return {
|
||
|
|
"ok": False,
|
||
|
|
"reason": "recognizer-failed",
|
||
|
|
"returncode": completed.returncode,
|
||
|
|
"groups": (),
|
||
|
|
"message": message,
|
||
|
|
}
|
||
|
|
|
||
|
|
try:
|
||
|
|
parsed = parse_asitus_probe_payload(completed.stdout)
|
||
|
|
except (json.JSONDecodeError, TypeError, ValueError) as exc:
|
||
|
|
return {"ok": False, "reason": "bad-json", "groups": (), "message": str(exc), "stdout": completed.stdout}
|
||
|
|
groups = tuple(parsed.get("groups", ()))
|
||
|
|
|
||
|
|
return {
|
||
|
|
"ok": True,
|
||
|
|
"reason": "ok",
|
||
|
|
"groups": groups,
|
||
|
|
"hole_count": len(groups),
|
||
|
|
"valid_brep": parsed.get("valid_brep"),
|
||
|
|
"face_count": parsed.get("face_count"),
|
||
|
|
"aag_node_count": parsed.get("aag_node_count"),
|
||
|
|
"faces": tuple(parsed.get("faces", ())),
|
||
|
|
"adjacency": tuple(parsed.get("adjacency", ())),
|
||
|
|
"geometric_relations": tuple(parsed.get("geometric_relations", ())),
|
||
|
|
"surface_summary": dict(parsed.get("surface_summary", {}) or {}),
|
||
|
|
"angle_summary": dict(parsed.get("angle_summary", {}) or {}),
|
||
|
|
"geometric_relation_summary": dict(parsed.get("geometric_relation_summary", {}) or {}),
|
||
|
|
"geometric_relation_mode": str(parsed.get("geometric_relation_mode") or ""),
|
||
|
|
"cli": str(cli),
|
||
|
|
"message": f"Analysis Situs recognized {len(groups)} hole groups.",
|
||
|
|
}
|