feat: 完善 Face 参数化编辑和隔离执行
This commit is contained in:
@@ -0,0 +1,330 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import subprocess
|
||||
import sys
|
||||
import tempfile
|
||||
from pathlib import Path
|
||||
|
||||
from OCC.Core.BRepPrimAPI import BRepPrimAPI_MakeBox
|
||||
|
||||
|
||||
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 _float_close(value: object, target: float, tolerance: float = 1e-5) -> bool:
|
||||
try:
|
||||
return abs(float(value) - target) <= tolerance
|
||||
except (TypeError, ValueError):
|
||||
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 _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 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,
|
||||
)
|
||||
|
||||
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()
|
||||
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_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()}")
|
||||
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_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)
|
||||
print("isolated owning window job ok")
|
||||
return 0
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
raise SystemExit(main())
|
||||
Reference in New Issue
Block a user