feat: 推进一级关系参数化编辑与参数导出

This commit is contained in:
2026-08-13 17:50:20 +08:00
parent 19364d81b5
commit 7d814d2939
21 changed files with 2075 additions and 231 deletions
+570
View File
@@ -0,0 +1,570 @@
from __future__ import annotations
import argparse
from datetime import datetime
import json
import re
from pathlib import Path
import sys
import traceback
from .isolated_edit_worker import _execute
from .model import StepModel
COMPONENT_SCHEMA = "step-editor-parametric-component-v1"
_COMPONENT_SEQUENCE_RE = re.compile(r"^(?P<index>\d{3,})_(?P<name>.+)$")
_ACTION_OPERATION_MAP = {
"push_pull_face": "push_pull_face",
"push_pull_face_keep_relations": "push_pull_face_keep_relations",
"move_selected_face_plane_position_local": "move_face_plane_offset_local",
"move_selected_face_plane_position_by_translation": "translate_face_plane_offset_owning",
"resize_face_width_local": "resize_face_size_local",
"resize_face_height_local": "resize_face_size_local",
"resize_face_width_keep_relations": "resize_face_size_local_keep_relations",
"resize_face_height_keep_relations": "resize_face_size_local_keep_relations",
"resize_face_width_owning_scale": "resize_face_size_owning_scale",
"resize_face_height_owning_scale": "resize_face_size_owning_scale",
"resize_shell_thickness": "resize_shell_thickness",
"resize_shell_thickness_owning_scale": "resize_shell_thickness_owning_scale",
"resize_hole": "resize_cylindrical_hole",
"resize_cylindrical_owning_scale": "resize_cylindrical_owning_scale",
"resize_hole_depth": "resize_cylindrical_depth",
"resize_hole_depth_owning_scale": "resize_cylindrical_depth_owning_scale",
"resize_slot_width": "resize_cylindrical_slot_width",
"resize_slot_depth": "resize_cylindrical_slot_depth",
"resize_slot_arc_length": "resize_cylindrical_slot_arc_length",
"resize_slot_angular_span": "resize_cylindrical_slot_angular_span",
"resize_slot_total_length": "resize_cylindrical_slot_total_length",
"resize_slot_center_distance": "resize_cylindrical_slot_center_distance",
"move_cylindrical_hole_axis": "move_cylindrical_hole_axis",
"move_cylindrical_slot_axis": "move_cylindrical_slot_axis",
"suppress_hole": "suppress_cylindrical_hole",
"resize_boss": "resize_cylindrical_boss",
"resize_boss_height": "resize_cylindrical_boss_height",
"resize_cylinder_height": "resize_cylindrical_height",
"resize_cylindrical_height_owning_scale": "resize_cylindrical_height_owning_scale",
"move_cylindrical_boss_axis": "move_cylindrical_boss_axis",
"resize_cone_reference_radius": "resize_cone_reference_radius",
"resize_cone_semi_angle": "resize_cone_semi_angle",
"resize_sphere_radius": "resize_sphere_radius",
"resize_torus_major_radius": "resize_torus_radius",
"resize_torus_minor_radius": "resize_torus_radius",
"resize_any_edge_length": "resize_general_edge_length",
"move_edge_start_point": "move_edge_endpoint",
"move_edge_end_point": "move_edge_endpoint",
"move_edge_center_point": "move_edge_center",
"move_circular_edge_axis_center": "move_circular_edge_axis_center",
"resize_ellipse_edge_major_radius": "resize_ellipse_edge_axis_radius",
"resize_ellipse_edge_minor_radius": "resize_ellipse_edge_axis_radius",
"resize_existing_fillet": "resize_existing_fillet",
"resize_existing_chamfer": "resize_existing_chamfer",
"fillet_edge": "fillet_edge",
"chamfer_edge": "chamfer_edge",
"chamfer_edge_asymmetric": "chamfer_edge_asymmetric",
"chamfer_edge_distance_angle": "chamfer_edge_distance_angle",
}
def sanitize_component_name(value: object, fallback: str = "STEP_Parametric") -> str:
text = str(value or "").strip() or fallback
text = re.sub(r'[<>:"/\\|?*\x00-\x1f]+', "_", text)
text = re.sub(r"\s+", "_", text).strip(" ._")
return text or fallback
def default_component_root(project_root: Path | None = None) -> Path:
root = project_root or Path(__file__).resolve().parent.parent
return root / "nodes"
def next_component_dir(root: Path, component_name: object) -> Path:
base_name = sanitize_component_name(component_name)
max_index = -1
if root.is_dir():
for child in root.iterdir():
if not child.is_dir():
continue
match = _COMPONENT_SEQUENCE_RE.match(child.name)
if match:
max_index = max(max_index, int(match.group("index")))
return root / f"{max_index + 1:03d}_{base_name}"
def component_name_from_step(step_path: object) -> str:
try:
stem = Path(str(step_path)).stem
except Exception:
stem = ""
return sanitize_component_name(f"{stem}_STEP参数化组件" if stem else "STEP参数化组件")
def numeric_text(value: object) -> str:
text = str(value if value is not None else "").strip()
return text
def json_script_literal(value: object) -> str:
return json.dumps(value, ensure_ascii=False, indent=4)
def _float_or_text(value: object) -> object:
if isinstance(value, (int, float)):
return value
text = str(value if value is not None else "").strip()
try:
return float(text)
except ValueError:
return text
def _as_list3(value: object) -> list[float] | None:
if isinstance(value, str):
chunks = [chunk.strip() for chunk in value.strip().strip("()[]").replace(";", ",").split(",") if chunk.strip()]
elif isinstance(value, (tuple, list)):
chunks = list(value)
else:
return None
if len(chunks) != 3:
return None
try:
return [float(chunks[0]), float(chunks[1]), float(chunks[2])]
except (TypeError, ValueError):
return None
def _target_object_id(spec: dict[str, object], selected_kind: str | None, selected_face_id: int | None, selected_edge_id: int | None) -> int | None:
if spec.get("source_face_id") not in {"", None}:
return int(spec["source_face_id"])
action = str(spec.get("action") or "")
kind = str(selected_kind or "")
if "edge" in action and selected_edge_id is not None and kind == "edge":
return int(selected_edge_id)
if selected_face_id is not None:
return int(selected_face_id)
if selected_edge_id is not None:
return int(selected_edge_id)
return None
def _target_kind_for_action(action: str, selected_kind: str | None) -> str:
if "edge" in action:
return "edge"
if selected_kind in {"face", "feature", "edge"}:
return str(selected_kind)
return "face"
def operation_for_action(action: object) -> str | None:
return _ACTION_OPERATION_MAP.get(str(action or ""))
def _axis_arg_for_face_size(action: str) -> str | None:
if "face_width" in action:
return "width"
if "face_height" in action:
return "height"
return None
def _mode_arg_for_torus(action: str) -> str | None:
if action == "resize_torus_major_radius":
return "major"
if action == "resize_torus_minor_radius":
return "minor"
return None
def _endpoint_arg_for_edge(action: str) -> str | None:
if action == "move_edge_start_point":
return "start"
if action == "move_edge_end_point":
return "end"
return None
def _ellipse_axis_arg(action: str) -> str | None:
if action == "resize_ellipse_edge_major_radius":
return "major"
if action == "resize_ellipse_edge_minor_radius":
return "minor"
return None
def component_edit_config_from_spec(
*,
parameter_row: dict[str, str],
spec: dict[str, object],
selected_kind: str | None,
selected_face_id: int | None,
selected_edge_id: int | None,
step_path: Path | None,
) -> dict[str, object] | None:
action = str(spec.get("action") or "")
operation = operation_for_action(action)
target_id = _target_object_id(spec, selected_kind, selected_face_id, selected_edge_id)
if not operation or target_id is None:
return None
value_type = str(spec.get("value_type", "number"))
default_value = parameter_row.get("default", "")
target_value: object
if value_type == "vector3":
target_value = _as_list3(default_value) or _as_list3(spec.get("current_raw")) or default_value
elif value_type in {"number", "positive", "integer", "integer_or_empty"}:
target_value = _float_or_text(default_value)
else:
target_value = default_value
args: list[object] = [int(target_id)]
target_arg: object = {"param": parameter_row["name"]}
transform = str(spec.get("target_transform") or "")
if transform:
target_arg = {
"param": parameter_row["name"],
"transform": transform,
"context": spec.get("transform_context", {}),
}
if action == "suppress_hole":
target_value = ""
else:
args.append(target_arg)
axis_arg = _axis_arg_for_face_size(action)
if axis_arg is not None:
args.append(axis_arg)
elif action in {"resize_torus_major_radius", "resize_torus_minor_radius"}:
args.append(_mode_arg_for_torus(action))
elif action in {"move_edge_start_point", "move_edge_end_point"}:
args.insert(1, _endpoint_arg_for_edge(action))
elif action in {"resize_ellipse_edge_major_radius", "resize_ellipse_edge_minor_radius"}:
args.append(_ellipse_axis_arg(action))
elif action in {
"resize_hole_depth",
"resize_hole_depth_owning_scale",
"resize_slot_width",
"resize_slot_depth",
"resize_slot_arc_length",
"resize_slot_total_length",
"resize_slot_center_distance",
}:
manual_id = spec.get("manual_bottom_face_id")
if manual_id in {"", None}:
manual_id = spec.get("slot_pair_manual_face_id")
args.append("" if manual_id in {"", None} else manual_id)
return {
"parameter": parameter_row["name"],
"displayName": parameter_row.get("displayName", parameter_row["name"]),
"targetKind": _target_kind_for_action(action, selected_kind),
"targetId": int(target_id),
"uiAction": action,
"operation": operation,
"args": args,
"default": target_value,
"valueType": value_type,
"scope": spec.get("scope_key", spec.get("scope_default", "")),
"scopeLabel": spec.get("scope_label", spec.get("scope_text", "")),
"sourceStep": str(step_path or ""),
"parameterKey": spec.get("key", ""),
}
def render_component_main_py(component: dict[str, object]) -> str:
project_root = str(Path(__file__).resolve().parent.parent)
component_name = sanitize_component_name(component.get("componentName") or component_name_from_step(component.get("sourceStep")))
output_parameter = {
"name": "output_step",
"displayName": "输出STEP",
"type": "file",
"ioRole": "output",
"default": "",
}
return f'''# -*- coding: utf-8 -*-
"""
STEP 参数化组件。
这个文件按 FlowEditor 节点脚本方式生成:
1. INPUT_PARAMETERS 是从软件“导出参数”勾选行直接嵌入的输入参数。
2. PARAMETERS 会额外加上输出 STEP 文件参数,供节点设计器生成输出端口。
3. execute(inputs, params, context) 是 FlowEditor 调用入口。
4. main() 只用于本地命令行调试。
"""
from __future__ import annotations
import argparse
import json
import os
from pathlib import Path
import sys
PROJECT_ROOT = {json.dumps(project_root, ensure_ascii=False)}
if PROJECT_ROOT and PROJECT_ROOT not in sys.path:
sys.path.insert(0, PROJECT_ROOT)
from step_editor.parametric_component import run_embedded_component
INPUT_PARAMETERS = {json_script_literal(component.get("parameters", []))}
OUTPUT_PARAMETERS = [
{json_script_literal(output_parameter)}
]
PARAMETERS = INPUT_PARAMETERS + OUTPUT_PARAMETERS
COMPONENT = {json_script_literal(component)}
NODE_INFO = {{
"typeName": {json.dumps(component_name, ensure_ascii=False)},
"displayName": {json.dumps(component_name, ensure_ascii=False)},
"category": "几何参数化",
"icon": "icon.svg",
"parameters": PARAMETERS,
}}
def _value_from_inputs(name, inputs, params, default=""):
value = inputs.get(name) if isinstance(inputs, dict) else None
if value in (None, "") and isinstance(params, dict):
value = params.get(name)
if value in (None, ""):
value = default
return value
def _component_input_values(inputs, params):
values = {{}}
for item in INPUT_PARAMETERS:
name = item.get("name")
if not name:
continue
values[name] = _value_from_inputs(name, inputs, params, item.get("default", ""))
return values
def _default_output_step(work_dir, output_dir):
output_root = output_dir or os.path.join(work_dir, "output")
os.makedirs(output_root, exist_ok=True)
return os.path.join(output_root, COMPONENT.get("outputName") or "modified.step")
def run(inputs=None, output_step=None, work_dir=None):
return run_embedded_component(COMPONENT, inputs=inputs, output_step=output_step, work_dir=work_dir)
def execute(inputs, params, context):
"""
FlowEditor 调用入口。
inputs:上游节点传入值,优先级高于 params。
params:节点属性面板参数。
contextFlowEditor 上下文,常见字段包括 work_dir / input_dir / output_dir。
"""
inputs = inputs or {{}}
params = params or {{}}
context = context or {{}}
work_dir = context.get("work_dir") or os.getcwd()
output_dir = context.get("output_dir") or os.path.join(work_dir, "output")
output_step = _default_output_step(work_dir, output_dir)
result = run(
inputs=_component_input_values(inputs, params),
output_step=output_step,
work_dir=work_dir,
)
if not result.get("ok"):
raise RuntimeError(result.get("error") or json.dumps(result, ensure_ascii=False))
return {{
"output_step": result.get("outputStep", output_step),
"outputStep": result.get("outputStep", output_step),
"sourceStep": result.get("sourceStep", ""),
"messages": result.get("messages", []),
}}
def main(argv=None):
parser = argparse.ArgumentParser(description="Run generated STEP parametric component.")
parser.add_argument("--inputs", default="", help="JSON file or JSON object with input parameter values.")
parser.add_argument("--output-step", default="", help="Output STEP file path.")
parser.add_argument("--work-dir", default="", help="Runtime output directory.")
args = parser.parse_args(argv)
inputs = args.inputs
if inputs:
candidate = Path(inputs)
if candidate.is_file():
inputs = json.loads(candidate.read_text(encoding="utf-8-sig"))
else:
inputs = json.loads(inputs)
else:
inputs = {{}}
result = run(inputs=inputs, output_step=args.output_step or None, work_dir=args.work_dir or None)
print(json.dumps(result, ensure_ascii=False, indent=2))
return 0 if result.get("ok") else 2
if __name__ == "__main__":
raise SystemExit(main())
'''
def export_parametric_component(
*,
parameters: list[dict[str, str]],
edits: list[dict[str, object]],
source_step: Path | None,
component_root: Path | None = None,
component_name: str | None = None,
) -> Path:
if not parameters:
raise ValueError("No input parameters selected.")
root = component_root or default_component_root()
target_dir = next_component_dir(root, component_name or component_name_from_step(source_step))
target_dir.mkdir(parents=True, exist_ok=True)
component = {
"schema": COMPONENT_SCHEMA,
"createdAt": datetime.now().isoformat(timespec="seconds"),
"componentName": component_name or component_name_from_step(source_step),
"sourceStep": str(source_step or ""),
"parameters": parameters,
"edits": edits,
"outputName": "modified.step",
}
target = target_dir / "main.py"
target.write_text(render_component_main_py(component), encoding="utf-8")
return target
def _input_values(inputs: object) -> dict[str, object]:
if not isinstance(inputs, dict):
return {}
if isinstance(inputs.get("inputs"), dict):
return dict(inputs["inputs"])
rows = inputs.get("parameters")
if isinstance(rows, list):
result: dict[str, object] = {}
for row in rows:
if isinstance(row, dict) and row.get("name"):
result[str(row["name"])] = row.get("value", row.get("default", ""))
return result
return dict(inputs)
def _parse_vector3(value: object) -> list[float]:
vector = _as_list3(value)
if vector is None:
raise ValueError(f"Expected 3D vector value, got {value!r}.")
return vector
def _apply_arg_transform(value: object, transform: str, context: object) -> object:
context = context if isinstance(context, dict) else {}
if transform == "plane_target_position_to_offset":
current = float(context.get("current_plane_position"))
return float(value) - current
if transform == "radius_to_diameter":
return float(value) * 2.0
if transform == "diameter_to_radius":
return float(value) * 0.5
if transform == "degrees_to_radians":
import math
return math.radians(float(value))
if transform == "slot_open_angle_degrees_to_angular_span":
import math
return math.tau - math.radians(float(value))
if transform == "target_center_to_translation":
current = _parse_vector3(context.get("current_center"))
target = _parse_vector3(value)
return [target[index] - current[index] for index in range(3)]
return value
def _resolve_arg(value: object, values: dict[str, object], defaults: dict[str, object]) -> object:
if isinstance(value, dict) and "param" in value:
name = str(value.get("param") or "")
resolved = values.get(name, defaults.get(name, ""))
transform = str(value.get("transform") or "")
if transform:
return _apply_arg_transform(resolved, transform, value.get("context"))
return resolved
return value
def run_embedded_component(
component: dict[str, object],
*,
inputs: object | None = None,
output_step: str | Path | None = None,
work_dir: str | Path | None = None,
) -> dict[str, object]:
started = datetime.now().isoformat(timespec="seconds")
try:
source_step = Path(str(component.get("sourceStep") or "")).expanduser()
if not source_step.is_file():
return {"ok": False, "error": f"Source STEP does not exist: {source_step}", "startedAt": started}
output_path = Path(output_step) if output_step else None
if output_path is None:
output_root = Path(work_dir) if work_dir else Path.cwd()
output_path = output_root / str(component.get("outputName") or "modified.step")
output_path.parent.mkdir(parents=True, exist_ok=True)
parameters = [row for row in component.get("parameters", []) if isinstance(row, dict)]
defaults = {str(row.get("name") or ""): row.get("default", "") for row in parameters if row.get("name")}
values = _input_values(inputs)
model = StepModel.load(source_step)
messages: list[str] = []
for edit in component.get("edits", []):
if not isinstance(edit, dict):
continue
operation = str(edit.get("operation") or "")
args = [_resolve_arg(arg, values, defaults) for arg in list(edit.get("args") or [])]
messages.append(_execute(model, operation, args))
model.export_all(output_path)
return {
"ok": True,
"startedAt": started,
"finishedAt": datetime.now().isoformat(timespec="seconds"),
"sourceStep": str(source_step),
"outputStep": str(output_path),
"messages": messages,
"parameters": parameters,
}
except Exception as exc:
return {
"ok": False,
"startedAt": started,
"finishedAt": datetime.now().isoformat(timespec="seconds"),
"error": str(exc),
"traceback": traceback.format_exc(),
}
def _load_inputs(path_or_json: str) -> object:
if not path_or_json:
return {}
candidate = Path(path_or_json)
if candidate.is_file():
return json.loads(candidate.read_text(encoding="utf-8-sig"))
return json.loads(path_or_json)
def main(argv: list[str] | None = None) -> int:
parser = argparse.ArgumentParser(description="Run a STEP parametric component JSON.")
parser.add_argument("component", help="Component JSON file.")
parser.add_argument("--inputs", default="", help="JSON file or inline JSON object.")
parser.add_argument("--output-step", default="", help="Output STEP path.")
parser.add_argument("--work-dir", default="", help="Runtime work directory.")
parsed = parser.parse_args(argv)
try:
component = json.loads(Path(parsed.component).read_text(encoding="utf-8-sig"))
result = run_embedded_component(
component,
inputs=_load_inputs(parsed.inputs),
output_step=parsed.output_step or None,
work_dir=parsed.work_dir or None,
)
except Exception as exc:
result = {"ok": False, "error": str(exc), "traceback": traceback.format_exc()}
print(json.dumps(result, ensure_ascii=False, indent=2))
return 0 if result.get("ok") else 2
if __name__ == "__main__":
raise SystemExit(main())