import json import os from pathlib import Path import uuid import json from datetime import datetime import FreeCAD as App try: import FreeCADGui as Gui except ImportError: Gui = None try: import ImportGui except ImportError: ImportGui = None import DevicePreview import TemplateSemantics import TerminalObjects ROOT_GROUP_NAME = "QETExchangeDevices" ROOT_GROUP_LABEL = "QET Exchange Devices" CABINET_MODEL_GROUP_NAME = "QETCabinetModel" CABINET_GROUP_PREFIX = "QETCabinet_" DEVICE_GROUP_PREFIX = "QETDevice_" TERMINAL_GROUP_PREFIX = "QETTerminals_" WIRE_GROUP_PREFIX = "QETWires_" GROUP_KIND_TERMINALS = "Terminals" GROUP_KIND_WIRES = "Wires" ASSEMBLY_STATE_PENDING = "Pending" ASSEMBLY_STATE_PLACED = "Placed" class DeviceImportError(RuntimeError): pass def _debug_log_path(): local_app_data = os.environ.get("LOCALAPPDATA", "").strip() if local_app_data: return os.path.join(local_app_data, "QETDeps", "freecad_exchange_bootstrap.log") return os.path.join(str(Path.home()), "AppData", "Local", "QETDeps", "freecad_exchange_bootstrap.log") def _append_debug_log(message): try: log_path = _debug_log_path() os.makedirs(os.path.dirname(log_path), exist_ok=True) with open(log_path, "a", encoding="utf-8") as handle: handle.write( "[{0}] {1}\n".format( datetime.now().astimezone().isoformat(timespec="seconds"), message, ) ) except Exception: pass def _safe_token(value): text = (value or "").strip() if not text: return "unknown" chars = [] for ch in text: if ch.isalnum(): chars.append(ch) else: chars.append("_") return "".join(chars) def _unique_object_name(doc, base_name): base = _safe_token(base_name) or "QETObject" if doc.getObject(base) is None: return base suffix = 1 while doc.getObject("{0}_{1}".format(base, suffix)) is not None: suffix += 1 return "{0}_{1}".format(base, suffix) def _native_path(value): text = (value or "").strip() if not text: return "" return os.path.normpath(os.path.expandvars(os.path.expanduser(text))) def _normalized_path_key(value): text = _native_path(value) if not text: return "" return os.path.normcase(os.path.normpath(text)) def _existing_object_names(doc): return {obj.Name for obj in doc.Objects} def _new_objects_since(doc, before_names): return [obj for obj in doc.Objects if obj.Name not in before_names] def _top_level_imported_objects(imported_objects): imported_by_name = {obj.Name: obj for obj in imported_objects} child_names = set() for obj in imported_objects: for parent in list(getattr(obj, "InList", []) or []): if getattr(parent, "Name", None) in imported_by_name: child_names.add(obj.Name) for child in list(getattr(obj, "Group", []) or []): if getattr(child, "Name", None) in imported_by_name: child_names.add(child.Name) return [obj for obj in imported_objects if obj.Name not in child_names] def _top_level_document_objects(doc): return _top_level_imported_objects(list(getattr(doc, "Objects", []) or [])) def _ensure_string_property(obj, prop_name, group_name, description, value): if prop_name not in getattr(obj, "PropertiesList", []): obj.addProperty("App::PropertyString", prop_name, group_name, description) setattr(obj, prop_name, value or "") def _ensure_bool_property(obj, prop_name, group_name, description, value): if prop_name not in getattr(obj, "PropertiesList", []): obj.addProperty("App::PropertyBool", prop_name, group_name, description) setattr(obj, prop_name, bool(value)) def _ensure_child_group(doc, parent_group, element_uuid, instance_id, name_prefix, label, group_kind, project_uuid=""): target_uuid = (element_uuid or "").strip() preferred_name = name_prefix + _safe_token(target_uuid) group = doc.getObject(preferred_name) if group is None: for candidate in getattr(parent_group, "Group", []) or []: if getattr(candidate, "QetGroupKind", "").strip() != group_kind: continue if target_uuid and getattr(candidate, "QetElementUuid", "").strip() != target_uuid: continue group = candidate break if group is None: group = doc.addObject("App::DocumentObjectGroup", preferred_name) if group not in getattr(parent_group, "Group", []): parent_group.addObject(group) group.Label = label project_uuid = (project_uuid or "").strip() or getattr(group, "QetProjectUuid", "").strip() element_uuid = (element_uuid or "").strip() or getattr(group, "QetElementUuid", "").strip() instance_id = (instance_id or "").strip() or getattr(group, "QetInstanceId", "").strip() _ensure_string_property( group, "QetGroupKind", "QET Exchange", "FreeCADExchange group kind", group_kind, ) _ensure_string_property( group, "QetProjectUuid", "QET Exchange", "Project UUID from QET exchange", project_uuid, ) _ensure_string_property( group, "QetElementUuid", "QET Exchange", "Parent element UUID from QET exchange", element_uuid, ) _ensure_string_property( group, "QetInstanceId", "QET Exchange", "Parent instance id from QET exchange", instance_id, ) return group def _ensure_document(scene_path): preferred_name = _safe_token(Path(scene_path).stem if scene_path else "QETScene")[:48] or "QETScene" normalized_scene_path = _native_path(scene_path) _append_debug_log( "DeviceImport _ensure_document: preferred_name={0}, normalized_scene_path={1}".format( preferred_name, normalized_scene_path or "", ) ) if normalized_scene_path and os.path.isfile(normalized_scene_path): normalized_target = os.path.normcase(os.path.normpath(normalized_scene_path)) for candidate in App.listDocuments().values(): candidate_path = getattr(candidate, "FileName", "") or "" if candidate_path and os.path.normcase(os.path.normpath(candidate_path)) == normalized_target: _append_debug_log( "DeviceImport _ensure_document reusing already open scene doc: name={0}, path={1}, objects={2}".format( getattr(candidate, "Name", ""), candidate_path, len(list(getattr(candidate, "Objects", []) or [])), ) ) _activate_document(candidate) return candidate try: doc = App.openDocument(normalized_scene_path) except Exception as exc: raise DeviceImportError( "Cannot open existing FreeCAD scene file: {0}".format(normalized_scene_path) ) from exc if doc is None: raise DeviceImportError( "Cannot open existing FreeCAD scene file: {0}".format(normalized_scene_path) ) _append_debug_log( "DeviceImport _ensure_document opened existing scene doc: name={0}, path={1}, objects={2}".format( getattr(doc, "Name", ""), getattr(doc, "FileName", "") or normalized_scene_path, len(list(getattr(doc, "Objects", []) or [])), ) ) _activate_document(doc) return doc existing_doc = DevicePreview.find_main_exchange_document(preferred_name) if existing_doc is not None: _append_debug_log( "DeviceImport _ensure_document reusing unsaved exchange doc: name={0}, path={1}, objects={2}".format( getattr(existing_doc, "Name", ""), getattr(existing_doc, "FileName", "") or "", len(list(getattr(existing_doc, "Objects", []) or [])), ) ) _activate_document(existing_doc) return existing_doc doc = App.newDocument(preferred_name) _append_debug_log( "DeviceImport _ensure_document created new scene doc: name={0}, path={1}, objects={2}".format( getattr(doc, "Name", ""), getattr(doc, "FileName", "") or "", len(list(getattr(doc, "Objects", []) or [])), ) ) _activate_document(doc) return doc def _activate_document(doc): if doc is None: return current_doc = getattr(App, "ActiveDocument", None) if current_doc is doc: _append_debug_log( "DeviceImport _activate_document skipped: already active name={0}, path={1}".format( getattr(doc, "Name", ""), getattr(doc, "FileName", "") or "", ) ) return _append_debug_log( "DeviceImport _activate_document: name={0}, path={1}, objects={2}".format( getattr(doc, "Name", ""), getattr(doc, "FileName", "") or "", len(list(getattr(doc, "Objects", []) or [])), ) ) setter = getattr(App, "setActiveDocument", None) if callable(setter): try: setter(doc.Name) except Exception: pass try: App.ActiveDocument = doc except Exception: pass try: Gui.ActiveDocument = Gui.getDocument(doc.Name) except Exception: pass def _cabinet_label_text(cabinet): if not isinstance(cabinet, dict): return "QET Cabinet" label = (cabinet.get("display_text") or "").strip() if label: return label label = (cabinet.get("label") or "").strip() if label: return label label = (cabinet.get("name") or "").strip() if label: return label return "QET Cabinet" def _ensure_root_group(doc, cabinet=None, project_uuid=""): root = doc.getObject(ROOT_GROUP_NAME) if root is None: root = doc.addObject("App::DocumentObjectGroup", ROOT_GROUP_NAME) if isinstance(cabinet, dict): root.Label = _cabinet_label_text(cabinet) else: root.Label = ROOT_GROUP_LABEL _ensure_string_property( root, "QetCabinetLabel", "QET Exchange", "Cabinet label from QET exchange", cabinet.get("label", "") if isinstance(cabinet, dict) else "", ) _ensure_string_property( root, "QetCabinetName", "QET Exchange", "Cabinet name from QET exchange", cabinet.get("name", "") if isinstance(cabinet, dict) else "", ) _ensure_string_property( root, "QetCabinetDisplayText", "QET Exchange", "Cabinet display text from QET exchange", cabinet.get("display_text", "") if isinstance(cabinet, dict) else "", ) _ensure_string_property( root, "QetCabinetFileSet", "QET Exchange", "Associated fileset from QET exchange", cabinet.get("associated_fileset", "") if isinstance(cabinet, dict) else "", ) _ensure_string_property( root, "QetCabinetRelativePath", "QET Exchange", "Relative 3D cabinet path from QET exchange", cabinet.get("three_d_relative_path", "") if isinstance(cabinet, dict) else "", ) _ensure_string_property( root, "QetCabinetResolvedScenePath", "QET Exchange", "Resolved local cabinet scene path from QET exchange", cabinet.get("resolved_scene_path", "") if isinstance(cabinet, dict) else "", ) _ensure_string_property( root, "QetCabinetLocationId", "QET Exchange", "Cabinet location id from QET exchange", str(cabinet.get("location_id") or "") if isinstance(cabinet, dict) else "", ) project_uuid = (project_uuid or "").strip() or getattr(root, "QetProjectUuid", "").strip() _ensure_string_property( root, "QetProjectUuid", "QET Exchange", "Project UUID from QET exchange", project_uuid, ) return root def _cabinet_instance_id(cabinet): if not isinstance(cabinet, dict): return "" for field_name in ("cabinet_instance_id", "cabinet_uuid", "location_id"): value = cabinet.get(field_name) if value is None: continue text = str(value).strip() if text: return text return "default" def _cabinet_group_label(cabinet): label = _cabinet_label_text(cabinet) if label: return label return "3D机柜" def _find_cabinet_group(doc, cabinet_instance_id): target_id = (cabinet_instance_id or "").strip() preferred_name = CABINET_GROUP_PREFIX + _safe_token(target_id or "default") group = doc.getObject(preferred_name) if group is not None: return group legacy_group = doc.getObject(CABINET_MODEL_GROUP_NAME) if legacy_group is not None: legacy_instance_id = getattr(legacy_group, "QetCabinetInstanceId", "").strip() if not target_id or not legacy_instance_id or legacy_instance_id == target_id: return legacy_group for candidate in getattr(doc, "Objects", []) or []: if "QetCabinetInstanceId" not in getattr(candidate, "PropertiesList", []): continue if getattr(candidate, "QetCabinetInstanceId", "").strip() == target_id: return candidate return None def _ensure_cabinet_model_group(doc, root_group, cabinet=None, project_uuid=""): cabinet_instance_id = _cabinet_instance_id(cabinet) group = _find_cabinet_group(doc, cabinet_instance_id) if group is None: group = doc.addObject( "App::DocumentObjectGroup", CABINET_GROUP_PREFIX + _safe_token(cabinet_instance_id or "default"), ) group.Label = _cabinet_group_label(cabinet) if group not in getattr(root_group, "Group", []): root_group.addObject(group) _ensure_string_property( group, "QetCabinetInstanceId", "QET Exchange", "Cabinet instance id from QET exchange", cabinet_instance_id, ) _ensure_string_property( group, "QetCabinetResolvedScenePath", "QET Exchange", "Resolved local cabinet scene path from QET exchange", cabinet.get("resolved_scene_path", "") if isinstance(cabinet, dict) else "", ) _ensure_string_property( group, "QetProjectUuid", "QET Exchange", "Project UUID from QET exchange", project_uuid, ) return group def _find_device_group(doc, element_uuid): target_uuid = (element_uuid or "").strip() if not target_uuid: return None preferred_name = DEVICE_GROUP_PREFIX + _safe_token(target_uuid) obj = doc.getObject(preferred_name) if obj is not None and getattr(obj, "Name", "").startswith(DEVICE_GROUP_PREFIX): return obj for candidate in doc.Objects: if not getattr(candidate, "Name", "").startswith(DEVICE_GROUP_PREFIX): continue try: if not candidate.isDerivedFrom("App::DocumentObjectGroup"): continue except Exception: continue if "QetElementUuid" in getattr(candidate, "PropertiesList", []): if getattr(candidate, "QetElementUuid", "").strip() == target_uuid: return candidate return None def _find_device_group_by_instance_id(doc, instance_id): target_instance_id = (instance_id or "").strip() if not target_instance_id: return None for candidate in doc.Objects: if not getattr(candidate, "Name", "").startswith(DEVICE_GROUP_PREFIX): continue if "QetInstanceId" not in getattr(candidate, "PropertiesList", []): continue if getattr(candidate, "QetInstanceId", "").strip() == target_instance_id: return candidate return None def _device_label_text(display_tag, instance_id, element_uuid): label = (display_tag or "").strip() if label: return label fallback = (instance_id or "").strip() or (element_uuid or "").strip() if fallback: return fallback return "QET Device" def _device_warning_subject(display_tag, element_uuid): label = (display_tag or "").strip() element_uuid = (element_uuid or "").strip() if label and element_uuid: return "设备 {0} ({1})".format(label, element_uuid) if label: return "设备 {0}".format(label) if element_uuid: return "设备 {0}".format(element_uuid) return "设备" def _device_report_label(display_tag, instance_id, element_uuid=""): label = (display_tag or "").strip() if label: return label fallback = (instance_id or "").strip() or (element_uuid or "").strip() return fallback or "未命名设备" def _payload_device_instance_id(device): if not isinstance(device, dict): return "" return (device.get("device_instance_id") or "").strip() def _payload_device_element_uuid(device): if not isinstance(device, dict): return "" for terminal in device.get("terminals", []) or []: if not isinstance(terminal, dict): continue element_uuid = (terminal.get("element_uuid") or "").strip() if element_uuid: return element_uuid return "" def _payload_terminal_uuid_set(device): result = set() if not isinstance(device, dict): return result for terminal in device.get("terminals", []) or []: if not isinstance(terminal, dict): continue terminal_uuid = (terminal.get("terminal_uuid") or "").strip() if terminal_uuid: result.add(terminal_uuid) return result def _terminal_signature_token(element_uuid, terminal_uuid): return "{0}|{1}".format((element_uuid or "").strip(), (terminal_uuid or "").strip()) def _payload_terminal_signature_counts(device): result = {} if not isinstance(device, dict): return result for terminal in device.get("terminals", []) or []: if not isinstance(terminal, dict): continue terminal_uuid = (terminal.get("terminal_uuid") or "").strip() element_uuid = (terminal.get("element_uuid") or "").strip() if not terminal_uuid: continue token = _terminal_signature_token(element_uuid, terminal_uuid) result[token] = result.get(token, 0) + 1 return result def _existing_qet_terminal_uuids(device_group): terminal_group = TerminalObjects.find_child_group_by_kind( device_group, TerminalObjects.TERMINAL_GROUP_KIND, ) result = set() for terminal_obj in TerminalObjects.collect_terminal_objects(terminal_group): terminal_uuid = (getattr(terminal_obj, "QetTerminalUuid", "") or "").strip() if not terminal_uuid or TerminalObjects.is_local_terminal_uuid(terminal_uuid): continue result.add(terminal_uuid) return result def _existing_qet_terminal_signature_counts(device_group): raw_json = (getattr(device_group, "QetPayloadTerminalSignaturesJson", "") or "").strip() if raw_json: try: parsed = json.loads(raw_json) if isinstance(parsed, dict): result = {} for key, value in parsed.items(): key_text = str(key or "").strip() if not key_text: continue try: count_value = int(value) except Exception: count_value = 0 if count_value > 0: result[key_text] = count_value if result: return result except Exception: pass terminal_group = TerminalObjects.find_child_group_by_kind( device_group, TerminalObjects.TERMINAL_GROUP_KIND, ) result = {} for terminal_obj in TerminalObjects.collect_terminal_objects(terminal_group): terminal_uuid = (getattr(terminal_obj, "QetTerminalUuid", "") or "").strip() element_uuid = (getattr(terminal_obj, "QetElementUuid", "") or "").strip() if not terminal_uuid or TerminalObjects.is_local_terminal_uuid(terminal_uuid): continue token = _terminal_signature_token(element_uuid, terminal_uuid) result[token] = result.get(token, 0) + 1 return result def _store_device_payload_terminal_signatures(device_group, signature_counts): normalized = {} for key, value in (signature_counts or {}).items(): key_text = str(key or "").strip() if not key_text: continue try: count_value = int(value) except Exception: count_value = 0 if count_value > 0: normalized[key_text] = count_value _ensure_string_property( device_group, "QetPayloadTerminalSignaturesJson", "QET Exchange", "Serialized terminal-entry signatures from the last QET payload", json.dumps(normalized, ensure_ascii=False, sort_keys=True), ) def _device_change_detail( display_tag, instance_id, element_uuid="", change_types=None, added_terminal_uuids=None, removed_terminal_uuids=None, previous_terminal_entry_count=0, current_terminal_entry_count=0, previous_display_tag="", previous_model_path="", resolved_model_path="", ): return { "display_tag": (display_tag or "").strip(), "instance_id": (instance_id or "").strip(), "element_uuid": (element_uuid or "").strip(), "label": _device_report_label(display_tag, instance_id, element_uuid), "change_types": list(change_types or []), "added_terminal_uuids": list(added_terminal_uuids or []), "removed_terminal_uuids": list(removed_terminal_uuids or []), "previous_terminal_entry_count": int(previous_terminal_entry_count or 0), "current_terminal_entry_count": int(current_terminal_entry_count or 0), "previous_display_tag": (previous_display_tag or "").strip(), "previous_model_path": (previous_model_path or "").strip(), "resolved_model_path": (resolved_model_path or "").strip(), } def _update_device_group_metadata(device_group, root_group, element_uuid, instance_id, model_path, display_tag): if device_group is None: return current_element_uuid = getattr(device_group, "QetElementUuid", "").strip() current_instance_id = getattr(device_group, "QetInstanceId", "").strip() current_model_path = getattr(device_group, "QetResolvedModelPath", "").strip() final_element_uuid = (element_uuid or "").strip() or current_element_uuid final_instance_id = (instance_id or "").strip() or current_instance_id final_model_path = (model_path or "").strip() or current_model_path final_display_tag = (display_tag or "").strip() device_group.Label = _device_label_text( final_display_tag, final_instance_id, final_element_uuid, ) _ensure_string_property( device_group, "QetElementUuid", "QET Exchange", "2D element UUID from QET", final_element_uuid, ) _ensure_string_property( device_group, "QetInstanceId", "QET Exchange", "3D instance id from QET/FreeCAD exchange", final_instance_id, ) _ensure_string_property( device_group, "QetResolvedModelPath", "QET Exchange", "Resolved local model path from QET exchange", final_model_path, ) _ensure_string_property( device_group, "QetDisplayTag", "QET Exchange", "2D display tag from QET exchange", final_display_tag, ) _ensure_string_property( device_group, "QetProjectUuid", "QET Exchange", "Project UUID from QET exchange", getattr(root_group, "QetProjectUuid", "").strip(), ) def _set_device_assembly_state(device_group, state): _ensure_string_property( device_group, "QetAssemblyState", "QET Assembly", "Assembly state in the FreeCAD scene.", state, ) def _ensure_device_group(doc, root_group, element_uuid, instance_id, model_path, display_tag, layout_index): created_now = False device_group = _find_device_group_by_instance_id(doc, instance_id) if device_group is None: device_group = _find_device_group(doc, element_uuid) if device_group is not None and getattr(device_group, "TypeId", "") != "App::Part": _remove_object_tree(doc, device_group) device_group = None if device_group is None: group_token = ( (element_uuid or "").strip() or (instance_id or "").strip() or (display_tag or "").strip() or "device-{0}".format(layout_index) ) device_group = doc.addObject( "App::Part", DEVICE_GROUP_PREFIX + _safe_token(group_token), ) created_now = True if device_group not in getattr(root_group, "Group", []): root_group.addObject(device_group) _update_device_group_metadata( device_group, root_group, element_uuid, instance_id, model_path, display_tag, ) _ensure_bool_property( device_group, "QetAutoPlaced", "QET Exchange", "Whether the device has been placed by the QET auto layout.", created_now, ) if created_now: device_group.Placement = App.Placement() _ensure_string_property( device_group, "QetProjectUuid", "QET Exchange", "Project UUID from QET exchange", getattr(root_group, "QetProjectUuid", "").strip(), ) _ensure_child_group( doc, device_group, element_uuid, instance_id, TERMINAL_GROUP_PREFIX, "QET Terminals", GROUP_KIND_TERMINALS, project_uuid=getattr(root_group, "QetProjectUuid", "").strip(), ) _ensure_child_group( doc, device_group, element_uuid, instance_id, WIRE_GROUP_PREFIX, "QET Wires", GROUP_KIND_WIRES, project_uuid=getattr(root_group, "QetProjectUuid", "").strip(), ) return device_group, created_now def _register_pending_device(report, device_group, display_tag, instance_id, element_uuid, resolved_model_path): _set_device_assembly_state(device_group, ASSEMBLY_STATE_PENDING) report.setdefault("pending_devices", 0) report.setdefault("pending_device_details", []) report["pending_devices"] += 1 report["pending_device_details"].append( _device_change_detail( display_tag, instance_id, element_uuid=element_uuid, change_types=["待装配"], resolved_model_path=resolved_model_path, ) ) def _looks_like_qet_device_group(obj): if obj is None: return False if not getattr(obj, "Name", "").startswith(DEVICE_GROUP_PREFIX): return False return bool((getattr(obj, "QetInstanceId", "") or "").strip()) def _find_device_group_from_object(obj): if _looks_like_qet_device_group(obj): return obj pending = list(getattr(obj, "InList", []) or []) seen = set() while pending: parent = pending.pop(0) parent_name = getattr(parent, "Name", "") if parent_name in seen: continue seen.add(parent_name) if _looks_like_qet_device_group(parent): return parent pending.extend(list(getattr(parent, "InList", []) or [])) return None def list_pending_devices(doc): if doc is None: return [] root = doc.getObject(ROOT_GROUP_NAME) if root is None: return [] pending_devices = [] for child in list(getattr(root, "Group", []) or []): if not _looks_like_qet_device_group(child): continue if (getattr(child, "QetAssemblyState", "") or "").strip() != ASSEMBLY_STATE_PENDING: continue pending_devices.append( { "device": child, "instance_id": (getattr(child, "QetInstanceId", "") or "").strip(), "element_uuid": (getattr(child, "QetElementUuid", "") or "").strip(), "display_tag": (getattr(child, "QetDisplayTag", "") or "").strip(), "label": getattr(child, "Label", "") or getattr(child, "Name", ""), "resolved_model_path": ( getattr(child, "QetResolvedModelPath", "") or "" ).strip(), } ) return pending_devices def _target_mount_kind(target_obj): if target_obj is None: return "" kind = (getattr(target_obj, "QetCarrierKind", "") or "").strip() if kind: return kind text = " ".join( [ getattr(target_obj, "Label", "") or "", getattr(target_obj, "Name", "") or "", getattr(target_obj, "TypeId", "") or "", ] ).lower() if "rail" in text or "din" in text or "导轨" in text: return "rail" if "wireduct" in text or "wire_duct" in text or "线槽" in text: return "wire_duct" if "plate" in text or "panel" in text or "安装板" in text or "面板" in text: return "mounting_plate" if "cabinet" in text or "柜" in text: return "cabinet" return "" def _placement_for_mount_target(mount_target, fallback_rotation=None): placement = getattr(mount_target, "Placement", None) base = getattr(placement, "Base", None) if placement is None or base is None: return None rotation = getattr(placement, "Rotation", None) or fallback_rotation or App.Rotation() return App.Placement(base, rotation) def _vector_payload(vector): return { "x": float(getattr(vector, "x", 0.0) or 0.0), "y": float(getattr(vector, "y", 0.0) or 0.0), "z": float(getattr(vector, "z", 0.0) or 0.0), } def _normalized_vector(vector): if vector is None: return None x = float(getattr(vector, "x", 0.0) or 0.0) y = float(getattr(vector, "y", 0.0) or 0.0) z = float(getattr(vector, "z", 0.0) or 0.0) length = (x * x + y * y + z * z) ** 0.5 if length <= 1e-9: return None return App.Vector(x / length, y / length, z / length) def _placement_with_normal_offset(placement, normal=None, offset_mm=0.0): if placement is None: return None normal = _normalized_vector(normal) if normal is None or not float(offset_mm or 0.0): return placement base = getattr(placement, "Base", None) if base is None: return placement offset = float(offset_mm or 0.0) moved_base = App.Vector( float(getattr(base, "x", 0.0) or 0.0) + normal.x * offset, float(getattr(base, "y", 0.0) or 0.0) + normal.y * offset, float(getattr(base, "z", 0.0) or 0.0) + normal.z * offset, ) return App.Placement(moved_base, getattr(placement, "Rotation", App.Rotation())) def _set_device_mount_metadata(device_group, mount_target, normal=None, offset_mm=0.0): if device_group is None or mount_target is None: return target_name = getattr(mount_target, "Name", "") or "" target_label = getattr(mount_target, "Label", "") or target_name _ensure_string_property( device_group, "QetMountMode", "QET Mount", "How this QET device was mounted in the FreeCAD scene.", "manual_insert", ) _ensure_string_property( device_group, "QetMountHostName", "QET Mount", "Mount target object name.", target_name, ) _ensure_string_property( device_group, "QetMountHostLabel", "QET Mount", "Mount target object label.", target_label, ) _ensure_string_property( device_group, "QetMountHostKind", "QET Mount", "Mount target kind.", _target_mount_kind(mount_target), ) normal = _normalized_vector(normal) if normal is not None: _ensure_string_property( device_group, "QetMountHostNormalJson", "QET Mount", "Mount target face normal at insert time.", json.dumps(_vector_payload(normal), sort_keys=True), ) _ensure_string_property( device_group, "QetMountOffsetMm", "QET Mount", "Mount offset in target normal direction.", "{0:.6f}".format(float(offset_mm or 0.0)), ) def insert_pending_device( doc, device_group, source_doc_cache=None, mount_target=None, mount_placement=None, mount_normal=None, mount_offset_mm=0.0, ): if doc is None: raise DeviceImportError("A FreeCAD document is required.") device_group = _find_device_group_from_object(device_group) if device_group is None: raise DeviceImportError("请选择一个待装配 QET 设备。") model_path = _native_path(getattr(device_group, "QetResolvedModelPath", "")) if not model_path: raise DeviceImportError("待装配设备缺少模型路径。") if not os.path.isfile(model_path): raise DeviceImportError("待装配设备模型文件不存在：{0}".format(model_path)) if not _supported_for_import(model_path): raise DeviceImportError("待装配设备模型格式暂不支持：{0}".format(model_path)) existing_model_objects = _existing_model_objects(doc, device_group) if existing_model_objects: _set_device_assembly_state(device_group, ASSEMBLY_STATE_PLACED) target_placement = mount_placement or _placement_for_mount_target( mount_target, getattr(getattr(device_group, "Placement", None), "Rotation", None), ) target_placement = _placement_with_normal_offset( target_placement, mount_normal, mount_offset_mm, ) if target_placement is not None: device_group.Placement = target_placement _set_device_mount_metadata( device_group, mount_target, normal=mount_normal, offset_mm=mount_offset_mm, ) return { "device": device_group, "imported_objects": existing_model_objects, "already_placed": True, } _clear_group_contents(doc, device_group) imported_objects = _import_model_into_group( doc, device_group, model_path, source_doc_cache=source_doc_cache if source_doc_cache is not None else {}, ) target_placement = mount_placement or _placement_for_mount_target( mount_target, getattr(getattr(device_group, "Placement", None), "Rotation", None), ) target_placement = _placement_with_normal_offset( target_placement, mount_normal, mount_offset_mm, ) if target_placement is not None: device_group.Placement = target_placement _set_device_mount_metadata( device_group, mount_target, normal=mount_normal, offset_mm=mount_offset_mm, ) _set_device_assembly_state(device_group, ASSEMBLY_STATE_PLACED) try: doc.recompute() except Exception: pass return { "device": device_group, "imported_objects": list(imported_objects or []), "already_placed": False, } def _remove_object_tree(doc, obj): if obj is None: return obj_name = _object_name(obj) if not obj_name or doc.getObject(obj_name) is None: _detach_from_parent_groups(obj) return children = list(getattr(obj, "Group", []) or []) for child in children: _remove_object_tree(doc, child) _detach_from_parent_groups(obj) if doc.getObject(obj_name) is not None: doc.removeObject(obj_name) def _object_name(obj): try: return getattr(obj, "Name", "") except Exception: return "" def _object_exists(doc, obj): obj_name = _object_name(obj) return bool(obj_name and doc.getObject(obj_name) is not None) def _detach_from_parent_groups(obj): try: parents = list(getattr(obj, "InList", []) or []) except Exception: return for parent in parents: try: group_children = list(getattr(parent, "Group", []) or []) except Exception: continue if obj not in group_children: continue remover = getattr(parent, "removeObject", None) if callable(remover): try: remover(obj) continue except Exception: pass try: while obj in getattr(parent, "Group", []): parent.Group.remove(obj) except Exception: pass def _linked_document_objects(value): if value is None: return [] if hasattr(value, "Name") and hasattr(value, "TypeId"): return [value] if isinstance(value, dict): result = [] for item in value.values(): result.extend(_linked_document_objects(item)) return result if isinstance(value, (list, tuple, set)): result = [] for item in value: result.extend(_linked_document_objects(item)) return result return [] def _remove_template_terminal_hint_object(doc, obj): linked_objects = [] for prop_name in ("OriginFeatures",): linked_objects.extend(_linked_document_objects(getattr(obj, prop_name, None))) _remove_object_tree(doc, obj) for linked_obj in linked_objects: if linked_obj is not obj: _remove_object_tree(doc, linked_obj) def _remove_template_terminal_hints(doc, container): removed = 0 if container is None: return removed for child in list(getattr(container, "Group", []) or []): if TerminalObjects.is_template_terminal_object(child): _remove_template_terminal_hint_object(doc, child) removed += 1 continue if hasattr(child, "Group"): removed += _remove_template_terminal_hints(doc, child) return removed def _clear_group_contents(doc, group): for child in list(getattr(group, "Group", []) or []): child_name = getattr(child, "Name", "") if child_name.startswith(TERMINAL_GROUP_PREFIX) or child_name.startswith(WIRE_GROUP_PREFIX): continue if getattr(child, "QetGroupKind", "").strip() in {GROUP_KIND_TERMINALS, GROUP_KIND_WIRES}: continue _remove_object_tree(doc, child) def _existing_group_objects(doc, group): result = [] for child in list(getattr(group, "Group", []) or []): if _object_exists(doc, child): result.append(child) return result def _existing_model_objects(doc, group): return [ child for child in _existing_group_objects(doc, group) if not _is_exchange_sidecar_group(child) ] def _is_exchange_sidecar_group(obj): child_name = _object_name(obj) if child_name.startswith(TERMINAL_GROUP_PREFIX) or child_name.startswith(WIRE_GROUP_PREFIX): return True try: if TerminalObjects.is_terminal_hint_object(obj): return True except Exception: pass return getattr(obj, "QetGroupKind", "").strip() in {GROUP_KIND_TERMINALS, GROUP_KIND_WIRES} def _existing_model_objects(doc, group): return [ obj for obj in _existing_group_objects(doc, group) if not _is_exchange_sidecar_group(obj) ] def _remove_model_objects(doc, objects): for obj in list(objects or []): _remove_object_tree(doc, obj) def _keep_only_direct_model_children(device_group, direct_model_objects): allowed_ids = {id(obj) for obj in direct_model_objects if obj is not None} kept_children = [] for child in list(getattr(device_group, "Group", []) or []): if id(child) in allowed_ids: kept_children.append(child) continue if _is_exchange_sidecar_group(child): kept_children.append(child) try: device_group.Group = kept_children except Exception: try: device_group.Group[:] = kept_children except Exception: pass def _document_object_links(obj): linked_objects = [] for prop_name in ("Links", "OutList"): linked_objects.extend(_linked_document_objects(getattr(obj, prop_name, None))) return linked_objects def _shape_copy(shape): copier = getattr(shape, "copy", None) if callable(copier): try: return copier() except Exception: pass return shape def _can_materialize_shape(obj): if getattr(obj, "TypeId", "") == "Part::Feature": return False try: shape = getattr(obj, "Shape", None) except Exception: return False return shape is not None def _materialize_shape_object(doc, source_obj): source_name = _object_name(source_obj) or "Model" target = doc.addObject( "Part::Feature", _unique_object_name(doc, source_name + "_Shape"), ) target.Label = getattr(source_obj, "Label", source_name) target.Shape = _shape_copy(getattr(source_obj, "Shape", None)) try: target.Placement = getattr(source_obj, "Placement") except Exception: pass try: target.ViewObject.Visibility = getattr(source_obj.ViewObject, "Visibility", True) except Exception: pass try: target.ViewObject.ShapeColor = getattr(source_obj.ViewObject, "ShapeColor") except Exception: pass return target def _materialize_direct_model_objects(doc, device_group, model_objects): direct_objects = [] obsolete_objects = [] materialized_labels = [] for obj in model_objects: if not _object_exists(doc, obj): continue if not _can_materialize_shape(obj): direct_objects.append(obj) continue static_obj = _materialize_shape_object(doc, obj) if static_obj not in getattr(device_group, "Group", []): device_group.addObject(static_obj) direct_objects.append(static_obj) materialized_labels.append((static_obj, getattr(obj, "Label", _object_name(obj)))) obsolete_objects.append(obj) obsolete_objects.extend(_document_object_links(obj)) direct_names = {_object_name(obj) for obj in direct_objects} removed_names = set() for obsolete in obsolete_objects: obsolete_name = _object_name(obsolete) if not obsolete_name or obsolete_name in direct_names or obsolete_name in removed_names: continue _remove_object_tree(doc, obsolete) removed_names.add(obsolete_name) for obj, label in materialized_labels: try: obj.Label = label except Exception: pass return direct_objects def _generate_instance_id(project_uuid, element_uuid): seed = "QET:{0}:{1}".format((project_uuid or "").strip(), (element_uuid or "").strip()) return str(uuid.uuid5(uuid.NAMESPACE_URL, seed)) def _supported_for_import(model_path): suffix = Path(model_path).suffix.lower() return suffix in { ".step", ".stp", ".iges", ".igs", ".brep", ".brp", ".fcstd", } def _import_model_into_group( doc, device_group, model_path, merge=False, use_link_group=True, source_doc_cache=None, ): if Path(model_path).suffix.lower() == ".fcstd": return _import_fcstd_into_group( doc, device_group, model_path, source_doc_cache=source_doc_cache, ) before_names = _existing_object_names(doc) try: try: ImportGui.insert( name=model_path, docName=doc.Name, merge=bool(merge), useLinkGroup=bool(use_link_group), ) except Exception: for obj in _new_objects_since(doc, before_names): _remove_object_tree(doc, obj) raise imported_objects = _new_objects_since(doc, before_names) top_level_objects = _top_level_imported_objects(imported_objects) for obj in top_level_objects: if obj not in getattr(device_group, "Group", []): device_group.addObject(obj) TemplateSemantics.clear_stored_template_slot_hints(device_group) TerminalObjects.hide_template_terminal_hints(device_group) return top_level_objects finally: _activate_document(doc) def _open_fcstd_source_document(model_path, source_doc_cache=None): normalized_target = os.path.normcase(os.path.normpath(model_path)) if source_doc_cache is not None: cached_entry = source_doc_cache.get(normalized_target) if cached_entry is not None: cached_doc = cached_entry.get("doc") if cached_doc is not None: _append_debug_log( "DeviceImport _open_fcstd_source_document cache hit: name={0}, path={1}, objects={2}, should_close={3}".format( getattr(cached_doc, "Name", ""), getattr(cached_doc, "FileName", "") or model_path, len(list(getattr(cached_doc, "Objects", []) or [])), bool(cached_entry.get("should_close")), ) ) return cached_doc, False for candidate in App.listDocuments().values(): candidate_path = getattr(candidate, "FileName", "") or "" if candidate_path and os.path.normcase(os.path.normpath(candidate_path)) == normalized_target: _append_debug_log( "DeviceImport _open_fcstd_source_document reusing open source doc: name={0}, path={1}, objects={2}".format( getattr(candidate, "Name", ""), candidate_path, len(list(getattr(candidate, "Objects", []) or [])), ) ) if source_doc_cache is not None: source_doc_cache[normalized_target] = { "doc": candidate, "should_close": False, } return candidate, False source_doc = App.openDocument(model_path, hidden=True, temporary=True) _append_debug_log( "DeviceImport _open_fcstd_source_document opened temp source doc: name={0}, path={1}, objects={2}".format( getattr(source_doc, "Name", "") if source_doc is not None else "", getattr(source_doc, "FileName", "") if source_doc is not None else model_path, len(list(getattr(source_doc, "Objects", []) or [])) if source_doc is not None else -1, ) ) if source_doc_cache is not None and source_doc is not None: source_doc_cache[normalized_target] = { "doc": source_doc, "should_close": True, } return source_doc, False return source_doc, True def _import_fcstd_into_group(doc, device_group, model_path, source_doc_cache=None): source_doc = None should_close = False try: _append_debug_log( "DeviceImport _import_fcstd_into_group start: target_doc={0}, target_objects={1}, device_group={2}, model_path={3}".format( getattr(doc, "Name", ""), len(list(getattr(doc, "Objects", []) or [])), getattr(device_group, "Name", ""), model_path, ) ) source_doc, should_close = _open_fcstd_source_document( model_path, source_doc_cache=source_doc_cache, ) if source_doc is None: raise DeviceImportError("Cannot open FCStd file") TemplateSemantics.clear_stored_template_slot_hints(device_group) top_level_objects = _top_level_document_objects(source_doc) _append_debug_log( "DeviceImport _import_fcstd_into_group source ready: source_doc={0}, top_level_objects={1}, should_close={2}".format( getattr(source_doc, "Name", ""), len(top_level_objects), should_close, ) ) copied_objects = [] for source_obj in top_level_objects: copied_obj = doc.copyObject(source_obj, True) if copied_obj not in getattr(device_group, "Group", []): device_group.addObject(copied_obj) copied_objects.append(copied_obj) template_slots = TemplateSemantics.collect_live_terminal_hints(device_group) TemplateSemantics.store_template_slot_hints(device_group, template_slots) _remove_template_terminal_hints(doc, device_group) copied_model_objects = [ obj for obj in copied_objects if _object_exists(doc, obj) ] direct_model_objects = _materialize_direct_model_objects( doc, device_group, copied_model_objects, ) _keep_only_direct_model_children(device_group, direct_model_objects) _append_debug_log( "DeviceImport _import_fcstd_into_group completed: copied_objects={0}, direct_model_objects={1}, target_doc_objects={2}".format( len(copied_objects), len(direct_model_objects), len(list(getattr(doc, "Objects", []) or [])), ) ) return direct_model_objects finally: if should_close and source_doc is not None: try: _append_debug_log( "DeviceImport _import_fcstd_into_group closing temp source doc: name={0}, path={1}".format( getattr(source_doc, "Name", ""), getattr(source_doc, "FileName", "") or model_path, ) ) App.closeDocument(source_doc.Name) except Exception: pass _activate_document(doc) def _close_cached_source_documents(source_doc_cache, target_doc=None): if not source_doc_cache: return closed_count = 0 for normalized_target, cached_entry in list(source_doc_cache.items()): cached_doc = cached_entry.get("doc") should_close = bool(cached_entry.get("should_close")) if not should_close or cached_doc is None: continue try: _append_debug_log( "DeviceImport _close_cached_source_documents closing cached doc: name={0}, path={1}".format( getattr(cached_doc, "Name", ""), getattr(cached_doc, "FileName", "") or normalized_target, ) ) App.closeDocument(cached_doc.Name) closed_count += 1 except Exception as exc: _append_debug_log( "DeviceImport _close_cached_source_documents failed: name={0}, error={1}".format( getattr(cached_doc, "Name", ""), exc, ) ) source_doc_cache.clear() _append_debug_log( "DeviceImport _close_cached_source_documents completed: closed_count={0}".format( closed_count ) ) if target_doc is not None: _activate_document(target_doc) def _model_index(payload): index = {} for item in payload.get("device_models", []): instance_id = (item.get("device_instance_id") or "").strip() if instance_id and instance_id not in index: index[instance_id] = item return index def _import_cabinet_model(doc, root_group, cabinet, report, source_doc_cache=None): if not isinstance(cabinet, dict): return resolved_scene_path = _native_path(cabinet.get("resolved_scene_path", "")) _append_debug_log( "DeviceImport cabinet resolved_scene_path={0}".format(resolved_scene_path) ) if not resolved_scene_path: report["cabinet_skipped_missing_model"] += 1 return if not os.path.isfile(resolved_scene_path): report["cabinet_skipped_missing_file"] += 1 report["warnings"].append( "机柜 3D 文件不存在：{0}".format(resolved_scene_path) ) return if not _supported_for_import(resolved_scene_path): report["cabinet_skipped_unsupported_format"] += 1 report["warnings"].append( "机柜 3D 文件格式暂不支持：{0}".format(resolved_scene_path) ) return project_uuid = getattr(root_group, "QetProjectUuid", "").strip() existing_group = _find_cabinet_group(doc, _cabinet_instance_id(cabinet)) previous_path = "" if existing_group is not None: previous_path = getattr(existing_group, "QetCabinetResolvedScenePath", "").strip() cabinet_group = _ensure_cabinet_model_group(doc, root_group, cabinet, project_uuid) existing_model_objects = _existing_model_objects(doc, cabinet_group) same_source = _normalized_path_key(previous_path) == _normalized_path_key(resolved_scene_path) if existing_model_objects and same_source: report.setdefault("cabinet_reused", 0) report["cabinet_reused"] += 1 _append_debug_log( "DeviceImport cabinet import skipped: reused existing cabinet group for instance_id={0}".format( getattr(cabinet_group, "QetCabinetInstanceId", "").strip() ) ) return had_existing_model = bool(existing_model_objects) _ensure_string_property( cabinet_group, "QetCabinetResolvedScenePath", "QET Exchange", "Resolved local cabinet scene path from QET exchange", resolved_scene_path, ) try: _append_debug_log( "DeviceImport importing cabinet model: {0}".format( resolved_scene_path ) ) _import_model_into_group( doc, cabinet_group, resolved_scene_path, merge=False, use_link_group=True, source_doc_cache=source_doc_cache, ) _remove_model_objects(doc, existing_model_objects) report["cabinet_imported"] += 1 if had_existing_model: report.setdefault("cabinet_reimported", 0) report["cabinet_reimported"] += 1 else: report.setdefault("cabinet_added", 0) report["cabinet_added"] += 1 _append_debug_log("DeviceImport cabinet import succeeded") except Exception as exc: if had_existing_model: _ensure_string_property( cabinet_group, "QetCabinetResolvedScenePath", "QET Exchange", "Resolved local cabinet scene path from QET exchange", previous_path, ) report["cabinet_skipped_import_error"] += 1 report["warnings"].append( "机柜 3D 导入失败：{0}".format(exc) ) _append_debug_log( "DeviceImport cabinet import failed: {0}".format(exc) ) def import_devices_from_payload(payload, scene_path="", auto_insert_pending_devices=False): _append_debug_log("DeviceImport.import_devices_from_payload entered") doc = _ensure_document(scene_path) cabinet = payload.get("cabinet") project_uuid = (payload.get("project_uuid") or "").strip() root_group = _ensure_root_group(doc, cabinet, project_uuid) models_by_element = _model_index(payload) source_doc_cache = {} report = { "document_name": doc.Name, "scene_path": scene_path or "", "total_devices": 0, "imported_devices": 0, "updated_devices": 0, "reused_devices": 0, "added_device_details": [], "updated_device_details": [], "reused_device_details": [], "imported_without_instance_id": 0, "skipped_missing_model": 0, "skipped_missing_file": 0, "skipped_unsupported_format": 0, "skipped_import_error": 0, "cabinet_imported": 0, "cabinet_added": 0, "cabinet_reimported": 0, "cabinet_reused": 0, "cabinet_skipped_missing_model": 0, "cabinet_skipped_missing_file": 0, "cabinet_skipped_unsupported_format": 0, "cabinet_skipped_import_error": 0, "pending_devices": 0, "pending_device_details": [], "warnings": [], } try: _import_cabinet_model( doc, root_group, cabinet, report, source_doc_cache=source_doc_cache, ) for index, device in enumerate(payload.get("devices", [])): report["total_devices"] += 1 original_instance_id = _payload_device_instance_id(device) instance_id = original_instance_id element_uuid = _payload_device_element_uuid(device) display_tag = (device.get("display_tag") or "").strip() payload_terminal_uuids = _payload_terminal_uuid_set(device) payload_terminal_signature_counts = _payload_terminal_signature_counts(device) payload_terminal_entry_count = sum(payload_terminal_signature_counts.values()) existing_device_group = _find_device_group_by_instance_id(doc, instance_id) if existing_device_group is None: existing_device_group = _find_device_group(doc, element_uuid) previous_display_tag = "" previous_path = "" existing_terminal_uuids = set() existing_terminal_signature_counts = {} existing_terminal_entry_count = 0 existing_model_objects = [] if existing_device_group is not None: previous_display_tag = getattr( existing_device_group, "QetDisplayTag", "", ).strip() previous_path = getattr( existing_device_group, "QetResolvedModelPath", "", ).strip() existing_terminal_uuids = _existing_qet_terminal_uuids( existing_device_group ) existing_terminal_signature_counts = _existing_qet_terminal_signature_counts( existing_device_group ) existing_terminal_entry_count = sum( existing_terminal_signature_counts.values() ) existing_model_objects = _existing_model_objects( doc, existing_device_group ) model_info = models_by_element.get(instance_id, {}) resolved_model_path = _native_path(model_info.get("resolved_model_path", "")) _append_debug_log( "DeviceImport device instance_id={0}, display_tag={1}, resolved_model_path={2}".format( instance_id, display_tag, resolved_model_path ) ) if not resolved_model_path: display_tag_changed = bool( existing_device_group is not None and previous_display_tag != display_tag ) terminals_changed = bool( existing_device_group is not None and payload_terminal_signature_counts != existing_terminal_signature_counts ) if existing_device_group is not None: _update_device_group_metadata( existing_device_group, root_group, element_uuid, instance_id, previous_path, display_tag, ) _store_device_payload_terminal_signatures( existing_device_group, payload_terminal_signature_counts, ) if display_tag_changed: change_types = ["标注"] if terminals_changed: change_types.append("端子") report["updated_devices"] += 1 report["updated_device_details"].append( _device_change_detail( display_tag, (instance_id or getattr(existing_device_group, "QetInstanceId", "")).strip(), element_uuid=element_uuid, change_types=change_types, previous_terminal_entry_count=existing_terminal_entry_count, current_terminal_entry_count=payload_terminal_entry_count, previous_display_tag=previous_display_tag, previous_model_path=previous_path, resolved_model_path=previous_path, ) ) elif terminals_changed: report["updated_devices"] += 1 report["updated_device_details"].append( _device_change_detail( display_tag, (instance_id or getattr(existing_device_group, "QetInstanceId", "")).strip(), element_uuid=element_uuid, change_types=["端子"], added_terminal_uuids=sorted( payload_terminal_uuids - existing_terminal_uuids ), removed_terminal_uuids=sorted( existing_terminal_uuids - payload_terminal_uuids ), previous_terminal_entry_count=existing_terminal_entry_count, current_terminal_entry_count=payload_terminal_entry_count, previous_display_tag=previous_display_tag, previous_model_path=previous_path, resolved_model_path=previous_path, ) ) report["skipped_missing_model"] += 1 report["warnings"].append( "{0} 缺少 resolved_model_path，已跳过。".format( _device_warning_subject(display_tag, instance_id) ) ) continue if not os.path.isfile(resolved_model_path): report["skipped_missing_file"] += 1 report["warnings"].append( "{0} 的模型文件不存在：{1}".format( _device_warning_subject(display_tag, instance_id), resolved_model_path, ) ) continue if not _supported_for_import(resolved_model_path): report["skipped_unsupported_format"] += 1 report["warnings"].append( "{0} 的模型格式暂不支持：{1}".format( _device_warning_subject(display_tag, instance_id), resolved_model_path, ) ) continue if not instance_id: instance_id = _generate_instance_id( project_uuid, display_tag or element_uuid or "device-{0}".format(index) ) report.setdefault("generated_instance_ids", 0) report["generated_instance_ids"] += 1 device_group, created_now = _ensure_device_group( doc, root_group, element_uuid, instance_id, resolved_model_path, display_tag, index, ) same_source = ( _normalized_path_key(previous_path) == _normalized_path_key(resolved_model_path) ) added_terminal_uuids = sorted( payload_terminal_uuids - existing_terminal_uuids ) removed_terminal_uuids = sorted( existing_terminal_uuids - payload_terminal_uuids ) terminals_changed = bool( added_terminal_uuids or removed_terminal_uuids or payload_terminal_signature_counts != existing_terminal_signature_counts ) display_tag_changed = ( not created_now and previous_display_tag != display_tag ) model_changed = ( not created_now and (not existing_model_objects or not same_source) ) if not auto_insert_pending_devices and not existing_model_objects: _register_pending_device( report, device_group, display_tag, instance_id, element_uuid, resolved_model_path, ) _append_debug_log( "DeviceImport registered pending device without importing model: instance_id={0}, model_path={1}".format( instance_id, resolved_model_path, ) ) continue if existing_model_objects and same_source: _set_device_assembly_state(device_group, ASSEMBLY_STATE_PLACED) if display_tag_changed or terminals_changed: change_types = [] if display_tag_changed: change_types.append("标注") if terminals_changed: change_types.append("端子") report["updated_devices"] += 1 report["updated_device_details"].append( _device_change_detail( display_tag, instance_id, element_uuid=element_uuid, change_types=change_types, added_terminal_uuids=added_terminal_uuids, removed_terminal_uuids=removed_terminal_uuids, previous_terminal_entry_count=existing_terminal_entry_count, current_terminal_entry_count=payload_terminal_entry_count, previous_display_tag=previous_display_tag, previous_model_path=previous_path, resolved_model_path=resolved_model_path, ) ) _append_debug_log( "DeviceImport import skipped: metadata-only change for instance_id={0}, display_tag_changed={1}, added_terminals={2}, removed_terminals={3}".format( instance_id, display_tag_changed, len(added_terminal_uuids), len(removed_terminal_uuids), ) ) _store_device_payload_terminal_signatures( device_group, payload_terminal_signature_counts, ) continue report["reused_devices"] += 1 report["reused_device_details"].append( _device_change_detail( display_tag, instance_id, element_uuid=element_uuid, previous_terminal_entry_count=existing_terminal_entry_count, current_terminal_entry_count=payload_terminal_entry_count, previous_display_tag=previous_display_tag, previous_model_path=previous_path, resolved_model_path=resolved_model_path, ) ) _append_debug_log( "DeviceImport import skipped: reused existing device group for instance_id={0}, model_path={1}, existing_model_objects={2}".format( instance_id, resolved_model_path, len(existing_model_objects), ) ) _store_device_payload_terminal_signatures( device_group, payload_terminal_signature_counts, ) continue if created_now or not existing_model_objects: _clear_group_contents(doc, device_group) try: _append_debug_log( "DeviceImport importing model for device_instance_id={0}: {1}".format( instance_id, resolved_model_path ) ) _import_model_into_group( doc, device_group, resolved_model_path, source_doc_cache=source_doc_cache, ) _append_debug_log( "DeviceImport import succeeded for device_instance_id={0}".format( instance_id ) ) if existing_model_objects: _remove_model_objects(doc, existing_model_objects) _set_device_assembly_state(device_group, ASSEMBLY_STATE_PLACED) except Exception as exc: if existing_model_objects: _ensure_string_property( device_group, "QetResolvedModelPath", "QET Exchange", "Resolved local model path from QET exchange", previous_path, ) report["skipped_import_error"] += 1 report["warnings"].append( "{0} 导入失败：{1}".format( _device_warning_subject(display_tag, element_uuid or instance_id), exc, ) ) _append_debug_log( "DeviceImport import failed for device_instance_id={0}: {1}".format( instance_id, exc ) ) continue if created_now: report["imported_devices"] += 1 report["added_device_details"].append( _device_change_detail( display_tag, instance_id, element_uuid=element_uuid, previous_terminal_entry_count=existing_terminal_entry_count, current_terminal_entry_count=payload_terminal_entry_count, previous_display_tag=previous_display_tag, previous_model_path=previous_path, resolved_model_path=resolved_model_path, ) ) else: report["updated_devices"] += 1 change_types = [] if display_tag_changed: change_types.append("标注") if model_changed: change_types.append("3D模型") if terminals_changed: change_types.append("端子") if not change_types: change_types.append("3D模型") report["updated_device_details"].append( _device_change_detail( display_tag, instance_id, element_uuid=element_uuid, change_types=change_types, added_terminal_uuids=added_terminal_uuids, removed_terminal_uuids=removed_terminal_uuids, previous_terminal_entry_count=existing_terminal_entry_count, current_terminal_entry_count=payload_terminal_entry_count, previous_display_tag=previous_display_tag, previous_model_path=previous_path, resolved_model_path=resolved_model_path, ) ) _store_device_payload_terminal_signatures( device_group, payload_terminal_signature_counts, ) if not original_instance_id: report["imported_without_instance_id"] += 1 finally: _close_cached_source_documents(source_doc_cache, target_doc=doc) TerminalObjects.sort_group_children(root_group) doc.recompute() _append_debug_log("DeviceImport ViewFit skipped during exchange import") _append_debug_log( "DeviceImport finished: cabinet_imported={0}, imported={1}, updated={2}, reused={3}, skipped_missing_model={4}, skipped_missing_file={5}, skipped_import_error={6}".format( report["cabinet_imported"], report["imported_devices"], report["updated_devices"], report["reused_devices"], report["skipped_missing_model"], report["skipped_missing_file"], report["skipped_import_error"], ) ) return report class CommandInsertPendingDevice: def GetResources(self): return { "MenuText": "插入待装配设备", "ToolTip": "将选中的 QET 待装配设备模型插入当前 3D 场景", } def IsActive(self): return getattr(App, "ActiveDocument", None) is not None and Gui is not None def Activated(self): if Gui is None: return selection = list(Gui.Selection.getSelection() or []) device_group = None for obj in selection: device_group = _find_device_group_from_object(obj) if device_group is not None: break if device_group is None: try: App.Console.PrintWarning("[FreeCADExchange] 请先选择一个待装配 QET 设备。\n") except Exception: pass return try: result = insert_pending_device(App.ActiveDocument, device_group) try: App.Console.PrintMessage( "[FreeCADExchange] 已插入设备：{0}，导入对象 {1} 个。\n".format( getattr(result["device"], "Label", ""), len(result.get("imported_objects", []) or []), ) ) except Exception: pass try: Gui.SendMsgToActiveView("ViewFit") except Exception: pass except Exception as exc: try: App.Console.PrintError("[FreeCADExchange] 插入待装配设备失败：{0}\n".format(exc)) except Exception: pass _COMMANDS_REGISTERED = False def register_commands(): global _COMMANDS_REGISTERED if _COMMANDS_REGISTERED: return if Gui is None or not hasattr(Gui, "addCommand"): return try: Gui.addCommand("QET_Exchange_InsertPendingDevice", CommandInsertPendingDevice()) _COMMANDS_REGISTERED = True except Exception as exc: _append_debug_log("failed to register pending device command: {0}".format(exc))