From a9a8ef5b1c332592294ba41d0a3a87e69af80628 Mon Sep 17 00:00:00 2001 From: zhanghao <2024138486@qq.com> Date: Fri, 12 Jun 2026 17:02:42 +0800 Subject: [PATCH] =?UTF-8?q?feature/=E4=BC=98=E5=8C=962d-3d=E6=95=B0?= =?UTF-8?q?=E6=8D=AE=E4=BA=A4=E6=8D=A2-0612?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- src/Mod/FreeCADExchange/DeviceImport.py | 154 +++++++++++++++++- src/Mod/FreeCADExchange/ExchangeBootstrap.py | 83 +++++++++- .../freecad_exchange_bootstrap_wiring_test.py | 27 ++- ...eecad_exchange_device_import_fcstd_test.py | 94 ++++++++++- 4 files changed, 347 insertions(+), 11 deletions(-) diff --git a/src/Mod/FreeCADExchange/DeviceImport.py b/src/Mod/FreeCADExchange/DeviceImport.py index a0e4643..878e397 100644 --- a/src/Mod/FreeCADExchange/DeviceImport.py +++ b/src/Mod/FreeCADExchange/DeviceImport.py @@ -1,3 +1,4 @@ +import json import os from pathlib import Path import uuid @@ -559,6 +560,26 @@ def _payload_terminal_uuid_set(device): 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, @@ -573,6 +594,64 @@ def _existing_qet_terminal_uuids(device_group): 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, @@ -580,6 +659,8 @@ def _device_change_detail( 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="", @@ -592,6 +673,8 @@ def _device_change_detail( "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(), @@ -1365,12 +1448,16 @@ def import_devices_from_payload(payload, scene_path=""): 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( @@ -1386,6 +1473,12 @@ def import_devices_from_payload(payload, scene_path=""): 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 ) @@ -1404,6 +1497,10 @@ def import_devices_from_payload(payload, scene_path=""): 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, @@ -1413,14 +1510,44 @@ def import_devices_from_payload(payload, scene_path=""): 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=["标注"], + 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, @@ -1481,7 +1608,9 @@ def import_devices_from_payload(payload, scene_path=""): existing_terminal_uuids - payload_terminal_uuids ) terminals_changed = bool( - added_terminal_uuids or removed_terminal_uuids + 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 @@ -1506,6 +1635,8 @@ def import_devices_from_payload(payload, scene_path=""): 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, @@ -1519,6 +1650,10 @@ def import_devices_from_payload(payload, scene_path=""): 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( @@ -1526,6 +1661,8 @@ def import_devices_from_payload(payload, scene_path=""): 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, @@ -1538,6 +1675,10 @@ def import_devices_from_payload(payload, scene_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: @@ -1592,6 +1733,8 @@ def import_devices_from_payload(payload, scene_path=""): 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, @@ -1616,12 +1759,19 @@ def import_devices_from_payload(payload, scene_path=""): 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: diff --git a/src/Mod/FreeCADExchange/ExchangeBootstrap.py b/src/Mod/FreeCADExchange/ExchangeBootstrap.py index 4a89645..782d36d 100644 --- a/src/Mod/FreeCADExchange/ExchangeBootstrap.py +++ b/src/Mod/FreeCADExchange/ExchangeBootstrap.py @@ -1099,19 +1099,44 @@ def _mark_stale_objects(payload): def _summary_message(summary, import_report=None, terminal_report=None, writeback_report=None, wiring_report=None, stale_report=None): + def _top_detail_labels(items, limit=5): + values = [] + for item in list(items or []): + if not isinstance(item, dict): + continue + label = ( + item.get("display_tag", "") + or item.get("label", "") + or item.get("instance_id", "") + or item.get("element_uuid", "") + ) + label = str(label or "").strip() + if label and label not in values: + values.append(label) + if len(values) >= limit: + break + if not values: + return "" + return " ({0})".format("、".join(values)) + lines = [ "QET exchange file loaded successfully.", ] if import_report or stale_report: lines.append("") + added_device_details = import_report.get("added_device_details", []) if import_report else [] updated_device_details = import_report.get("updated_device_details", []) if import_report else [] + stale_device_details = stale_report.get("stale_device_details", []) if stale_report else [] if summary.get("is_first_open"): lines.extend( [ "同步模式：首次打开（全量导入）", "新增机柜：{0}".format(import_report.get("cabinet_added", 0) if import_report else 0), - "新增设备：{0}".format(import_report.get("imported_devices", 0) if import_report else 0), + "新增设备：{0}{1}".format( + import_report.get("imported_devices", 0) if import_report else 0, + _top_detail_labels(added_device_details), + ), "更新设备：{0}".format(import_report.get("updated_devices", 0) if import_report else 0), ] ) @@ -1121,14 +1146,31 @@ def _summary_message(summary, import_report=None, terminal_report=None, writebac "同步模式：再次打开（增量更新）", "新增机柜：{0}".format(import_report.get("cabinet_added", 0) if import_report else 0), "失效机柜：{0}".format(stale_report.get("stale_cabinets", 0) if stale_report else 0), - "新增设备：{0}".format(import_report.get("imported_devices", 0) if import_report else 0), + "新增设备：{0}{1}".format( + import_report.get("imported_devices", 0) if import_report else 0, + _top_detail_labels(added_device_details), + ), "更新设备：{0}".format(import_report.get("updated_devices", 0) if import_report else 0), - "失效设备：{0}".format(stale_report.get("stale_devices", 0) if stale_report else 0), + "失效设备：{0}{1}".format( + stale_report.get("stale_devices", 0) if stale_report else 0, + _top_detail_labels(stale_device_details), + ), ] ) + if added_device_details: + lines.append("新增设备明细：") + for item in added_device_details[:10]: + lines.append( + "- {0} [{1}]".format( + item.get("label", "") or item.get("display_tag", "") or item.get("instance_id", ""), + item.get("instance_id", "") or "", + ) + ) + if len(added_device_details) > 10: + lines.append("- ... ({0} more)".format(len(added_device_details) - 10)) if updated_device_details: lines.append("修改设备：") - for item in updated_device_details[:5]: + for item in updated_device_details[:10]: change_types = " + ".join(item.get("change_types", []) or []) or "未知变化" detail_bits = [] if "标注" in (item.get("change_types", []) or []): @@ -1141,6 +1183,15 @@ def _summary_message(summary, import_report=None, terminal_report=None, writebac added_terms = item.get("added_terminal_uuids", []) or [] removed_terms = item.get("removed_terminal_uuids", []) or [] detail_bits.append("+{0}/-{1} 端子".format(len(added_terms), len(removed_terms))) + previous_terminal_entry_count = item.get("previous_terminal_entry_count", 0) or 0 + current_terminal_entry_count = item.get("current_terminal_entry_count", 0) or 0 + if previous_terminal_entry_count != current_terminal_entry_count: + detail_bits.append( + "端子条目 {0} -> {1}".format( + previous_terminal_entry_count, + current_terminal_entry_count, + ) + ) detail_suffix = "" if detail_bits: detail_suffix = " ({0})".format(", ".join(detail_bits)) @@ -1152,8 +1203,19 @@ def _summary_message(summary, import_report=None, terminal_report=None, writebac detail_suffix, ) ) - if len(updated_device_details) > 5: - lines.append("- ... ({0} more)".format(len(updated_device_details) - 5)) + if len(updated_device_details) > 10: + lines.append("- ... ({0} more)".format(len(updated_device_details) - 10)) + if stale_device_details: + lines.append("失效设备明细：") + for item in stale_device_details[:10]: + lines.append( + "- {0} [{1}]".format( + item.get("label", "") or item.get("display_tag", "") or item.get("instance_id", ""), + item.get("instance_id", "") or "", + ) + ) + if len(stale_device_details) > 10: + lines.append("- ... ({0} more)".format(len(stale_device_details) - 10)) lines.extend( [ @@ -1245,6 +1307,15 @@ def _summary_message(summary, import_report=None, terminal_report=None, writebac if "端子" in (item.get("change_types", []) or []): terminal_bits.append("+{0}".format(len(added_terms))) terminal_bits.append("-{0}".format(len(removed_terms))) + previous_terminal_entry_count = item.get("previous_terminal_entry_count", 0) or 0 + current_terminal_entry_count = item.get("current_terminal_entry_count", 0) or 0 + if previous_terminal_entry_count != current_terminal_entry_count: + terminal_bits.append( + "entries {0} -> {1}".format( + previous_terminal_entry_count, + current_terminal_entry_count, + ) + ) if "标注" in (item.get("change_types", []) or []): previous_display_tag = item.get("previous_display_tag", "") or "" current_display_tag = item.get("display_tag", "") or "" diff --git a/tests/python/freecad_exchange_bootstrap_wiring_test.py b/tests/python/freecad_exchange_bootstrap_wiring_test.py index 502d785..a6d1ffe 100644 --- a/tests/python/freecad_exchange_bootstrap_wiring_test.py +++ b/tests/python/freecad_exchange_bootstrap_wiring_test.py @@ -341,10 +341,16 @@ class ExchangeBootstrapWiringTest(unittest.TestCase): "cabinet_added": 0, "cabinet_reimported": 0, "cabinet_reused": 1, - "imported_devices": 0, + "imported_devices": 1, "updated_devices": 1, "reused_devices": 0, - "added_device_details": [], + "added_device_details": [ + { + "label": "QF2", + "display_tag": "QF2", + "instance_id": "device-inst-2", + } + ], "updated_device_details": [ { "label": "J3", @@ -363,10 +369,27 @@ class ExchangeBootstrapWiringTest(unittest.TestCase): "skipped_import_error": 0, "warnings": [], }, + stale_report={ + "stale_cabinets": 0, + "stale_devices": 1, + "stale_device_details": [ + { + "label": "J9", + "display_tag": "J9", + "instance_id": "device-inst-9", + } + ], + }, ) + self.assertIn("新增设备明细：", message) + self.assertIn("新增设备：1 (QF2)", message) + self.assertIn("QF2 [device-inst-2]", message) self.assertIn("更新设备：1", message) self.assertIn("修改设备：", message) + self.assertIn("失效设备：1 (J9)", message) + self.assertIn("失效设备明细：", message) + self.assertIn("J9 [device-inst-9]", message) self.assertIn("Updated device details:", message) self.assertIn("J3 [device-inst-1] -> 标注 (标注 J1 -> J3)", message) diff --git a/tests/python/freecad_exchange_device_import_fcstd_test.py b/tests/python/freecad_exchange_device_import_fcstd_test.py index 30dcf48..247dded 100644 --- a/tests/python/freecad_exchange_device_import_fcstd_test.py +++ b/tests/python/freecad_exchange_device_import_fcstd_test.py @@ -1114,6 +1114,96 @@ class FcstdDeviceImportTest(unittest.TestCase): self.assertEqual("J1", report["updated_device_details"][0]["previous_display_tag"]) self.assertEqual("J3", report["updated_device_details"][0]["display_tag"]) + def test_import_devices_from_payload_reports_terminal_entry_count_change_without_model_reimport(self): + with tempfile.TemporaryDirectory() as temp_dir: + model_path = Path(temp_dir) / "device.step" + model_path.write_text("fake step placeholder", encoding="utf-8") + _install_fake_freecad(None) + + device_import, _ = _reload_modules() + terminal_objects = importlib.import_module("TerminalObjects") + + doc = FakeDocument("QETScene") + doc.recompute = lambda: None + device_import._ensure_document = lambda scene_path: doc + + root = device_import._ensure_root_group(doc, None, "project-1") + device_group, _ = device_import._ensure_device_group( + doc, + root, + "element-a", + "device-inst-1", + str(model_path), + "J1", + 0, + ) + existing_body = doc.addObject("Part::Feature", "ExistingBody") + device_group.addObject(existing_body) + + terminal_group = terminal_objects.ensure_terminal_group( + doc, + device_group, + project_uuid="project-1", + instance_id="device-inst-1", + ) + existing_terminal = terminal_objects.create_lcs_object( + doc, + "QETTerminal_terminal_shared", + label="terminal-shared", + ) + terminal_group.addObject(existing_terminal) + terminal_objects.set_terminal_semantics( + existing_terminal, + "project-1", + "element-a", + "terminal-shared", + "device-inst-1", + label="terminal-shared", + slot_name="terminal-shared", + ) + + import_calls = [] + + def fake_import_model(*args, **kwargs): + import_calls.append((args, kwargs)) + return [] + + device_import._import_model_into_group = fake_import_model + + report = device_import.import_devices_from_payload( + { + "project_uuid": "project-1", + "devices": [ + { + "device_instance_id": "device-inst-1", + "display_tag": "J1", + "terminals": [ + { + "terminal_uuid": "terminal-shared", + "element_uuid": "element-a", + }, + { + "terminal_uuid": "terminal-shared", + "element_uuid": "element-b", + }, + ], + } + ], + "device_models": [ + { + "device_instance_id": "device-inst-1", + "resolved_model_path": str(model_path), + } + ], + } + ) + + self.assertEqual([], import_calls) + self.assertEqual(1, report["updated_devices"]) + self.assertEqual(["端子"], report["updated_device_details"][0]["change_types"]) + self.assertEqual(1, report["updated_device_details"][0]["previous_terminal_entry_count"]) + self.assertEqual(2, report["updated_device_details"][0]["current_terminal_entry_count"]) + def test_import_devices_from_payload_updates_existing_display_tag_even_when_model_path_missing(self): with tempfile.TemporaryDirectory() as temp_dir: model_path = Path(temp_dir) / "device.step" @@ -1168,7 +1258,9 @@ class FcstdDeviceImportTest(unittest.TestCase): self.assertEqual(str(model_path), device_group.QetResolvedModelPath) self.assertEqual(1, report["updated_devices"]) self.assertEqual(1, report["skipped_missing_model"]) - self.assertEqual(["标注"], report["updated_device_details"][0]["change_types"]) + self.assertEqual(["标注", "端子"], report["updated_device_details"][0]["change_types"]) + self.assertEqual(0, report["updated_device_details"][0]["previous_terminal_entry_count"]) + self.assertEqual(1, report["updated_device_details"][0]["current_terminal_entry_count"]) if __name__ == "__main__":