Files
pythonocc-step-editor/scripts/verify_isolated_face_edit.py
T

629 lines
27 KiB
Python

from __future__ import annotations
import json
import subprocess
import sys
import tempfile
from pathlib import Path
from OCC.Core.BRepAlgoAPI import BRepAlgoAPI_Cut
from OCC.Core.BRepPrimAPI import BRepPrimAPI_MakeBox, BRepPrimAPI_MakeCylinder, BRepPrimAPI_MakeSphere, BRepPrimAPI_MakeTorus
from OCC.Core.gp import gp_Ax2, gp_Dir, gp_Pnt
PROJECT_ROOT = Path(__file__).resolve().parent.parent
DEFAULT_MODEL = PROJECT_ROOT / "assets" / "models" / "cube_10mm.step"
if str(PROJECT_ROOT) not in sys.path:
sys.path.insert(0, str(PROJECT_ROOT))
from step_editor.model import StepModel
from step_editor.geometry_utils import _finalize_boolean_result
from step_editor.step_io import _write_step
def _run_worker_case(
*,
label: str,
operation: str,
args: list[object],
validator,
input_path: Path = DEFAULT_MODEL,
):
with tempfile.TemporaryDirectory(prefix="geom_param_isolated_face_verify_") as temp_dir:
temp_root = Path(temp_dir)
output_path = temp_root / "output.step"
request_path = temp_root / "request.json"
request_path.write_text(
json.dumps(
{
"input_path": str(input_path),
"output_path": str(output_path),
"operation": operation,
"args": args,
},
ensure_ascii=False,
indent=2,
),
encoding="utf-8",
)
completed = subprocess.run(
[sys.executable, "-m", "step_editor.isolated_edit_worker", str(request_path)],
cwd=PROJECT_ROOT,
capture_output=True,
text=True,
encoding="utf-8",
errors="replace",
timeout=120,
check=False,
)
response_path = request_path.with_suffix(".response.json")
if completed.returncode != 0:
detail = response_path.read_text(encoding="utf-8") if response_path.exists() else completed.stderr
raise SystemExit(f"{label}: isolated worker failed with code {completed.returncode}: {detail}")
response = json.loads(response_path.read_text(encoding="utf-8"))
if not response.get("ok"):
raise SystemExit(f"{label}: isolated worker returned failure: {response}")
if not output_path.exists():
raise SystemExit(f"{label}: isolated worker did not produce output STEP")
model = StepModel.load(output_path)
stats = model.stats()
if stats.solids != 1:
raise SystemExit(f"{label}: isolated Face edit changed solid count unexpectedly: {stats}")
validator(label, model)
print(f"isolated Face edit ok: {label}")
print(str(response.get("message", "")).encode("ascii", "backslashreplace").decode("ascii"))
return model
def _run_main_worker_entry_case() -> None:
with tempfile.TemporaryDirectory(prefix="geom_param_main_worker_entry_") as temp_dir:
temp_root = Path(temp_dir)
output_path = temp_root / "output.step"
request_path = temp_root / "request.json"
request_path.write_text(
json.dumps(
{
"input_path": str(DEFAULT_MODEL),
"output_path": str(output_path),
"operation": "resize_face_area_local",
"args": [0, 144.0],
},
ensure_ascii=False,
indent=2,
),
encoding="utf-8",
)
completed = subprocess.run(
[sys.executable, "main.py", "--isolated-edit-worker", str(request_path)],
cwd=PROJECT_ROOT,
capture_output=True,
text=True,
encoding="utf-8",
errors="replace",
timeout=120,
check=False,
)
response_path = request_path.with_suffix(".response.json")
if completed.returncode != 0:
detail = response_path.read_text(encoding="utf-8") if response_path.exists() else completed.stderr
raise SystemExit(f"main worker entry failed with code {completed.returncode}: {detail}")
response = json.loads(response_path.read_text(encoding="utf-8"))
if not response.get("ok") or not output_path.exists():
raise SystemExit(f"main worker entry did not produce an edited STEP: {response}")
model = StepModel.load(output_path)
_assert_face_area("main worker entry", model, 144.0)
print("main worker entry ok")
def _float_close(value: object, target: float, tolerance: float = 1e-5) -> bool:
try:
return abs(float(value) - target) <= tolerance
except (TypeError, ValueError):
return False
def _logical_region_has_width(model: StepModel, logical_id: int, target_width: float) -> bool:
for face_id in model.face_ids_for_logical_id(logical_id):
info = model.face_info(face_id)
width = info.get("local_face_width")
if _float_close(width, target_width, tolerance=1e-4):
return True
return False
def _triple_close(value: object, target: tuple[float, float, float], tolerance: float = 1e-5) -> bool:
if not isinstance(value, (list, tuple)) or len(value) != 3:
return False
return all(abs(float(value[index]) - target[index]) <= tolerance for index in range(3))
def _assert_plane_position(label: str, model, target_position: float = 10.0) -> None:
matches: list[tuple[int, float]] = []
for face_id in range(len(model.faces)):
info = model.face_info(face_id)
if info.get("surface") != "plane":
continue
frame = model.face_plane_offset_frame(face_id)
if frame is None:
continue
_origin, _direction, position = frame
if abs(abs(float(position)) - target_position) <= 1e-5:
matches.append((face_id, float(position)))
if not matches:
raise SystemExit(f"{label}: output does not contain a plane at target position {target_position}")
print(f"matched_planes={matches}")
def _assert_face_area(label: str, model, target_area: float = 225.0) -> None:
matches = [
(face_id, float(info.get("area")))
for face_id in range(len(model.faces))
for info in (model.face_info(face_id),)
if _float_close(info.get("area"), target_area, tolerance=1e-4)
]
if not matches:
raise SystemExit(f"{label}: output does not contain a Face with area {target_area}")
print(f"matched_areas={matches}")
def _assert_face_size_axis(label: str, model, axis_key: str, target_size: float = 25.0) -> None:
matches = []
for face_id in range(len(model.faces)):
info = model.face_info(face_id)
if _float_close(info.get(axis_key), target_size, tolerance=1e-4):
matches.append((face_id, axis_key, float(info[axis_key])))
if not matches:
raise SystemExit(f"{label}: output does not contain a Face {axis_key} of {target_size}")
print(f"matched_sizes={matches}")
def _assert_face_width(label: str, model, target_size: float = 25.0) -> None:
_assert_face_size_axis(label, model, "local_face_width", target_size)
def _assert_face_height(label: str, model, target_size: float = 25.0) -> None:
_assert_face_size_axis(label, model, "local_face_height", target_size)
def _assert_face_center(label: str, model, target_center: tuple[float, float, float] = (15.0, 5.0, 0.0)) -> None:
matches = [
(face_id, info.get("area_center") or info.get("bbox_center"))
for face_id in range(len(model.faces))
for info in (model.face_info(face_id),)
if _triple_close(info.get("area_center") or info.get("bbox_center"), target_center, tolerance=1e-4)
]
if not matches:
raise SystemExit(f"{label}: output does not contain a Face centered at {target_center}")
print(f"matched_centers={matches}")
def _write_shell_plate(path: Path) -> None:
_write_step(BRepPrimAPI_MakeBox(30.0, 20.0, 2.0).Shape(), path)
def _write_sphere_model(path: Path) -> None:
_write_step(BRepPrimAPI_MakeSphere(5.0).Shape(), path)
def _write_torus_model(path: Path) -> None:
_write_step(BRepPrimAPI_MakeTorus(8.0, 2.0).Shape(), path)
def _write_hollow_cylinder_model(path: Path) -> None:
outer = BRepPrimAPI_MakeCylinder(8.0, 78.0).Shape()
inner = BRepPrimAPI_MakeCylinder(
gp_Ax2(gp_Pnt(0.0, 0.0, -1.0), gp_Dir(0.0, 0.0, 1.0)),
3.0,
80.0,
).Shape()
cut = BRepAlgoAPI_Cut(outer, inner)
_write_step(_finalize_boolean_result(cut, "verify isolated hollow cylinder cut", use_glue=False), path)
def _first_face_by_surface(model: StepModel, surface: str) -> int:
for face_id in range(len(model.faces)):
if model.face_info(face_id).get("surface") == surface:
return face_id
raise SystemExit(f"no {surface} Face was recognized")
def _top_planar_face(model: StepModel) -> int:
best: tuple[float, int] | None = None
for face_id in range(len(model.faces)):
info = model.face_info(face_id)
if info.get("surface") != "plane":
continue
center = info.get("area_center") or info.get("bbox_center")
if not isinstance(center, tuple) or len(center) != 3:
continue
z_value = float(center[2])
if best is None or z_value > best[0]:
best = (z_value, face_id)
if best is None:
raise SystemExit("no top planar Face was recognized")
return best[1]
def _first_shell_face(model: StepModel, source_thickness: float = 2.0, tolerance: float = 1e-5) -> int:
candidates: list[tuple[int, int]] = []
for face_id in range(len(model.faces)):
info = model.feature_info(face_id)
if info.get("surface") != "plane":
continue
if info.get("shell_region_status") != "candidate":
continue
thickness = float(info.get("shell_thickness_estimate") or 0.0)
if abs(thickness - source_thickness) > tolerance:
continue
confidence_rank = {"high": 0, "medium": 1, "low": 2}.get(str(info.get("shell_confidence")), 3)
candidates.append((confidence_rank, face_id))
if not candidates:
raise SystemExit(f"no shell thickness candidate near {source_thickness:g}")
candidates.sort()
return candidates[0][1]
def _assert_shell_thickness(label: str, model: StepModel, target_thickness: float = 4.0) -> None:
size = tuple(float(value) for value in model.geometry_stats()["bbox_size"])
thickness = min(size)
if abs(thickness - target_thickness) > 1e-4:
raise SystemExit(f"{label}: output thickness should be {target_thickness:g}, got {thickness:g}")
print(f"matched_shell_thickness={thickness:g}, bbox_size={size}")
def _assert_sphere_radius(label: str, model: StepModel, target_radius: float = 10.0) -> None:
matches = []
for face_id in range(len(model.faces)):
info = model.face_info(face_id)
if info.get("surface") != "sphere":
continue
radius = info.get("radius")
if _float_close(radius, target_radius, tolerance=1e-4):
matches.append((face_id, float(radius)))
if not matches:
raise SystemExit(f"{label}: output does not contain a sphere Face radius {target_radius:g}")
print(f"matched_sphere_radii={matches}")
def _assert_torus_minor_radius(label: str, model: StepModel, target_minor: float = 4.0) -> None:
matches = []
for face_id in range(len(model.faces)):
info = model.face_info(face_id)
if info.get("surface") != "torus":
continue
major = info.get("major_radius")
minor = info.get("minor_radius")
if _float_close(minor, target_minor, tolerance=1e-4):
matches.append((face_id, float(major), float(minor)))
if not matches:
raise SystemExit(f"{label}: output does not contain a torus Face minor radius {target_minor:g}")
print(f"matched_torus_radii={matches}")
def _assert_hollow_cylinder_height(label: str, model: StepModel, target_height: float) -> None:
bbox_size = model.geometry_stats().get("bbox_size")
if not isinstance(bbox_size, tuple) or len(bbox_size) != 3:
raise SystemExit(f"{label}: output bbox_size is missing")
height = float(bbox_size[2])
if abs(height - target_height) > 1e-4:
raise SystemExit(f"{label}: hollow cylinder height should be {target_height:g}, got {height:g}")
radii: list[float] = []
for face_id in range(len(model.faces)):
info = model.face_info(face_id)
if info.get("surface") != "cylinder":
continue
radius = float(info.get("radius") or 0.0)
if radius <= 0.0:
continue
if not any(abs(radius - existing) <= max(radius, existing, 1.0) * 1e-5 for existing in radii):
radii.append(radius)
radii.sort()
if len(radii) != 2 or abs(radii[0] - 3.0) > 1e-4 or abs(radii[1] - 8.0) > 1e-4:
raise SystemExit(f"{label}: hollow cylinder should preserve inner/outer radii, got {radii}")
print(f"matched_hollow_cylinder_height={height:g}, radii={radii}")
def _assert_hollow_cylinder_extension(label: str, model: StepModel) -> None:
_assert_hollow_cylinder_height(label, model, 167.0)
def main() -> int:
push_pull_probe = StepModel.load(DEFAULT_MODEL)
large_push_plan = push_pull_probe.push_pull_plan(0, 11.0)
if large_push_plan.get("status") == "blocked" or str(large_push_plan.get("risk")) != "high":
raise SystemExit(f"large push/pull should be allowed as an isolated high-risk edit: {large_push_plan}")
_run_worker_case(
label="large push/pull current Face",
operation="push_pull_face",
args=[0, 11.0],
validator=lambda label, model: _assert_plane_position(label, model, 11.0),
)
with tempfile.TemporaryDirectory(prefix="geom_param_isolated_hollow_cylinder_verify_") as temp_dir:
hollow_path = Path(temp_dir) / "hollow_cylinder.step"
_write_hollow_cylinder_model(hollow_path)
hollow_probe = StepModel.load(hollow_path)
hollow_face_id = _top_planar_face(hollow_probe)
hollow_plan = hollow_probe.push_pull_plan(hollow_face_id, 89.0)
if hollow_plan.get("cylindrical_cap_extension_kind") != "coaxial-tube":
raise SystemExit(f"hollow cylinder cap should use coaxial-tube plan in worker test: {hollow_plan}")
_run_worker_case(
label="large hollow cylinder cap push/pull",
operation="push_pull_face",
args=[hollow_face_id, 89.0],
validator=_assert_hollow_cylinder_extension,
input_path=hollow_path,
)
_run_worker_case(
label="hollow cylinder cap inward push/pull",
operation="push_pull_face",
args=[hollow_face_id, -39.0],
validator=lambda label, model: _assert_hollow_cylinder_height(label, model, 39.0),
input_path=hollow_path,
)
hollow_height_probe = StepModel.load(hollow_path)
hollow_cylinder_face_id = _first_face_by_surface(hollow_height_probe, "cylinder")
height_plan = hollow_height_probe.cylindrical_height_plan(hollow_cylinder_face_id, 167.0)
if height_plan.get("status") == "blocked":
raise SystemExit(f"hollow cylinder side Face height edit should be isolated, not blocked: {height_plan}")
_run_worker_case(
label="large hollow cylinder side Face height",
operation="resize_cylindrical_height",
args=[hollow_cylinder_face_id, 167.0],
validator=_assert_hollow_cylinder_extension,
input_path=hollow_path,
)
_run_worker_case(
label="偏移(局部重建)",
operation="move_face_plane_offset_local",
args=[0, 10.0],
validator=_assert_plane_position,
)
_run_worker_case(
label="面积(局部重建)",
operation="resize_face_area_local",
args=[0, 225.0],
validator=_assert_face_area,
)
_run_worker_case(
label="面积(缩放特征)",
operation="resize_face_area",
args=[0, 400.0],
validator=lambda label, model: _assert_face_area(label, model, 400.0),
)
_run_worker_case(
label="面内长度(局部重建)",
operation="resize_face_size_local",
args=[0, 25.0, "width"],
validator=_assert_face_width,
)
_run_worker_case(
label="面内宽度(局部重建)",
operation="resize_face_size_local",
args=[0, 25.0, "height"],
validator=_assert_face_height,
)
_run_worker_case(
label="面内长度(缩放特征)",
operation="resize_face_size_owning_scale",
args=[0, 25.0, "width"],
validator=lambda label, model: _assert_face_width(label, model, 25.0),
)
_run_worker_case(
label="面内宽度(缩放特征)",
operation="resize_face_size_owning_scale",
args=[0, 25.0, "height"],
validator=lambda label, model: _assert_face_height(label, model, 25.0),
)
_run_worker_case(
label="中心(局部重建)",
operation="move_face_center_local",
args=[0, [15.0, 5.0, 0.0]],
validator=_assert_face_center,
)
with tempfile.TemporaryDirectory(prefix="geom_param_isolated_shell_verify_") as temp_dir:
shell_path = Path(temp_dir) / "plate.step"
_write_shell_plate(shell_path)
shell_probe = StepModel.load(shell_path)
shell_face_id = _first_shell_face(shell_probe)
shell_plan = shell_probe.shell_thickness_plan(shell_face_id, 4.0)
if str(shell_plan.get("risk")) != "high":
raise SystemExit(f"shell thickness isolation case should be high risk, got {shell_plan}")
_run_worker_case(
label="壳体厚度(拉伸/切除)",
operation="resize_shell_thickness",
args=[shell_face_id, 4.0],
validator=_assert_shell_thickness,
input_path=shell_path,
)
shell_owning_plan = shell_probe.shell_thickness_owning_scale_plan(shell_face_id, 4.0)
if str(shell_owning_plan.get("risk")) != "high":
raise SystemExit(f"shell thickness owning isolation case should be high risk, got {shell_owning_plan}")
_run_worker_case(
label="壳体厚度(缩放特征)",
operation="resize_shell_thickness_owning_scale",
args=[shell_face_id, 4.0],
validator=_assert_shell_thickness,
input_path=shell_path,
)
with tempfile.TemporaryDirectory(prefix="geom_param_isolated_curved_face_verify_") as temp_dir:
temp_root = Path(temp_dir)
sphere_path = temp_root / "sphere.step"
_write_sphere_model(sphere_path)
sphere_probe = StepModel.load(sphere_path)
sphere_face_id = _first_face_by_surface(sphere_probe, "sphere")
sphere_plan = sphere_probe.spherical_radius_plan(sphere_face_id, 10.0)
if str(sphere_plan.get("risk")) != "high":
raise SystemExit(f"sphere radius isolation case should be high risk, got {sphere_plan}")
_run_worker_case(
label="sphere radius",
operation="resize_sphere_radius",
args=[sphere_face_id, 10.0],
validator=_assert_sphere_radius,
input_path=sphere_path,
)
torus_path = temp_root / "torus.step"
_write_torus_model(torus_path)
torus_probe = StepModel.load(torus_path)
torus_face_id = _first_face_by_surface(torus_probe, "torus")
torus_plan = torus_probe.toroidal_radius_plan(torus_face_id, 4.0, "minor")
if str(torus_plan.get("risk")) != "high":
raise SystemExit(f"torus minor-radius isolation case should be high risk, got {torus_plan}")
_run_worker_case(
label="torus minor radius",
operation="resize_torus_radius",
args=[torus_face_id, 4.0, "minor"],
validator=_assert_torus_minor_radius,
input_path=torus_path,
)
from step_editor.window_actions import WindowActionMixin
class _IsolatedJobProbe(WindowActionMixin):
def __init__(self) -> None:
self.model = StepModel.load(DEFAULT_MODEL)
self.step_path = DEFAULT_MODEL
self.current_info_values = self.model.face_info(0)
probe = _IsolatedJobProbe()
action_plan = probe._push_pull_plan_for_action(0, 11.0)
if action_plan.get("status") == "blocked" or str(action_plan.get("risk")) != "high":
raise SystemExit(f"UI push/pull action plan should use the model-layer high-risk plan: {action_plan}")
if float(action_plan.get("push_pull_distance_to_owning_axis_span_ratio") or 0.0) <= 1.0:
raise SystemExit(f"UI push/pull action plan should expose model-layer span ratio: {action_plan}")
complex_path = PROJECT_ROOT / "assets" / "models" / "geom_extract.step"
if complex_path.exists():
complex_probe = _IsolatedJobProbe()
complex_probe.model = StepModel.load(complex_path)
complex_probe.step_path = complex_path
complex_probe.selected_face_id = 1
complex_probe.current_info_values = complex_probe.model.quick_face_info(1)
complex_plan = complex_probe._push_pull_plan_for_action(1, 89.0)
if not complex_plan.get("ui_deferred_model_plan"):
raise SystemExit(f"complex holed cap UI plan should defer full model plan: {complex_plan}")
if str(complex_plan.get("risk")) != "high":
raise SystemExit(f"deferred complex holed cap UI plan should be high risk: {complex_plan}")
if complex_probe._isolation_for_plan(complex_plan, "push_pull_face", [1, 89.0]) is None:
raise SystemExit(f"deferred complex holed cap UI plan should request isolated execution: {complex_plan}")
if probe._isolated_edit_command(Path("request.json")) != [
sys.executable,
"-m",
"step_editor.isolated_edit_worker",
"request.json",
]:
raise SystemExit("source isolation command should use python -m step_editor.isolated_edit_worker")
had_frozen_attr = hasattr(sys, "frozen")
previous_frozen = getattr(sys, "frozen", None)
try:
setattr(sys, "frozen", True)
if probe._isolated_edit_command(Path("request.json")) != [
sys.executable,
"--isolated-edit-worker",
"request.json",
]:
raise SystemExit("frozen isolation command should call main.exe --isolated-edit-worker")
finally:
if had_frozen_attr:
setattr(sys, "frozen", previous_frozen)
else:
delattr(sys, "frozen")
for title in (
"面内长度(局部重建)",
"面内宽度(局部重建)",
"面内长度(缩放特征)",
"面内宽度(缩放特征)",
"壳体厚度(缩放特征)缩放所属对象",
):
if not probe._quick_edit_title_supports_isolation(title):
raise SystemExit(f"{title}: quick edit title should support isolated execution")
context = {
"operation_name": "面内长度(局部重建)",
"target": "Face 0",
"parameters": {"part_id": 1, "face_id": 0},
"target_kind": "face",
"target_id": 0,
"target_logical_id": 0,
"pick_position": None,
"show_same_domain_internal_edges": False,
"edit_result_deflection": 2.4,
}
snapshot = probe.model.snapshot()
result = probe._run_isolated_edit_job(
context=context,
isolation={
"operation": "resize_face_size_local",
"args": [0, 25.0, "width"],
"timeout_seconds": 120.0,
},
snapshot=snapshot,
before_stats=probe.model.stats(),
before_part_stats=probe.model.part_topology_stats(1),
before_quality=probe._edit_quality_info_or_none(probe.model, context, 1),
before_geometry={},
)
if "隔离子进程" not in str(result.get("message", "")):
raise SystemExit(f"isolated window job did not report isolated execution: {result}")
after_model = result.get("after_model")
if not isinstance(after_model, StepModel):
raise SystemExit(f"isolated window job should return after_model for UI-thread installation: {result}")
if probe.model.stats().solids != 1:
raise SystemExit(f"isolated window job changed solid count unexpectedly: {probe.model.stats()}")
if _logical_region_has_width(probe.model, 0, 25.0):
raise SystemExit("isolated window job replaced the window model before returning to the UI thread")
if after_model.stats().solids != 1:
raise SystemExit(f"isolated window after_model changed solid count unexpectedly: {after_model.stats()}")
if result.get("model_polydata") is None or result.get("edge_polydata") is None:
raise SystemExit("isolated window job should return prebuilt display polydata")
if not _logical_region_has_width(after_model, 0, 25.0):
raise SystemExit("isolated window job did not preserve the original logical Face ID on the edited width")
print("isolated window job ok")
owning_probe = _IsolatedJobProbe()
owning_plan = owning_probe.model.face_size_owning_scale_plan(0, 25.0, "width")
owning_isolation = owning_probe._isolation_for_plan(
owning_plan,
"resize_face_size_owning_scale",
[0, 25.0, "width"],
)
if owning_isolation is None:
raise SystemExit(f"Face owning size high-risk plan should request isolated execution: {owning_plan}")
owning_result = owning_probe._run_isolated_edit_job(
context={
**context,
"operation_name": "面内长度(缩放特征)",
},
isolation=owning_isolation,
snapshot=owning_probe.model.snapshot(),
before_stats=owning_probe.model.stats(),
before_part_stats=owning_probe.model.part_topology_stats(1),
before_quality=owning_probe._edit_quality_info_or_none(owning_probe.model, context, 1),
before_geometry={},
)
if "隔离子进程" not in str(owning_result.get("message", "")):
raise SystemExit(f"isolated owning window job did not report isolated execution: {owning_result}")
owning_after_model = owning_result.get("after_model")
if not isinstance(owning_after_model, StepModel):
raise SystemExit(
f"isolated owning window job should return after_model for UI-thread installation: {owning_result}"
)
if _logical_region_has_width(owning_probe.model, 0, 25.0):
raise SystemExit("isolated owning window job replaced the window model before returning to the UI thread")
if owning_result.get("model_polydata") is None or owning_result.get("edge_polydata") is None:
raise SystemExit("isolated owning window job should return prebuilt display polydata")
_assert_face_width("面内长度(整体窗口任务)", owning_after_model, 25.0)
if not _logical_region_has_width(owning_after_model, 0, 25.0):
raise SystemExit("isolated owning window job did not preserve the original logical Face ID on the edited width")
print("isolated owning window job ok")
_run_main_worker_entry_case()
return 0
if __name__ == "__main__":
raise SystemExit(main())