571 lines
21 KiB
Python
571 lines
21 KiB
Python
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：节点属性面板参数。
|
||
context：FlowEditor 上下文，常见字段包括 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())
|