|
|
|
|
@ -100,6 +100,14 @@ def _dominant_axis(vector):
|
|
|
|
|
return axis
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
def _point_payload(point):
|
|
|
|
|
return {
|
|
|
|
|
"x": float(getattr(point, "x", 0.0)),
|
|
|
|
|
"y": float(getattr(point, "y", 0.0)),
|
|
|
|
|
"z": float(getattr(point, "z", 0.0)),
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
def _coerce_waypoint(point_like):
|
|
|
|
|
if isinstance(point_like, dict):
|
|
|
|
|
point = _vector_from_point(
|
|
|
|
|
@ -129,7 +137,10 @@ def _coerce_waypoint(point_like):
|
|
|
|
|
"point": point,
|
|
|
|
|
"support_axis": support_axis,
|
|
|
|
|
"anchor_kind": (point_like.get("anchor_kind", "") or "").strip(),
|
|
|
|
|
"carrier_kind": (point_like.get("carrier_kind", "") or "").strip(),
|
|
|
|
|
"carrier_axis": (point_like.get("carrier_axis", "") or "").strip().lower(),
|
|
|
|
|
"source_label": (point_like.get("source_label", "") or "").strip(),
|
|
|
|
|
"source_object_name": (point_like.get("source_object_name", "") or "").strip(),
|
|
|
|
|
"subelement_name": (point_like.get("subelement_name", "") or "").strip(),
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
@ -140,7 +151,10 @@ def _coerce_waypoint(point_like):
|
|
|
|
|
"point": point,
|
|
|
|
|
"support_axis": None,
|
|
|
|
|
"anchor_kind": "",
|
|
|
|
|
"carrier_kind": "",
|
|
|
|
|
"carrier_axis": "",
|
|
|
|
|
"source_label": "",
|
|
|
|
|
"source_object_name": "",
|
|
|
|
|
"subelement_name": "",
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
@ -209,22 +223,279 @@ def _manual_waypoints_payload(waypoints):
|
|
|
|
|
},
|
|
|
|
|
"support_axis": waypoint.get("support_axis", ""),
|
|
|
|
|
"anchor_kind": waypoint.get("anchor_kind", ""),
|
|
|
|
|
"carrier_kind": waypoint.get("carrier_kind", ""),
|
|
|
|
|
"carrier_axis": waypoint.get("carrier_axis", ""),
|
|
|
|
|
"source_label": waypoint.get("source_label", ""),
|
|
|
|
|
"source_object_name": waypoint.get("source_object_name", ""),
|
|
|
|
|
"subelement_name": waypoint.get("subelement_name", ""),
|
|
|
|
|
}
|
|
|
|
|
)
|
|
|
|
|
return payload
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
def _route_node_payload(
|
|
|
|
|
role,
|
|
|
|
|
point,
|
|
|
|
|
terminal=None,
|
|
|
|
|
waypoint=None,
|
|
|
|
|
waypoint_index=None,
|
|
|
|
|
):
|
|
|
|
|
payload = {
|
|
|
|
|
"role": role,
|
|
|
|
|
"point": _point_payload(point),
|
|
|
|
|
}
|
|
|
|
|
if terminal is not None:
|
|
|
|
|
payload["terminal_uuid"] = getattr(terminal, "QetTerminalUuid", "").strip()
|
|
|
|
|
payload["instance_id"] = getattr(terminal, "QetInstanceId", "").strip()
|
|
|
|
|
if waypoint is not None:
|
|
|
|
|
payload["waypoint_index"] = int(waypoint_index or 0)
|
|
|
|
|
payload["support_axis"] = waypoint.get("support_axis", "") or ""
|
|
|
|
|
payload["anchor_kind"] = waypoint.get("anchor_kind", "") or ""
|
|
|
|
|
payload["carrier_kind"] = waypoint.get("carrier_kind", "") or ""
|
|
|
|
|
payload["carrier_axis"] = waypoint.get("carrier_axis", "") or ""
|
|
|
|
|
payload["source_label"] = waypoint.get("source_label", "") or ""
|
|
|
|
|
payload["source_object_name"] = waypoint.get("source_object_name", "") or ""
|
|
|
|
|
payload["subelement_name"] = waypoint.get("subelement_name", "") or ""
|
|
|
|
|
return payload
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
def _manual_route_nodes(
|
|
|
|
|
start_terminal,
|
|
|
|
|
end_terminal,
|
|
|
|
|
normalized_waypoints=None,
|
|
|
|
|
terminal_exit_length=0.0,
|
|
|
|
|
):
|
|
|
|
|
start_origin = TerminalObjects.terminal_origin(start_terminal)
|
|
|
|
|
end_origin = TerminalObjects.terminal_origin(end_terminal)
|
|
|
|
|
exit_length = max(float(terminal_exit_length or 0.0), 0.0)
|
|
|
|
|
|
|
|
|
|
nodes = [
|
|
|
|
|
_route_node_payload("start_terminal", start_origin, terminal=start_terminal)
|
|
|
|
|
]
|
|
|
|
|
if exit_length > 0:
|
|
|
|
|
nodes.append(
|
|
|
|
|
_route_node_payload(
|
|
|
|
|
"start_exit",
|
|
|
|
|
_offset_point(
|
|
|
|
|
start_origin,
|
|
|
|
|
_terminal_exit_direction(start_terminal),
|
|
|
|
|
exit_length,
|
|
|
|
|
),
|
|
|
|
|
terminal=start_terminal,
|
|
|
|
|
)
|
|
|
|
|
)
|
|
|
|
|
|
|
|
|
|
for index, waypoint in enumerate(normalized_waypoints or [], start=1):
|
|
|
|
|
nodes.append(
|
|
|
|
|
_route_node_payload(
|
|
|
|
|
"waypoint",
|
|
|
|
|
waypoint["point"],
|
|
|
|
|
waypoint=waypoint,
|
|
|
|
|
waypoint_index=index,
|
|
|
|
|
)
|
|
|
|
|
)
|
|
|
|
|
|
|
|
|
|
if exit_length > 0:
|
|
|
|
|
nodes.append(
|
|
|
|
|
_route_node_payload(
|
|
|
|
|
"end_exit",
|
|
|
|
|
_offset_point(
|
|
|
|
|
end_origin,
|
|
|
|
|
_terminal_exit_direction(end_terminal),
|
|
|
|
|
exit_length,
|
|
|
|
|
),
|
|
|
|
|
terminal=end_terminal,
|
|
|
|
|
)
|
|
|
|
|
)
|
|
|
|
|
nodes.append(_route_node_payload("end_terminal", end_origin, terminal=end_terminal))
|
|
|
|
|
return nodes
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
def _json_array_value(text):
|
|
|
|
|
if not text:
|
|
|
|
|
return []
|
|
|
|
|
try:
|
|
|
|
|
value = json.loads(text)
|
|
|
|
|
except Exception:
|
|
|
|
|
return []
|
|
|
|
|
if isinstance(value, list):
|
|
|
|
|
return value
|
|
|
|
|
return []
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
def _diagnostic(severity, code, message):
|
|
|
|
|
return {
|
|
|
|
|
"severity": severity,
|
|
|
|
|
"code": code,
|
|
|
|
|
"message": message,
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
def diagnose_manual_wire(wire_obj):
|
|
|
|
|
diagnostics = []
|
|
|
|
|
points = WiringObjects.wire_shape_points(wire_obj)
|
|
|
|
|
if len(points) < 2:
|
|
|
|
|
diagnostics.append(
|
|
|
|
|
_diagnostic("error", "wire_points_missing", "导线至少需要两个几何点。")
|
|
|
|
|
)
|
|
|
|
|
|
|
|
|
|
route_nodes = _json_array_value(getattr(wire_obj, "QetRouteNodesJson", ""))
|
|
|
|
|
if not route_nodes:
|
|
|
|
|
diagnostics.append(
|
|
|
|
|
_diagnostic("warning", "route_nodes_missing", "导线缺少语义路线节点。")
|
|
|
|
|
)
|
|
|
|
|
return diagnostics
|
|
|
|
|
|
|
|
|
|
roles = [str(node.get("role", "")) for node in route_nodes if isinstance(node, dict)]
|
|
|
|
|
if not roles or roles[0] != "start_terminal":
|
|
|
|
|
diagnostics.append(
|
|
|
|
|
_diagnostic("warning", "start_route_node_missing", "导线缺少起点端子路线节点。")
|
|
|
|
|
)
|
|
|
|
|
if not roles or roles[-1] != "end_terminal":
|
|
|
|
|
diagnostics.append(
|
|
|
|
|
_diagnostic("warning", "end_route_node_missing", "导线缺少终点端子路线节点。")
|
|
|
|
|
)
|
|
|
|
|
|
|
|
|
|
for node in route_nodes:
|
|
|
|
|
if not isinstance(node, dict):
|
|
|
|
|
continue
|
|
|
|
|
if node.get("role") != "waypoint":
|
|
|
|
|
continue
|
|
|
|
|
if node.get("carrier_kind") != "wire_duct":
|
|
|
|
|
continue
|
|
|
|
|
if not (node.get("source_object_name") or "").strip():
|
|
|
|
|
diagnostics.append(
|
|
|
|
|
_diagnostic(
|
|
|
|
|
"warning",
|
|
|
|
|
"wire_duct_source_missing",
|
|
|
|
|
"线槽折点缺少载体对象,不能可靠判断是否同一线槽。",
|
|
|
|
|
)
|
|
|
|
|
)
|
|
|
|
|
if not (node.get("carrier_axis") or "").strip():
|
|
|
|
|
diagnostics.append(
|
|
|
|
|
_diagnostic(
|
|
|
|
|
"warning",
|
|
|
|
|
"wire_duct_axis_missing",
|
|
|
|
|
"线槽折点缺少轴向信息,不能可靠沿线槽方向走线。",
|
|
|
|
|
)
|
|
|
|
|
)
|
|
|
|
|
return diagnostics
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
def _remove_from_group(group, obj):
|
|
|
|
|
if group is None or obj is None:
|
|
|
|
|
return
|
|
|
|
|
try:
|
|
|
|
|
if hasattr(group, "removeObject"):
|
|
|
|
|
group.removeObject(obj)
|
|
|
|
|
else:
|
|
|
|
|
group.Group = [candidate for candidate in group.Group if candidate is not obj]
|
|
|
|
|
except Exception:
|
|
|
|
|
pass
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
def _clear_manual_diagnostics(doc, diagnostic_group):
|
|
|
|
|
for obj in list(getattr(diagnostic_group, "Group", []) or []):
|
|
|
|
|
if (getattr(obj, "QetDiagnosticSource", "") or "").strip() != "ManualWiring":
|
|
|
|
|
continue
|
|
|
|
|
_remove_from_group(diagnostic_group, obj)
|
|
|
|
|
try:
|
|
|
|
|
if doc.getObject(getattr(obj, "Name", "")) is not None:
|
|
|
|
|
doc.removeObject(obj.Name)
|
|
|
|
|
except Exception:
|
|
|
|
|
pass
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
def _set_diagnostic_property(obj, prop_name, value, description):
|
|
|
|
|
TerminalObjects.ensure_string_property(
|
|
|
|
|
obj,
|
|
|
|
|
prop_name,
|
|
|
|
|
"QET Wiring Diagnostics",
|
|
|
|
|
description,
|
|
|
|
|
value,
|
|
|
|
|
)
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
def _create_diagnostic_object(doc, diagnostic_group, wire_obj, diagnostic, index):
|
|
|
|
|
name = "QETWireDiagnostic_{0}".format(index)
|
|
|
|
|
suffix = 1
|
|
|
|
|
base_name = name
|
|
|
|
|
while doc.getObject(name) is not None:
|
|
|
|
|
name = "{0}_{1}".format(base_name, suffix)
|
|
|
|
|
suffix += 1
|
|
|
|
|
|
|
|
|
|
obj = doc.addObject("App::FeaturePython", name)
|
|
|
|
|
message = diagnostic.get("message", "")
|
|
|
|
|
obj.Label = "{0}: {1}".format(diagnostic.get("severity", "warning"), message)
|
|
|
|
|
_set_diagnostic_property(obj, "QetDiagnosticSource", "ManualWiring", "Diagnostic source")
|
|
|
|
|
_set_diagnostic_property(obj, "QetDiagnosticSeverity", diagnostic.get("severity", ""), "Diagnostic severity")
|
|
|
|
|
_set_diagnostic_property(obj, "QetDiagnosticCode", diagnostic.get("code", ""), "Diagnostic code")
|
|
|
|
|
_set_diagnostic_property(obj, "QetDiagnosticMessage", message, "Diagnostic message")
|
|
|
|
|
_set_diagnostic_property(obj, "QetWireObjectName", getattr(wire_obj, "Name", ""), "Wire object name")
|
|
|
|
|
_set_diagnostic_property(obj, "QetWireLabel", getattr(wire_obj, "Label", ""), "Wire label")
|
|
|
|
|
diagnostic_group.addObject(obj)
|
|
|
|
|
return obj
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
def write_document_wire_diagnostics(doc, wires=None, project_uuid=""):
|
|
|
|
|
if doc is None:
|
|
|
|
|
raise ManualWiringError("No active FreeCAD document is available.")
|
|
|
|
|
|
|
|
|
|
if wires is None:
|
|
|
|
|
wires = WiringObjects.iter_routed_wire_objects(doc)
|
|
|
|
|
|
|
|
|
|
project_uuid = (
|
|
|
|
|
(project_uuid or "").strip()
|
|
|
|
|
or getattr(TerminalObjects.ensure_root_group(doc), "QetProjectUuid", "").strip()
|
|
|
|
|
)
|
|
|
|
|
diagnostic_group = WiringObjects.ensure_diagnostic_group(doc, project_uuid)
|
|
|
|
|
_clear_manual_diagnostics(doc, diagnostic_group)
|
|
|
|
|
|
|
|
|
|
created = []
|
|
|
|
|
issue_count = 0
|
|
|
|
|
for wire_obj in list(wires or []):
|
|
|
|
|
for diagnostic in diagnose_manual_wire(wire_obj):
|
|
|
|
|
issue_count += 1
|
|
|
|
|
created.append(
|
|
|
|
|
_create_diagnostic_object(
|
|
|
|
|
doc,
|
|
|
|
|
diagnostic_group,
|
|
|
|
|
wire_obj,
|
|
|
|
|
diagnostic,
|
|
|
|
|
issue_count,
|
|
|
|
|
)
|
|
|
|
|
)
|
|
|
|
|
|
|
|
|
|
try:
|
|
|
|
|
doc.recompute()
|
|
|
|
|
except Exception:
|
|
|
|
|
pass
|
|
|
|
|
return {
|
|
|
|
|
"wire_count": len(list(wires or [])),
|
|
|
|
|
"issue_count": issue_count,
|
|
|
|
|
"diagnostic_objects": created,
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
def _append_unique_point(points, point):
|
|
|
|
|
if not points or not _vector_close(points[-1], point):
|
|
|
|
|
points.append(point)
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
def _append_orthogonal_segment(points, target_point, preferred_axis=None):
|
|
|
|
|
def _append_orthogonal_segment(points, target_point, preferred_axis=None, leading_axis=None):
|
|
|
|
|
if not points:
|
|
|
|
|
points.append(target_point)
|
|
|
|
|
return
|
|
|
|
|
if leading_axis in {"x", "y", "z"}:
|
|
|
|
|
axis_order = [leading_axis] + [
|
|
|
|
|
axis for axis in ("x", "y", "z") if axis != leading_axis
|
|
|
|
|
]
|
|
|
|
|
segment = _orthogonal_segment_points_for_axis_order(
|
|
|
|
|
points[-1],
|
|
|
|
|
target_point,
|
|
|
|
|
axis_order,
|
|
|
|
|
)
|
|
|
|
|
else:
|
|
|
|
|
segment = _orthogonal_segment_points(
|
|
|
|
|
points[-1],
|
|
|
|
|
target_point,
|
|
|
|
|
@ -234,6 +505,47 @@ def _append_orthogonal_segment(points, target_point, preferred_axis=None):
|
|
|
|
|
_append_unique_point(points, point)
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
def _orthogonal_segment_points_for_axis_order(start_point, end_point, axis_order):
|
|
|
|
|
if _vector_close(start_point, end_point):
|
|
|
|
|
return [start_point]
|
|
|
|
|
|
|
|
|
|
points = [start_point]
|
|
|
|
|
current = start_point
|
|
|
|
|
for axis in axis_order:
|
|
|
|
|
if axis not in {"x", "y", "z"}:
|
|
|
|
|
continue
|
|
|
|
|
target = _axis_value(end_point, axis)
|
|
|
|
|
if abs(_axis_value(current, axis) - target) <= 0.000001:
|
|
|
|
|
continue
|
|
|
|
|
current = _vector_with_axis(current, axis, target)
|
|
|
|
|
if not _vector_close(current, points[-1]):
|
|
|
|
|
points.append(current)
|
|
|
|
|
|
|
|
|
|
if not _vector_close(points[-1], end_point):
|
|
|
|
|
points.append(end_point)
|
|
|
|
|
return points
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
def _same_carrier_run(left, right):
|
|
|
|
|
if not left or not right:
|
|
|
|
|
return None
|
|
|
|
|
if (left.get("carrier_kind") or "") != "wire_duct":
|
|
|
|
|
return None
|
|
|
|
|
if (right.get("carrier_kind") or "") != "wire_duct":
|
|
|
|
|
return None
|
|
|
|
|
left_source = (left.get("source_object_name") or "").strip()
|
|
|
|
|
right_source = (right.get("source_object_name") or "").strip()
|
|
|
|
|
if not left_source or not right_source:
|
|
|
|
|
return None
|
|
|
|
|
if left_source != right_source:
|
|
|
|
|
return None
|
|
|
|
|
left_axis = (left.get("carrier_axis") or "").strip().lower()
|
|
|
|
|
right_axis = (right.get("carrier_axis") or "").strip().lower()
|
|
|
|
|
if left_axis and left_axis == right_axis and left_axis in {"x", "y", "z"}:
|
|
|
|
|
return left_axis
|
|
|
|
|
return None
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
def _terminal_points(start_terminal, end_terminal, waypoints=None, terminal_exit_length=0.0):
|
|
|
|
|
start_origin = TerminalObjects.terminal_origin(start_terminal)
|
|
|
|
|
end_origin = TerminalObjects.terminal_origin(end_terminal)
|
|
|
|
|
@ -251,16 +563,20 @@ def _terminal_points(start_terminal, end_terminal, waypoints=None, terminal_exit
|
|
|
|
|
)
|
|
|
|
|
)
|
|
|
|
|
|
|
|
|
|
previous_waypoint = None
|
|
|
|
|
for point_like in waypoints or []:
|
|
|
|
|
waypoint = _coerce_waypoint(point_like)
|
|
|
|
|
if waypoint is None:
|
|
|
|
|
continue
|
|
|
|
|
normalized_waypoints.append(waypoint)
|
|
|
|
|
leading_axis = _same_carrier_run(previous_waypoint, waypoint)
|
|
|
|
|
_append_orthogonal_segment(
|
|
|
|
|
points,
|
|
|
|
|
waypoint["point"],
|
|
|
|
|
preferred_axis=waypoint.get("support_axis"),
|
|
|
|
|
leading_axis=leading_axis,
|
|
|
|
|
)
|
|
|
|
|
previous_waypoint = waypoint
|
|
|
|
|
|
|
|
|
|
if exit_length > 0:
|
|
|
|
|
end_exit = _offset_point(
|
|
|
|
|
@ -293,6 +609,8 @@ def _set_wire_properties(
|
|
|
|
|
wire_mark="",
|
|
|
|
|
wire_mark_is_manual=False,
|
|
|
|
|
manual_waypoints=None,
|
|
|
|
|
route_nodes=None,
|
|
|
|
|
terminal_exit_length=0.0,
|
|
|
|
|
):
|
|
|
|
|
WiringObjects.set_routed_wire_semantics(
|
|
|
|
|
obj,
|
|
|
|
|
@ -325,6 +643,28 @@ def _set_wire_properties(
|
|
|
|
|
)
|
|
|
|
|
except Exception:
|
|
|
|
|
pass
|
|
|
|
|
try:
|
|
|
|
|
if "QetRouteNodesJson" not in getattr(obj, "PropertiesList", []):
|
|
|
|
|
obj.addProperty(
|
|
|
|
|
"App::PropertyString",
|
|
|
|
|
"QetRouteNodesJson",
|
|
|
|
|
"QET Wiring",
|
|
|
|
|
"Manual route semantic nodes",
|
|
|
|
|
)
|
|
|
|
|
obj.QetRouteNodesJson = json.dumps(route_nodes or [], ensure_ascii=False)
|
|
|
|
|
except Exception:
|
|
|
|
|
pass
|
|
|
|
|
try:
|
|
|
|
|
if "QetTerminalExitLength" not in getattr(obj, "PropertiesList", []):
|
|
|
|
|
obj.addProperty(
|
|
|
|
|
"App::PropertyFloat",
|
|
|
|
|
"QetTerminalExitLength",
|
|
|
|
|
"QET Wiring",
|
|
|
|
|
"Terminal exit length in millimeters",
|
|
|
|
|
)
|
|
|
|
|
obj.QetTerminalExitLength = max(float(terminal_exit_length or 0.0), 0.0)
|
|
|
|
|
except Exception:
|
|
|
|
|
pass
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
def _set_wire_points(obj, points):
|
|
|
|
|
@ -447,6 +787,12 @@ def create_manual_wire(
|
|
|
|
|
waypoints=waypoints,
|
|
|
|
|
terminal_exit_length=terminal_exit_length,
|
|
|
|
|
)
|
|
|
|
|
route_nodes = _manual_route_nodes(
|
|
|
|
|
start_terminal,
|
|
|
|
|
end_terminal,
|
|
|
|
|
normalized_waypoints=normalized_waypoints,
|
|
|
|
|
terminal_exit_length=terminal_exit_length,
|
|
|
|
|
)
|
|
|
|
|
if len(points) < 2:
|
|
|
|
|
raise ManualWiringError("A wire requires at least two points.")
|
|
|
|
|
|
|
|
|
|
@ -464,6 +810,8 @@ def create_manual_wire(
|
|
|
|
|
wire_mark=wire_mark,
|
|
|
|
|
wire_mark_is_manual=wire_mark_is_manual,
|
|
|
|
|
manual_waypoints=normalized_waypoints,
|
|
|
|
|
route_nodes=route_nodes,
|
|
|
|
|
terminal_exit_length=terminal_exit_length,
|
|
|
|
|
)
|
|
|
|
|
|
|
|
|
|
if parent_group is None:
|
|
|
|
|
|