ngskcloud
/
LightWork3D
2
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

feature/优化2d-3d数据交换-0612

Browse Source
dev^2
zhanghao 2 weeks ago
parent e861c2408a
commit a9a8ef5b1c
4 changed files with 347 additions and 11 deletions
Show Stats Download Patch File Download Diff File

154
src/Mod/FreeCADExchange/DeviceImport.py
Unescape Escape View File

@ -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:

83
src/Mod/FreeCADExchange/ExchangeBootstrap.py
Unescape Escape View File

@ -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 "<empty>",
)
)
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 "<empty>",
)
)
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 "<empty>"
current_display_tag = item.get("display_tag", "") or "<empty>"

27
tests/python/freecad_exchange_bootstrap_wiring_test.py
Unescape Escape View File

@ -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)

94
tests/python/freecad_exchange_device_import_fcstd_test.py
Unescape Escape View File

@ -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__":

Write Preview
Loading…
Cancel
Save