from __future__ import annotations import json import subprocess import sys import tempfile from pathlib import Path from OCC.Core.BRepPrimAPI import BRepPrimAPI_MakeBox, BRepPrimAPI_MakeSphere, BRepPrimAPI_MakeTorus 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.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 _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 _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 main() -> int: _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 probe = _IsolatedJobProbe() 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}") if probe.model.stats().solids != 1: raise SystemExit(f"isolated window job changed solid count unexpectedly: {probe.model.stats()}") if not _logical_region_has_width(probe.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}") _assert_face_width("面宽(整体窗口任务)", owning_probe.model, 25.0) if not _logical_region_has_width(owning_probe.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())