pythonocc-step-editor/scripts/verify_large_stepped_cap_push_pull.py

from __future__ import annotations

import sys
import time
import json
import subprocess
import tempfile
from pathlib import Path

from OCC.Core.BRepAdaptor import BRepAdaptor_Surface
from OCC.Core.BRepGProp import brepgprop
from OCC.Core.GeomAbs import GeomAbs_Cylinder, GeomAbs_Plane
from OCC.Core.GProp import GProp_GProps
from OCC.Core.TopAbs import TopAbs_WIRE
from OCC.Core.TopExp import TopExp_Explorer

PROJECT_ROOT = Path(__file__).resolve().parent.parent
if str(PROJECT_ROOT) not in sys.path:
    sys.path.insert(0, str(PROJECT_ROOT))

from step_editor.model import StepModel
from step_editor.records import OperationRecord
from step_editor.window_actions import WindowActionMixin
from step_editor.window_state import WindowStateMixin


class _WindowActionProbe(WindowActionMixin):
    pass


class _WindowStateProbe(WindowStateMixin):
    pass


def _wire_count(face) -> int:
    count = 0
    explorer = TopExp_Explorer(face, TopAbs_WIRE)
    while explorer.More():
        count += 1
        explorer.Next()
    return count


def _adjacent_cylinder_radii(model: StepModel, face_id: int) -> list[float]:
    adjacent_ids: list[int] = []
    for edge_id in model._face_boundary_edge_ids(face_id):
        for adjacent_id in model._edge_adjacent_face_ids(edge_id):
            if adjacent_id == face_id or adjacent_id in adjacent_ids:
                continue
            adjacent_ids.append(adjacent_id)

    radii: list[float] = []
    for adjacent_id in adjacent_ids:
        surf = BRepAdaptor_Surface(model.faces[adjacent_id])
        if surf.GetType() != GeomAbs_Cylinder:
            continue
        radius = float(surf.Cylinder().Radius())
        if not any(abs(radius - existing) <= max(radius, existing, 1.0) * 1e-5 for existing in radii):
            radii.append(radius)
    return sorted(radii)


def _large_stepped_cap_face(model: StepModel) -> int:
    best: tuple[float, int] | None = None
    for face_id, face in enumerate(model.faces):
        surf = BRepAdaptor_Surface(face)
        if surf.GetType() != GeomAbs_Plane:
            continue
        if _wire_count(face) != 2:
            continue
        radii = _adjacent_cylinder_radii(model, face_id)
        if len(radii) != 2:
            continue
        if abs(radii[0] - 4.1) > 1e-4 or abs(radii[1] - 15.0) > 1e-4:
            continue
        props = GProp_GProps()
        brepgprop.SurfaceProperties(face, props)
        center = props.CentreOfMass()
        score = abs(center.X() - 120.0) + abs(center.Y() + 39.0) + abs(center.Z() + 131.0)
        if best is None or score < best[0]:
            best = (score, face_id)
    if best is None:
        raise SystemExit("large stepped cap Face was not found in geom_extract.step")
    return best[1]


def _large_multi_boundary_cap_face(model: StepModel) -> int:
    best: tuple[float, int] | None = None
    for face_id, face in enumerate(model.faces):
        surf = BRepAdaptor_Surface(face)
        if surf.GetType() != GeomAbs_Plane:
            continue
        if _wire_count(face) != 6:
            continue
        props = GProp_GProps()
        brepgprop.SurfaceProperties(face, props)
        area = float(props.Mass())
        if area < 20000.0:
            continue
        center = props.CentreOfMass()
        score = abs(center.Y() + 57.5) + abs(center.Z() + 249.7) * 0.02
        if best is None or score < best[0]:
            best = (score, face_id)
    if best is None:
        raise SystemExit("large multi-boundary cap Face was not found in geom_extract.step")
    return best[1]


def _face_topology_counts(model: StepModel, face_id: int) -> dict[str, int]:
    topology = model.face_first_level_topology(face_id)
    info = model.quick_face_info(face_id)
    return {
        "boundary_edges": int(topology.get("first_level_boundary_edge_count", 0) or 0),
        "boundary_vertices": int(topology.get("first_level_boundary_vertex_count", 0) or 0),
        "adjacent_faces": int(topology.get("first_level_adjacent_face_count", 0) or 0),
        "inner_wires": int(info.get("inner_boundary_wires", 0) or 0),
    }


def _assert_topology_preserved(
    label: str,
    before: dict[str, int],
    after: dict[str, int],
    *,
    min_inner_wires: int | None = None,
) -> None:
    for key in ("boundary_edges", "boundary_vertices", "adjacent_faces"):
        if after[key] < before[key]:
            raise SystemExit(f"{label} lost first-level {key}: before={before}, after={after}")
    expected_inner_wires = before["inner_wires"] if min_inner_wires is None else min_inner_wires
    if after["inner_wires"] < expected_inner_wires:
        raise SystemExit(f"{label} lost inner boundary wires: before={before}, after={after}")


def main() -> int:
    path = PROJECT_ROOT / "assets" / "models" / "geom_extract.step"
    if not path.exists():
        print("large stepped cap fixture is absent; skipping")
        return 0

    model = StepModel.load(path)
    face_id = _large_stepped_cap_face(model)
    logical_id = model.face_region_logical_id(face_id)
    before_topology = _face_topology_counts(model, face_id)
    started = time.perf_counter()
    plan = model.push_pull_plan(face_id, 89.0)
    plan_elapsed = time.perf_counter() - started
    if plan_elapsed > 5.0:
        raise SystemExit(f"large stepped cap push/pull plan took too long: {plan_elapsed:.3f}s; plan={plan}")
    if plan.get("cylindrical_cap_extension_kind") != "coaxial-stepped-cap":
        raise SystemExit(f"large stepped cap should be recognized as stepped cap: {plan}")
    if plan.get("cylindrical_cap_extension_method") != "local-shell-rebuild":
        raise SystemExit(f"large stepped cap should use local shell rebuild: {plan}")
    stepped_isolation = _WindowActionProbe()._isolation_for_plan(plan, "push_pull_face", [face_id, 89.0])
    if not stepped_isolation or stepped_isolation.get("reason") != "large-model-smooth-ui-isolated-occ-edit":
        raise SystemExit(f"large stepped cap should use the smooth UI background process: {stepped_isolation}")

    started = time.perf_counter()
    result = model.push_pull_face(face_id, 89.0)
    elapsed = time.perf_counter() - started
    if elapsed > 30.0:
        raise SystemExit(f"large stepped cap push/pull took too long: {elapsed:.3f}s; result={result}")
    if "local shell rebuild" not in result or "coaxial-stepped-cap" not in result:
        raise SystemExit(f"large stepped cap push/pull should report local shell rebuild: {result}")
    stats = model.stats()
    if stats.solids != 1:
        raise SystemExit(f"large stepped cap push/pull should keep one Solid: {stats}")
    if "actual=50" not in result or "inner_wires=1" not in result:
        raise SystemExit(f"large stepped cap push/pull should verify target cap and inner wire: {result}")
    retained_ids = model.face_ids_for_logical_id(logical_id)
    if len(retained_ids) != 1:
        raise SystemExit(f"large stepped cap logical Face should map to one new cap: {retained_ids}")
    retained_info = model.quick_face_info(retained_ids[0])
    retained_origin = retained_info.get("plane_origin")
    retained_outward = retained_info.get("push_pull_outward_direction") or retained_info.get("normal")
    retained_position = (
        sum(float(retained_origin[index]) * float(retained_outward[index]) for index in range(3))
        if isinstance(retained_origin, (tuple, list))
        and isinstance(retained_outward, (tuple, list))
        and len(retained_origin) == 3
        and len(retained_outward) == 3
        else None
    )
    if retained_position is None or abs(retained_position - 50.0) > 1e-4:
        raise SystemExit(
            f"large stepped cap logical Face should follow the moved cap at 50: "
            f"ids={retained_ids}, position={retained_position}"
        )
    after_topology = _face_topology_counts(model, retained_ids[0])
    _assert_topology_preserved("large stepped cap", before_topology, after_topology)

    with tempfile.TemporaryDirectory(prefix="verify_large_stepped_cap_isolated_") 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(path),
                    "output_path": str(output_path),
                    "operation": "push_pull_face",
                    "args": [face_id, 89.0],
                },
                ensure_ascii=False,
                indent=2,
            ),
            encoding="utf-8",
        )
        started = time.perf_counter()
        completed = subprocess.run(
            [sys.executable, "-m", "step_editor.isolated_edit_worker", str(request_path)],
            cwd=PROJECT_ROOT,
            stdout=subprocess.PIPE,
            stderr=subprocess.PIPE,
            text=True,
            encoding="utf-8",
            errors="replace",
            timeout=60.0,
            check=False,
        )
        isolated_elapsed = time.perf_counter() - started
        response_path = request_path.with_suffix(".response.json")
        response = json.loads(response_path.read_text(encoding="utf-8")) if response_path.exists() else {}
        if completed.returncode != 0 or not response.get("ok"):
            raise SystemExit(
                "large stepped cap isolated worker failed: "
                f"returncode={completed.returncode}, stdout={completed.stdout!r}, "
                f"stderr={completed.stderr!r}, response={response}"
            )
        isolated_message = str(response.get("message") or "")
        if isolated_elapsed > 30.0:
            raise SystemExit(
                f"large stepped cap isolated worker took too long: {isolated_elapsed:.3f}s; "
                f"message={isolated_message}"
            )
        if not output_path.exists():
            raise SystemExit("large stepped cap isolated worker did not export output STEP")
        if "local shell rebuild" not in isolated_message or "actual=50" not in isolated_message:
            raise SystemExit(f"large stepped cap isolated worker used the wrong path: {isolated_message}")

    multi_model = StepModel.load(path)
    multi_face_id = _large_multi_boundary_cap_face(multi_model)
    multi_logical_id = multi_model.face_region_logical_id(multi_face_id)
    multi_before_topology = _face_topology_counts(multi_model, multi_face_id)
    started = time.perf_counter()
    multi_plan = multi_model.push_pull_plan(multi_face_id, 34.5)
    multi_plan_elapsed = time.perf_counter() - started
    if multi_plan_elapsed > 5.0:
        raise SystemExit(
            f"multi-boundary cap push/pull plan should be quick: {multi_plan_elapsed:.3f}s; plan={multi_plan}"
        )
    if abs(float(multi_plan.get("current_plane_position") or 0.0) - 57.5) > 1e-9:
        raise SystemExit(f"multi-boundary cap should start at 57.5: {multi_plan}")
    if abs(float(multi_plan.get("target_plane_position") or 0.0) - 92.0) > 1e-9:
        raise SystemExit(f"multi-boundary cap should target 92: {multi_plan}")
    if multi_plan.get("planar_cap_extension_kind") != "multi-boundary-planar-cap":
        raise SystemExit(f"multi-boundary cap should be recognized as planar cap: {multi_plan}")
    if multi_plan.get("planar_cap_extension_method") != "boundary-shell-rebuild":
        raise SystemExit(f"multi-boundary cap should use boundary shell rebuild: {multi_plan}")
    multi_isolation = _WindowActionProbe()._isolation_for_plan(multi_plan, "push_pull_face", [multi_face_id, 34.5])
    if not multi_isolation or multi_isolation.get("reason") != "large-model-smooth-ui-isolated-occ-edit":
        raise SystemExit(f"multi-boundary cap should use the smooth UI background process: {multi_isolation}")
    ui_probe = _WindowActionProbe()
    ui_probe.model = multi_model
    ui_probe.current_info_values = multi_model.quick_face_info(multi_face_id)
    ui_plan = ui_probe._push_pull_plan_for_action(multi_face_id, 34.5)
    if bool(ui_plan.get("ui_deferred_model_plan")):
        raise SystemExit(f"large multi-boundary UI plan should use the fast local plan now: {ui_plan}")
    if ui_plan.get("planar_cap_extension_method") != "boundary-shell-rebuild":
        raise SystemExit(f"large multi-boundary UI plan should use boundary-shell rebuild: {ui_plan}")
    ui_isolation = ui_probe._isolation_for_plan(ui_plan, "push_pull_face", [multi_face_id, 34.5])
    if not ui_isolation or ui_isolation.get("reason") != "large-model-smooth-ui-isolated-occ-edit":
        raise SystemExit(f"large multi-boundary UI plan should use the smooth UI background process: {ui_isolation}")
    if ui_probe._edit_preflight_blocker({"parameters": ui_plan}) is not None:
        raise SystemExit(f"large multi-boundary UI plan should not be blocked before editing: {ui_plan}")
    started = time.perf_counter()
    multi_inward_plan = multi_model.push_pull_plan(multi_face_id, -1.0)
    multi_inward_elapsed = time.perf_counter() - started
    if multi_inward_elapsed > 5.0:
        raise SystemExit(
            f"multi-boundary cap inward push/pull plan should be quick: {multi_inward_elapsed:.3f}s"
        )
    if multi_inward_plan.get("status") == "blocked":
        raise SystemExit(
            f"multi-boundary cap shallow inward push/pull should be allowed by boundary shell rebuild: "
            f"{multi_inward_plan}"
        )
    if multi_inward_plan.get("planar_cap_extension_method") != "boundary-shell-rebuild":
        raise SystemExit(
            f"multi-boundary cap shallow inward should use boundary shell rebuild: {multi_inward_plan}"
        )
    multi_inward_message = str(multi_inward_plan.get("message") or "")
    if "向内收缩" not in multi_inward_message and "重建侧壁" not in multi_inward_message:
        raise SystemExit(
            f"multi-boundary cap shallow inward message should explain boundary shell semantics: "
            f"{multi_inward_plan}"
        )
    multi_inward_result_model = StepModel.load(path)
    started = time.perf_counter()
    multi_inward_result = multi_inward_result_model.push_pull_face(multi_face_id, -1.0)
    multi_inward_result_elapsed = time.perf_counter() - started
    if multi_inward_result_elapsed > 15.0:
        raise SystemExit(
            f"multi-boundary cap shallow inward push/pull took too long: "
            f"{multi_inward_result_elapsed:.3f}s; result={multi_inward_result}"
        )
    if "boundary-shell rebuild" not in multi_inward_result or "actual=56.5" not in multi_inward_result:
        raise SystemExit(
            f"multi-boundary cap shallow inward should move to 56.5 by boundary shell rebuild: {multi_inward_result}"
        )
    if multi_inward_result_model.stats().solids != 1:
        raise SystemExit(f"multi-boundary cap shallow inward should keep one Solid: {multi_inward_result_model.stats()}")
    inward_retained_ids = multi_inward_result_model.face_ids_for_logical_id(multi_logical_id)
    if len(inward_retained_ids) != 1:
        raise SystemExit(f"multi-boundary cap shallow inward logical Face should follow the moved cap: {inward_retained_ids}")
    inward_info = multi_inward_result_model.quick_face_info(inward_retained_ids[0])
    if int(inward_info.get("inner_boundary_wires", 0) or 0) < 5:
        raise SystemExit(f"multi-boundary cap shallow inward should keep inner wires: {inward_info}")

    multi_deep_inward_plan = multi_model.push_pull_plan(multi_face_id, -77.5)
    if multi_deep_inward_plan.get("status") != "blocked":
        raise SystemExit(
            f"multi-boundary cap deep inward push/pull should still be blocked: {multi_deep_inward_plan}"
        )
    multi_deep_inward_message = str(multi_deep_inward_plan.get("message") or "")
    if "材料厚度" not in multi_deep_inward_message and "二级关系" not in multi_deep_inward_message:
        raise SystemExit(
            f"multi-boundary cap deep inward blocker should explain cut-through or topology depth: "
            f"{multi_deep_inward_plan}"
        )

    started = time.perf_counter()
    multi_result = multi_model.push_pull_face(multi_face_id, 34.5)
    multi_elapsed = time.perf_counter() - started
    if multi_elapsed > 15.0:
        raise SystemExit(
            f"multi-boundary cap push/pull took too long: {multi_elapsed:.3f}s; result={multi_result}"
        )
    if "boundary-shell rebuild" not in multi_result:
        raise SystemExit(f"multi-boundary cap push/pull should report boundary-shell rebuild: {multi_result}")
    multi_stats = multi_model.stats()
    if multi_stats.solids != 1:
        raise SystemExit(f"multi-boundary cap push/pull should keep one Solid: {multi_stats}")
    if "actual=92" not in multi_result or "inner_wires=5" not in multi_result:
        raise SystemExit(
            f"multi-boundary cap push/pull should verify target cap and inner wires: {multi_result}"
        )
    multi_retained_ids = multi_model.face_ids_for_logical_id(multi_logical_id)
    if len(multi_retained_ids) != 1:
        raise SystemExit(
            f"multi-boundary cap logical Face should map to one moved cap: {multi_retained_ids}"
        )
    multi_retained_info = multi_model.quick_face_info(multi_retained_ids[0])
    multi_origin = multi_retained_info.get("plane_origin")
    multi_outward = multi_retained_info.get("push_pull_outward_direction") or multi_retained_info.get("normal")
    multi_position = (
        sum(float(multi_origin[index]) * float(multi_outward[index]) for index in range(3))
        if isinstance(multi_origin, (tuple, list))
        and isinstance(multi_outward, (tuple, list))
        and len(multi_origin) == 3
        and len(multi_outward) == 3
        else None
    )
    if multi_position is None or abs(multi_position - 92.0) > 1e-4:
        raise SystemExit(
            f"multi-boundary cap logical Face should follow the moved cap at 92: "
            f"ids={multi_retained_ids}, position={multi_position}"
        )
    if int(multi_retained_info.get("inner_boundary_wires", 0) or 0) < 5:
        raise SystemExit(
            f"multi-boundary cap logical Face should keep inner boundary wires: {multi_retained_info}"
        )
    multi_after_topology = _face_topology_counts(multi_model, multi_retained_ids[0])
    _assert_topology_preserved(
        "multi-boundary cap",
        multi_before_topology,
        multi_after_topology,
        min_inner_wires=5,
    )
    locator_probe = _WindowStateProbe()
    locator_probe.model = multi_model
    locator_record = OperationRecord(
        summary="test",
        detail="test",
        operation_name="拉伸/切除平面",
        target=f"Face {multi_face_id}",
        parameters={
            "part_id": multi_plan.get("part_id"),
            "solid_id": multi_plan.get("solid_id"),
            "surface": "plane",
            "outward_direction": multi_plan.get("outward_direction") or multi_plan.get("plane_direction"),
            "target_plane_position": multi_plan.get("target_plane_position"),
            "bbox_diagonal": multi_plan.get("bbox_diagonal"),
        },
        result_message=multi_result,
        target_kind="face",
        target_id=multi_face_id,
        target_logical_id=multi_logical_id,
    )
    started = time.perf_counter()
    resolved_after_edit = locator_probe._resolve_record_face_id(locator_record)
    locator_elapsed = time.perf_counter() - started
    if resolved_after_edit != multi_retained_ids[0] or locator_elapsed > 0.5:
        raise SystemExit(
            f"large multi-boundary operation history locator should use the fast result Face: "
            f"resolved={resolved_after_edit}, expected={multi_retained_ids[0]}, elapsed={locator_elapsed:.3f}s"
        )
    no_hint_record = OperationRecord(
        summary="test",
        detail="test",
        operation_name="拉伸/切除平面",
        target=f"Face {multi_face_id}",
        parameters=locator_record.parameters,
        result_message="Planar face push/pull completed without result face hint.",
        target_kind="face",
        target_id=multi_face_id,
        target_logical_id=multi_logical_id,
    )
    started = time.perf_counter()
    fallback_after_edit = locator_probe._record_plane_position_face_id(no_hint_record)
    fallback_elapsed = time.perf_counter() - started
    if fallback_after_edit != multi_retained_ids[0] or fallback_elapsed > 1.0:
        raise SystemExit(
            f"large multi-boundary fallback locator should use lightweight plane positions: "
            f"resolved={fallback_after_edit}, expected={multi_retained_ids[0]}, elapsed={fallback_elapsed:.3f}s"
        )

    with tempfile.TemporaryDirectory(prefix="verify_large_multi_boundary_cap_isolated_") 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(path),
                    "output_path": str(output_path),
                    "operation": "push_pull_face",
                    "args": [multi_face_id, 34.5],
                },
                ensure_ascii=False,
                indent=2,
            ),
            encoding="utf-8",
        )
        started = time.perf_counter()
        completed = subprocess.run(
            [sys.executable, "-m", "step_editor.isolated_edit_worker", str(request_path)],
            cwd=PROJECT_ROOT,
            stdout=subprocess.PIPE,
            stderr=subprocess.PIPE,
            text=True,
            encoding="utf-8",
            errors="replace",
            timeout=60.0,
            check=False,
        )
        multi_isolated_elapsed = time.perf_counter() - started
        response_path = request_path.with_suffix(".response.json")
        response = json.loads(response_path.read_text(encoding="utf-8")) if response_path.exists() else {}
        if completed.returncode != 0 or not response.get("ok"):
            raise SystemExit(
                "large multi-boundary cap isolated worker failed: "
                f"returncode={completed.returncode}, stdout={completed.stdout!r}, "
                f"stderr={completed.stderr!r}, response={response}"
            )
        multi_isolated_message = str(response.get("message") or "")
        if multi_isolated_elapsed > 30.0:
            raise SystemExit(
                f"large multi-boundary cap isolated worker took too long: {multi_isolated_elapsed:.3f}s; "
                f"message={multi_isolated_message}"
            )
        if not output_path.exists():
            raise SystemExit("large multi-boundary cap isolated worker did not export output STEP")
        if "boundary-shell rebuild" not in multi_isolated_message or "actual=92" not in multi_isolated_message:
            raise SystemExit(
                f"large multi-boundary cap isolated worker used the wrong path: {multi_isolated_message}"
            )
        isolated_output_model = StepModel.load(output_path)
        probe = _WindowActionProbe()
        direction = multi_plan.get("outward_direction") or multi_plan.get("plane_direction")
        context = {
            "operation_name": "拉伸/切除平面",
            "target_kind": "face",
            "target_id": multi_face_id,
            "target_logical_id": multi_face_id,
            "parameters": {
                "part_id": multi_plan.get("part_id"),
                "solid_id": multi_plan.get("solid_id"),
                "surface": "plane",
                "outward_direction": direction,
                "current_plane_position": multi_plan.get("current_plane_position"),
                "target_plane_position": multi_plan.get("target_plane_position"),
                "bbox_diagonal": multi_plan.get("bbox_diagonal"),
                "resize_strategy": multi_plan.get("resize_strategy"),
            },
        }
        blocker = probe._face_target_integrity_blocker(isolated_output_model, context)
        if blocker:
            raise SystemExit(f"window-layer isolated target check should accept the moved cap: {blocker}")
        probe._preserve_isolated_face_logical_id(isolated_output_model, context, multi_isolated_message)
        retained_after_isolation = isolated_output_model.face_ids_for_logical_id(multi_face_id)
        if not retained_after_isolation:
            raise SystemExit("window-layer isolated logical Face retention lost the moved cap")
        if not probe._face_target_plane_position_matches(
            isolated_output_model,
            retained_after_isolation,
            float(multi_plan["target_plane_position"]),
            direction,
            float(multi_plan["bbox_diagonal"]),
        ):
            raise SystemExit(
                f"window-layer isolated logical Face retention points at the wrong cap: {retained_after_isolation}"
            )

    print(
        "large stepped cap push/pull ok: "
        f"face_id={face_id}, plan_elapsed={plan_elapsed:.3f}s, "
        f"elapsed={elapsed:.3f}s, isolated_elapsed={isolated_elapsed:.3f}s, "
        f"topology_before={before_topology}, topology_after={after_topology}, result={result}"
    )
    print(
        "large multi-boundary cap push/pull ok: "
        f"face_id={multi_face_id}, plan_elapsed={multi_plan_elapsed:.3f}s, elapsed={multi_elapsed:.3f}s, "
        f"isolated_elapsed={multi_isolated_elapsed:.3f}s, "
        f"topology_before={multi_before_topology}, topology_after={multi_after_topology}, result={multi_result}"
    )
    return 0


if __name__ == "__main__":
    raise SystemExit(main())