import math import re from pathlib import Path import FreeCAD as App import TerminalObjects try: import ImportGui except Exception: ImportGui = None class BatchAssemblyError(RuntimeError): pass TERMINAL_STRIP_NAME_PROPERTIES = ( "QetTerminalStripName", "QetTerminalBlockName", "QetTerminalGroupName", "QetStripName", "QetParentTerminalBlockName", ) TERMINAL_STRIP_ORDER_PROPERTIES = ( "QetTerminalStripIndex", "QetTerminalIndex", "QetTerminalSequence", "QetTerminalOrder", "QetTerminalNo", "QetTerminalDisplay", ) DEVICE_PREFIX_PROPERTIES = ( "QetDeviceTag", "QetDeviceName", "QetDisplayTag", "QetSymbolLabel", "QetInstanceId", "QetElementUuid", ) def _project_uuid(doc): try: root = TerminalObjects.ensure_root_group(doc) return (getattr(root, "QetProjectUuid", "") or "").strip() except Exception: return "" def _safe_label(text, fallback): value = str(text or "").strip() return value or fallback def _text_values(obj, include_children=False): values = [] for attr_name in ("Label", "Name"): value = (getattr(obj, attr_name, "") or "").strip() if value: values.append(value) for prop_name in DEVICE_PREFIX_PROPERTIES + TERMINAL_STRIP_NAME_PROPERTIES: value = (getattr(obj, prop_name, "") or "").strip() if value: values.append(value) if include_children: for child in list(getattr(obj, "Group", []) or []): for attr_name in ("Label", "Name"): value = (getattr(child, attr_name, "") or "").strip() if value: values.append(value) return values def _label_name_values(obj): values = [] for attr_name in ("Label", "Name"): value = (getattr(obj, attr_name, "") or "").strip() if value: values.append(value) return values def _natural_sort_key(value): text = str(value or "") key = [] for part in re.split(r"(\d+)", text): if part.isdigit(): key.append((0, int(part))) else: key.append((1, part.lower())) return key def _parse_strip_name_and_order(obj): for prop_name in TERMINAL_STRIP_NAME_PROPERTIES: strip_name = (getattr(obj, prop_name, "") or "").strip() if not strip_name: continue order = _explicit_order(obj) if order is None: order = _order_from_texts(_text_values(obj)) return strip_name, order for text in _label_name_values(obj): # Examples from QET trees: UD:1, UD-2, ID_006. match = re.match(r"^\s*([A-Za-z][A-Za-z0-9]{0,8})\s*[::_\-]\s*(\d+)\b", text) if match: return match.group(1), int(match.group(2)) return "", None def _explicit_order(obj): for prop_name in TERMINAL_STRIP_ORDER_PROPERTIES: value = (getattr(obj, prop_name, "") or "").strip() if not value: continue match = re.search(r"\d+", value) if match: return int(match.group(0)) return None def _order_from_texts(texts): for text in texts: match = re.search(r"(\d+)(?!.*\d)", str(text or "")) if match: return int(match.group(1)) return None def _is_group_like(obj): try: return bool(obj and obj.isDerivedFrom("App::DocumentObjectGroup")) except Exception: return bool(getattr(obj, "Group", None) is not None) def _qet_identity(obj): instance_id = (getattr(obj, "QetInstanceId", "") or "").strip() element_uuid = (getattr(obj, "QetElementUuid", "") or "").strip() return instance_id, element_uuid def _is_qet_device_object(obj): if obj is None: return False if TerminalObjects.is_terminal_object(obj): return False group_kind = (getattr(obj, "QetGroupKind", "") or "").strip() if group_kind in {TerminalObjects.TERMINAL_GROUP_KIND, TerminalObjects.WIRE_GROUP_KIND}: return False name = getattr(obj, "Name", "") or "" if name.startswith(TerminalObjects.TERMINAL_GROUP_PREFIX) or name.startswith(TerminalObjects.WIRE_GROUP_PREFIX): return False instance_id, element_uuid = _qet_identity(obj) if name.startswith(TerminalObjects.DEVICE_GROUP_PREFIX): return True if group_kind == "Device": return True return bool(instance_id or element_uuid) def _contains_terminal_slice_geometry(obj): text = " ".join(_text_values(obj, include_children=True)).lower() return any( token in text for token in ( "terminalslice", "terminal_slice", "terminal slice", "get_terminal_slice", "端子片", "端子排", ) ) def _contains_qet_terminal_group(obj): for child in list(getattr(obj, "Group", []) or []): if (getattr(child, "QetGroupKind", "") or "").strip() == TerminalObjects.TERMINAL_GROUP_KIND: return True if getattr(child, "Name", "").startswith(TerminalObjects.TERMINAL_GROUP_PREFIX): return True return False def _is_terminal_strip_device(obj): if not _is_qet_device_object(obj): return False strip_name, _order = _parse_strip_name_and_order(obj) if not strip_name: return False if _contains_terminal_slice_geometry(obj): return True if _contains_qet_terminal_group(obj): return True return False def _matches_prefix(obj, prefix): prefix = (prefix or "").strip().lower() if not prefix: return True for text in _text_values(obj): if text.lower().startswith(prefix): return True return False def _is_batch_generated(obj): return bool((getattr(obj, "QetBatchAssemblyKind", "") or "").strip()) def _existing_terminal_strip_devices(doc, strip_name=""): wanted = (strip_name or "").strip().lower() devices = [] for obj in list(getattr(doc, "Objects", []) or []): if not _is_terminal_strip_device(obj): continue current_strip, order = _parse_strip_name_and_order(obj) if wanted and current_strip.lower() != wanted: continue devices.append((current_strip, order, obj)) devices.sort( key=lambda item: ( item[0].lower(), item[1] if item[1] is not None else 10**9, _natural_sort_key(getattr(item[2], "Label", "") or getattr(item[2], "Name", "")), ) ) return [obj for _strip, _order, obj in devices] def available_terminal_strip_names(doc): names = [] seen = set() for obj in list(getattr(doc, "Objects", []) or []): if not _is_terminal_strip_device(obj): continue strip_name, _order = _parse_strip_name_and_order(obj) key = strip_name.lower() if strip_name and key not in seen: seen.add(key) names.append(strip_name) names.sort(key=_natural_sort_key) return names def _existing_devices_by_prefix(doc, prefix=""): devices = [] for obj in list(getattr(doc, "Objects", []) or []): if not _is_qet_device_object(obj): continue if _is_terminal_strip_device(obj): continue if _is_batch_generated(obj): continue if not _matches_prefix(obj, prefix): continue devices.append(obj) devices.sort(key=lambda obj: _natural_sort_key(getattr(obj, "Label", "") or getattr(obj, "Name", ""))) return devices def _axis_vector(rail): axis = (getattr(rail, "QetCarrierAxis", "") or "x").strip().lower() if axis == "y": vector = App.Vector(0, 1, 0) elif axis == "z": vector = App.Vector(0, 0, 1) else: vector = App.Vector(1, 0, 0) try: placement = getattr(rail, "Placement", None) rotation = getattr(placement, "Rotation", None) if rotation is not None and hasattr(rotation, "multVec"): vector = rotation.multVec(vector) except Exception: pass length = math.sqrt(vector.x * vector.x + vector.y * vector.y + vector.z * vector.z) if length <= 1e-9: return App.Vector(1, 0, 0) return App.Vector(vector.x / length, vector.y / length, vector.z / length) def _base_point(rail): placement = getattr(rail, "Placement", None) base = getattr(placement, "Base", None) if base is None: return App.Vector(0, 0, 0) return App.Vector(base.x, base.y, base.z) def _point_at(base, axis, offset): return App.Vector( base.x + axis.x * float(offset or 0.0), base.y + axis.y * float(offset or 0.0), base.z + axis.z * float(offset or 0.0), ) def _placement_at(rail, point): rotation = None try: rotation = getattr(getattr(rail, "Placement", None), "Rotation", None) except Exception: rotation = None return App.Placement(point, rotation or App.Rotation()) def _vector_copy(vector): return App.Vector( float(getattr(vector, "x", 0.0) or 0.0), float(getattr(vector, "y", 0.0) or 0.0), float(getattr(vector, "z", 0.0) or 0.0), ) def _vector_add(left, right): return App.Vector( float(getattr(left, "x", 0.0) or 0.0) + float(getattr(right, "x", 0.0) or 0.0), float(getattr(left, "y", 0.0) or 0.0) + float(getattr(right, "y", 0.0) or 0.0), float(getattr(left, "z", 0.0) or 0.0) + float(getattr(right, "z", 0.0) or 0.0), ) def _vector_sub(left, right): return App.Vector( float(getattr(left, "x", 0.0) or 0.0) - float(getattr(right, "x", 0.0) or 0.0), float(getattr(left, "y", 0.0) or 0.0) - float(getattr(right, "y", 0.0) or 0.0), float(getattr(left, "z", 0.0) or 0.0) - float(getattr(right, "z", 0.0) or 0.0), ) def _placement_base(obj): placement = getattr(obj, "Placement", None) base = getattr(placement, "Base", None) if base is None: return None return _vector_copy(base) def _placement_controls_children(obj): try: return bool(obj is not None and obj.isDerivedFrom("App::Part")) except Exception: return False def _iter_transform_children(obj): for child in list(getattr(obj, "Group", []) or []): yield child if not _placement_controls_children(child): for nested in _iter_transform_children(child): yield nested def _object_anchor_point(obj): if obj is None: return App.Vector(0, 0, 0) children = list(getattr(obj, "Group", []) or []) if _placement_controls_children(obj) or not children: return _placement_base(obj) or App.Vector(0, 0, 0) points = [] for child in _iter_transform_children(obj): base = _placement_base(child) if base is not None: points.append(base) if not points: return _placement_base(obj) or App.Vector(0, 0, 0) return App.Vector( sum(point.x for point in points) / len(points), sum(point.y for point in points) / len(points), sum(point.z for point in points) / len(points), ) def _translate_placement(obj, delta): placement = getattr(obj, "Placement", None) base = getattr(placement, "Base", None) if placement is None or base is None: return False new_base = _vector_add(base, delta) try: placement.Base = new_base obj.Placement = placement return True except Exception: try: obj.Placement = App.Placement(new_base, getattr(placement, "Rotation", App.Rotation())) return True except Exception: return False def _translate_group_children(obj, delta): moved = 0 for child in list(getattr(obj, "Group", []) or []): if _translate_placement(child, delta): moved += 1 if not _placement_controls_children(child): moved += _translate_group_children(child, delta) return moved def _ensure_rail(rail): if rail is None: raise BatchAssemblyError("请先选择一根导轨。") kind = (getattr(rail, "QetCarrierKind", "") or "").strip() if kind and kind != "rail": raise BatchAssemblyError("所选对象不是导轨。") return rail def _ensure_batch_root(doc, project_uuid=""): group = doc.getObject("QETBatchAssembly") if group is None: group = doc.addObject("App::DocumentObjectGroup", "QETBatchAssembly") group.Label = "QET Batch Assembly" if project_uuid: TerminalObjects.ensure_string_property( group, "QetProjectUuid", "QET Batch Assembly", "Project UUID", project_uuid, ) return group def _unique_object_name(doc, base_name): name = TerminalObjects.safe_token(base_name) or "QETObject" if doc.getObject(name) is None: return name suffix = 1 while doc.getObject("{0}_{1}".format(name, suffix)) is not None: suffix += 1 return "{0}_{1}".format(name, suffix) def _existing_object_names(doc): return {getattr(obj, "Name", "") for obj in list(getattr(doc, "Objects", []) or [])} def _new_objects_since(doc, before_names): return [ obj for obj in list(getattr(doc, "Objects", []) or []) if getattr(obj, "Name", "") not in before_names ] def _top_level_objects(objects): object_set = set(objects or []) result = [] for obj in objects or []: parents = set(getattr(obj, "InList", []) or []) if parents.intersection(object_set): continue result.append(obj) return result def _supported_model_path(path): suffix = Path(path or "").suffix.lower() return suffix in {".fcstd", ".step", ".stp"} def _set_source_model_path(obj, model_path): TerminalObjects.ensure_string_property( obj, "QetBatchSourceModelPath", "QET Batch Assembly", "Batch assembly imported model path", model_path, ) def _import_fcstd_objects(doc, path): if not hasattr(App, "openDocument") or not hasattr(doc, "copyObject"): return [] source_doc = None try: source_doc = App.openDocument(path, hidden=True, temporary=True) copied = [] for source_obj in _top_level_objects(list(getattr(source_doc, "Objects", []) or [])): copied.append(doc.copyObject(source_obj, True)) return copied finally: if source_doc is not None and hasattr(App, "closeDocument"): try: App.closeDocument(source_doc.Name) except Exception: pass def _import_step_objects(doc, path): if ImportGui is None: return [] before = _existing_object_names(doc) try: ImportGui.insert(name=path, docName=doc.Name, merge=False, useLinkGroup=True) except TypeError: ImportGui.insert(path, doc.Name) return _top_level_objects(_new_objects_since(doc, before)) def _import_model_objects(doc, model_path): path = str(model_path or "").strip() if not path: return [] if not _supported_model_path(path): raise BatchAssemblyError("请选择 STEP/STP/FCStd 模型文件。") if not Path(path).is_file(): raise BatchAssemblyError("模型文件不存在:{0}".format(path)) suffix = Path(path).suffix.lower() if suffix == ".fcstd": return _import_fcstd_objects(doc, path) return _import_step_objects(doc, path) def _set_batch_properties(obj, kind, batch_name, host): TerminalObjects.ensure_string_property( obj, "QetBatchAssemblyKind", "QET Batch Assembly", "Batch assembly kind", kind, ) TerminalObjects.ensure_string_property( obj, "QetBatchAssemblyName", "QET Batch Assembly", "Batch assembly name", batch_name, ) TerminalObjects.ensure_string_property( obj, "QetMountKind", "QET Mount", "Mount kind", "rail", ) TerminalObjects.ensure_string_property( obj, "QetMountHostName", "QET Mount", "Mount host object name", getattr(host, "Name", "") or "", ) TerminalObjects.ensure_string_property( obj, "QetMountHostKind", "QET Mount", "Mount host kind", (getattr(host, "QetCarrierKind", "") or "").strip(), ) def _set_layout_properties(obj, kind, batch_name, host, order_index, offset_mm): _set_batch_properties(obj, kind, batch_name, host) TerminalObjects.ensure_string_property( obj, "QetBatchAssemblyMode", "QET Batch Assembly", "Batch assembly mode", "layout_existing", ) TerminalObjects.ensure_string_property( obj, "QetBatchAssemblyOrder", "QET Batch Assembly", "Batch assembly order", str(int(order_index)), ) TerminalObjects.ensure_string_property( obj, "QetBatchAssemblyOffsetMm", "QET Batch Assembly", "Batch assembly offset in millimeters", "{0:.6f}".format(float(offset_mm or 0.0)), ) def _set_object_placement(obj, placement): if obj is not None and getattr(obj, "Group", None) and not _placement_controls_children(obj): current = _object_anchor_point(obj) target = getattr(placement, "Base", App.Vector(0, 0, 0)) delta = _vector_sub(target, current) moved_children = _translate_group_children(obj, delta) try: obj.Placement = placement except Exception: pass if moved_children: return True try: obj.Placement = placement return True except Exception: try: existing = getattr(obj, "Placement", None) if existing is not None: existing.Base = placement.Base existing.Rotation = placement.Rotation obj.Placement = existing return True except Exception: pass return False def _set_qet_device_properties(obj, project_uuid, element_uuid, instance_id): TerminalObjects.ensure_string_property( obj, "QetGroupKind", "QET Exchange", "FreeCADExchange group kind", "Device", ) TerminalObjects.ensure_string_property( obj, "QetProjectUuid", "QET Exchange", "Project UUID from QET exchange", project_uuid, ) TerminalObjects.ensure_string_property( obj, "QetElementUuid", "QET Exchange", "Parent element UUID from QET exchange", element_uuid, ) TerminalObjects.ensure_string_property( obj, "QetInstanceId", "QET Exchange", "Parent instance id from QET exchange", instance_id, ) def _create_device_group( doc, parent, label, placement, kind, batch_name, host, project_uuid, element_uuid, instance_id, ): group = doc.addObject( "App::DocumentObjectGroup", _unique_object_name(doc, "QETDevice_{0}".format(label)), ) group.Label = label try: group.Placement = placement except Exception: pass parent.addObject(group) _set_batch_properties(group, kind, batch_name, host) _set_qet_device_properties(group, project_uuid, element_uuid, instance_id) return group def _create_visual_placeholder(doc, device_group, label, placement, kind): try: body = doc.addObject("Part::Feature", "QETBatchBody_{0}".format(TerminalObjects.safe_token(label))) body.Label = "{0} 模型".format(label) body.Placement = placement try: import Part if kind == "breaker": body.Shape = Part.makeBox(18.0, 72.0, 80.0) else: body.Shape = Part.makeBox(5.2, 40.0, 45.0) except Exception: pass try: if kind == "breaker": body.ViewObject.ShapeColor = (0.78, 0.78, 0.72) else: body.ViewObject.ShapeColor = (0.35, 0.75, 0.35) except Exception: pass device_group.addObject(body) _set_batch_properties(body, kind, getattr(device_group, "Label", "") or label, device_group) return body except Exception: return None def _create_visual_model(doc, device_group, label, placement, kind, model_path=""): imported = _import_model_objects(doc, model_path) if imported: for index, obj in enumerate(imported): try: obj.Placement = placement except Exception: pass if index == 0: obj.Label = "{0} 模型".format(label) device_group.addObject(obj) _set_batch_properties(obj, kind, getattr(device_group, "Label", "") or label, device_group) _set_source_model_path(obj, model_path) return imported[0] return _create_visual_placeholder(doc, device_group, label, placement, kind) def _create_terminal( doc, device_group, project_uuid, element_uuid, instance_id, terminal_no, offset_index=0, label_prefix="", ): owner_label = _safe_label(label_prefix, getattr(device_group, "Label", "") or instance_id) label = "{0}:{1}".format(owner_label, terminal_no) base = getattr(getattr(device_group, "Placement", None), "Base", App.Vector()) terminal_point = App.Vector(base.x, base.y + float(offset_index) * 2.0, base.z) terminal = TerminalObjects.create_lcs_object( doc, "QETTerminal_{0}_{1}".format(TerminalObjects.safe_token(instance_id), TerminalObjects.safe_token(terminal_no)), placement=App.Placement(terminal_point, App.Rotation()), label=label, ) terminal_group = TerminalObjects.ensure_terminal_group( doc, device_group, project_uuid=project_uuid, instance_id=instance_id, ) terminal_group.addObject(terminal) TerminalObjects.set_terminal_semantics( terminal, project_uuid, element_uuid, "local:{0}:{1}".format(instance_id, terminal_no), instance_id, label=label, slot_name=str(terminal_no), ) TerminalObjects.hide_engineering_terminal(terminal) return terminal def _terminal_objects_for_devices(devices): terminals = [] for device in devices or []: try: terminals.extend(TerminalObjects.collect_terminal_objects(device)) except Exception: pass return terminals def _batch_report(kind, group, devices, terminals, source="fallback_created"): return { "kind": kind, "group": group, "devices": devices, "terminals": terminals, "created_devices": len(devices), "created_terminals": len(terminals), "updated_devices": 0, "updated_terminals": 0, "source": source, } def _layout_existing_objects( doc, rail, objects, kind, batch_name, pitch_mm, start_offset_mm, ): rail = _ensure_rail(rail) objects = [obj for obj in objects or [] if obj is not None] if not objects: return _batch_report(kind, rail, [], [], source="qet_existing") base = _base_point(rail) axis = _axis_vector(rail) updated = [] for index, obj in enumerate(objects): offset = float(start_offset_mm or 0.0) + index * float(pitch_mm or 0.0) placement = _placement_at(rail, _point_at(base, axis, offset)) if _set_object_placement(obj, placement): _set_layout_properties(obj, kind, batch_name, rail, index + 1, offset) updated.append(obj) try: doc.recompute() except Exception: pass terminals = _terminal_objects_for_devices(updated) return { "kind": kind, "group": rail, "devices": updated, "terminals": terminals, "created_devices": 0, "created_terminals": 0, "updated_devices": len(updated), "updated_terminals": len(terminals), "source": "qet_existing", } def layout_existing_terminal_block( doc, rail, block_name="", pitch_mm=5.2, start_offset_mm=0.0, ): if doc is None: raise BatchAssemblyError("请先打开 FreeCAD 工程。") rail = _ensure_rail(rail) devices = _existing_terminal_strip_devices(doc, block_name) if not devices: return _batch_report("terminal_block", rail, [], [], source="qet_existing") batch_name = _safe_label(block_name, _parse_strip_name_and_order(devices[0])[0] or "QET端子排") return _layout_existing_objects( doc, rail, devices, "terminal_block", batch_name, pitch_mm, start_offset_mm, ) def layout_existing_devices( doc, rail, prefix="QF", pitch_mm=18.0, start_offset_mm=0.0, kind="device_batch", ): if doc is None: raise BatchAssemblyError("请先打开 FreeCAD 工程。") rail = _ensure_rail(rail) devices = _existing_devices_by_prefix(doc, prefix) if not devices: return _batch_report(kind, rail, [], [], source="qet_existing") batch_name = _safe_label(prefix, "QET设备") return _layout_existing_objects( doc, rail, devices, kind, batch_name, pitch_mm, start_offset_mm, ) def create_terminal_block( doc, rail, block_name="XT1", count=10, pitch_mm=5.2, start_offset_mm=0.0, model_path="", ): if doc is None: raise BatchAssemblyError("请先打开 FreeCAD 工程。") rail = _ensure_rail(rail) count = int(count or 0) if count <= 0: raise BatchAssemblyError("端子数量必须大于 0。") project_uuid = _project_uuid(doc) batch_name = _safe_label(block_name, "XT1") root = _ensure_batch_root(doc, project_uuid) group = doc.addObject("App::DocumentObjectGroup", TerminalObjects.safe_token(batch_name)) group.Label = batch_name root.addObject(group) _set_batch_properties(group, "terminal_block", batch_name, rail) base = _base_point(rail) axis = _axis_vector(rail) devices = [] terminals = [] for index in range(count): terminal_no = str(index + 1) point = _point_at(base, axis, float(start_offset_mm or 0.0) + index * float(pitch_mm or 0.0)) device_label = "{0}_{1:03d}".format(batch_name, index + 1) device = _create_device_group( doc, group, device_label, _placement_at(rail, point), "terminal_slice", batch_name, rail, project_uuid, device_label, device_label, ) _create_visual_model(doc, device, device_label, _placement_at(rail, point), "terminal_slice", model_path) instance_id = device_label element_uuid = batch_name terminal = _create_terminal( doc, device, project_uuid, element_uuid, instance_id, terminal_no, label_prefix=batch_name, ) devices.append(device) terminals.append(terminal) try: doc.recompute() except Exception: pass return _batch_report("terminal_block", group, devices, terminals) def create_breakers( doc, rail, base_name="QF", count=3, pitch_mm=18.0, start_offset_mm=0.0, terminal_numbers=("1", "2", "3", "4", "5", "6"), model_path="", ): if doc is None: raise BatchAssemblyError("请先打开 FreeCAD 工程。") rail = _ensure_rail(rail) count = int(count or 0) if count <= 0: raise BatchAssemblyError("断路器数量必须大于 0。") terminal_numbers = [str(item).strip() for item in terminal_numbers or () if str(item).strip()] if not terminal_numbers: raise BatchAssemblyError("至少需要一个端子号。") project_uuid = _project_uuid(doc) batch_name = _safe_label(base_name, "QF") root = _ensure_batch_root(doc, project_uuid) group = doc.addObject("App::DocumentObjectGroup", "QETBatch_{0}".format(TerminalObjects.safe_token(batch_name))) group.Label = "{0} 批量断路器".format(batch_name) root.addObject(group) _set_batch_properties(group, "breaker_batch", batch_name, rail) base = _base_point(rail) axis = _axis_vector(rail) devices = [] terminals = [] for index in range(count): device_label = "{0}{1}".format(batch_name, index + 1) point = _point_at(base, axis, float(start_offset_mm or 0.0) + index * float(pitch_mm or 0.0)) device = _create_device_group( doc, group, device_label, _placement_at(rail, point), "breaker", batch_name, rail, project_uuid, device_label, device_label, ) _create_visual_model(doc, device, device_label, _placement_at(rail, point), "breaker", model_path) instance_id = device_label element_uuid = device_label for terminal_index, terminal_no in enumerate(terminal_numbers): terminals.append( _create_terminal( doc, device, project_uuid, element_uuid, instance_id, terminal_no, offset_index=terminal_index, ) ) devices.append(device) try: doc.recompute() except Exception: pass return _batch_report("breaker_batch", group, devices, terminals)