feat: 支持导线任务导入与手动布线
parent
58c9eae33c
commit
44c2e448b5
@ -0,0 +1,208 @@
|
||||
# FreeCADExchange wire task import helpers.
|
||||
|
||||
import json
|
||||
|
||||
import FreeCAD as App
|
||||
|
||||
import TerminalObjects
|
||||
import WiringObjects
|
||||
|
||||
|
||||
class WiringImportError(RuntimeError):
|
||||
pass
|
||||
|
||||
|
||||
def _string_value(item, field_name):
|
||||
value = item.get(field_name, "")
|
||||
if value is None:
|
||||
return ""
|
||||
if not isinstance(value, str):
|
||||
return str(value).strip()
|
||||
return value.strip()
|
||||
|
||||
|
||||
def _bool_value(item, field_name):
|
||||
return bool(item.get(field_name, False))
|
||||
|
||||
|
||||
def _conductor_uuids(item):
|
||||
values = item.get("conductor_uuids", [])
|
||||
if not isinstance(values, list):
|
||||
return []
|
||||
return [str(value).strip() for value in values if str(value).strip()]
|
||||
|
||||
|
||||
def _normalize_wire_entry(item, index):
|
||||
if not isinstance(item, dict):
|
||||
raise WiringImportError("Wire entry #{0} must be an object.".format(index))
|
||||
|
||||
wire_uuid = (
|
||||
_string_value(item, "wire_id")
|
||||
or _string_value(item, "wire_uuid")
|
||||
or _string_value(item, "id")
|
||||
)
|
||||
start_terminal_uuid = _string_value(item, "start_terminal_uuid")
|
||||
end_terminal_uuid = _string_value(item, "end_terminal_uuid")
|
||||
|
||||
if not wire_uuid:
|
||||
raise WiringImportError("Wire entry #{0} is missing wire_id.".format(index))
|
||||
if not start_terminal_uuid or not end_terminal_uuid:
|
||||
raise WiringImportError(
|
||||
"Wire {0} is missing start_terminal_uuid or end_terminal_uuid.".format(
|
||||
wire_uuid
|
||||
)
|
||||
)
|
||||
|
||||
wire_mark = _string_value(item, "wire_mark")
|
||||
wire_label = _string_value(item, "wire_label") or wire_mark or wire_uuid
|
||||
return {
|
||||
"wire_uuid": wire_uuid,
|
||||
"wire_label": wire_label,
|
||||
"net_uuid": _string_value(item, "net_uuid"),
|
||||
"group_uuid": _string_value(item, "group_uuid"),
|
||||
"wire_mark": wire_mark,
|
||||
"wire_mark_is_manual": _bool_value(item, "wire_mark_is_manual"),
|
||||
"start_element_uuid": _string_value(item, "start_element_uuid"),
|
||||
"start_terminal_uuid": start_terminal_uuid,
|
||||
"start_instance_id": _string_value(item, "start_instance_id"),
|
||||
"start_terminal_display": _string_value(item, "start_terminal_display"),
|
||||
"end_element_uuid": _string_value(item, "end_element_uuid"),
|
||||
"end_terminal_uuid": end_terminal_uuid,
|
||||
"end_instance_id": _string_value(item, "end_instance_id"),
|
||||
"end_terminal_display": _string_value(item, "end_terminal_display"),
|
||||
"conductor_uuids": _conductor_uuids(item),
|
||||
}
|
||||
|
||||
|
||||
def _unique_object_name(doc, base_name):
|
||||
name = TerminalObjects.safe_token(base_name)
|
||||
if doc.getObject(name) is None:
|
||||
return name
|
||||
suffix = 1
|
||||
while doc.getObject("{0}_{1}".format(name, suffix)) is not None:
|
||||
suffix += 1
|
||||
return "{0}_{1}".format(name, suffix)
|
||||
|
||||
|
||||
def _task_label(entry):
|
||||
mark = entry["wire_mark"] or entry["wire_label"] or entry["wire_uuid"]
|
||||
start = entry["start_terminal_display"] or entry["start_terminal_uuid"]
|
||||
end = entry["end_terminal_display"] or entry["end_terminal_uuid"]
|
||||
return "{0} {1} -> {2}".format(mark, start, end)
|
||||
|
||||
|
||||
def _find_task_by_wire_uuid(task_group, wire_uuid):
|
||||
target = (wire_uuid or "").strip()
|
||||
if not target:
|
||||
return None
|
||||
for candidate in list(getattr(task_group, "Group", []) or []):
|
||||
if getattr(candidate, "QetWireUuid", "").strip() == target:
|
||||
return candidate
|
||||
return None
|
||||
|
||||
|
||||
def _ensure_string_property(obj, prop_name, value, description="QET wire task property"):
|
||||
TerminalObjects.ensure_string_property(
|
||||
obj,
|
||||
prop_name,
|
||||
"QET Wiring",
|
||||
description,
|
||||
value,
|
||||
)
|
||||
|
||||
|
||||
def _set_task_extra_properties(task, entry):
|
||||
_ensure_string_property(task, "QetStartElementUuid", entry["start_element_uuid"])
|
||||
_ensure_string_property(task, "QetEndElementUuid", entry["end_element_uuid"])
|
||||
_ensure_string_property(task, "QetStartTerminalDisplay", entry["start_terminal_display"])
|
||||
_ensure_string_property(task, "QetEndTerminalDisplay", entry["end_terminal_display"])
|
||||
_ensure_string_property(
|
||||
task,
|
||||
"QetConductorUuidsJson",
|
||||
json.dumps(entry["conductor_uuids"], ensure_ascii=False),
|
||||
)
|
||||
|
||||
|
||||
def _upsert_wire_task(doc, task_group, project_uuid, entry):
|
||||
task = _find_task_by_wire_uuid(task_group, entry["wire_uuid"])
|
||||
created = task is None
|
||||
previous_status = ""
|
||||
if task is None:
|
||||
task = doc.addObject(
|
||||
"App::DocumentObjectGroup",
|
||||
_unique_object_name(doc, "QETWireTask_{0}".format(entry["wire_uuid"])),
|
||||
)
|
||||
task_group.addObject(task)
|
||||
else:
|
||||
previous_status = (getattr(task, "RouteStatus", "") or "").strip()
|
||||
|
||||
WiringObjects.set_wire_task_semantics(
|
||||
task,
|
||||
project_uuid,
|
||||
entry["wire_uuid"],
|
||||
entry["wire_label"],
|
||||
entry["start_terminal_uuid"],
|
||||
entry["end_terminal_uuid"],
|
||||
entry["start_instance_id"],
|
||||
entry["end_instance_id"],
|
||||
net_uuid=entry["net_uuid"],
|
||||
group_uuid=entry["group_uuid"],
|
||||
wire_mark=entry["wire_mark"],
|
||||
wire_mark_is_manual=entry["wire_mark_is_manual"],
|
||||
)
|
||||
_set_task_extra_properties(task, entry)
|
||||
if previous_status and previous_status != "Task":
|
||||
TerminalObjects.ensure_string_property(
|
||||
task,
|
||||
"RouteStatus",
|
||||
"QET Wiring",
|
||||
"Wire task route status",
|
||||
previous_status,
|
||||
)
|
||||
task.Label = _task_label(entry)
|
||||
return task, created
|
||||
|
||||
|
||||
def import_wire_tasks_from_payload(payload, doc=None):
|
||||
if doc is None:
|
||||
doc = getattr(App, "ActiveDocument", None)
|
||||
if doc is None:
|
||||
raise WiringImportError("No active FreeCAD document is available.")
|
||||
if not isinstance(payload, dict):
|
||||
raise WiringImportError("Exchange payload must be an object.")
|
||||
|
||||
project_uuid = _string_value(payload, "project_uuid")
|
||||
if not project_uuid:
|
||||
raise WiringImportError("Field 'project_uuid' is required for wire task import.")
|
||||
|
||||
wires = payload.get("wires", [])
|
||||
if wires is None:
|
||||
wires = []
|
||||
if not isinstance(wires, list):
|
||||
raise WiringImportError("Field 'wires' must be a list.")
|
||||
|
||||
task_group = WiringObjects.ensure_task_group(doc, project_uuid)
|
||||
report = {
|
||||
"project_uuid": project_uuid,
|
||||
"total_wires": len(wires),
|
||||
"imported_tasks": 0,
|
||||
"updated_tasks": 0,
|
||||
"skipped_invalid": 0,
|
||||
"warnings": [],
|
||||
}
|
||||
|
||||
for index, item in enumerate(wires):
|
||||
try:
|
||||
entry = _normalize_wire_entry(item, index)
|
||||
except WiringImportError as exc:
|
||||
report["skipped_invalid"] += 1
|
||||
report["warnings"].append(str(exc))
|
||||
continue
|
||||
|
||||
_task, created = _upsert_wire_task(doc, task_group, project_uuid, entry)
|
||||
if created:
|
||||
report["imported_tasks"] += 1
|
||||
else:
|
||||
report["updated_tasks"] += 1
|
||||
|
||||
return report
|
||||
@ -0,0 +1,159 @@
|
||||
import importlib
|
||||
import sys
|
||||
import types
|
||||
import unittest
|
||||
from pathlib import Path
|
||||
|
||||
|
||||
REPO_ROOT = Path(__file__).resolve().parents[2]
|
||||
MODULE_DIR = REPO_ROOT / "src" / "Mod" / "FreeCADExchange"
|
||||
if str(MODULE_DIR) not in sys.path:
|
||||
sys.path.insert(0, str(MODULE_DIR))
|
||||
|
||||
|
||||
def _install_fake_freecad():
|
||||
class Vector:
|
||||
def __init__(self, x=0.0, y=0.0, z=0.0):
|
||||
self.x = float(x)
|
||||
self.y = float(y)
|
||||
self.z = float(z)
|
||||
|
||||
class Rotation:
|
||||
pass
|
||||
|
||||
class Placement:
|
||||
def __init__(self, base=None, rotation=None):
|
||||
self.Base = base or Vector()
|
||||
self.Rotation = rotation or Rotation()
|
||||
|
||||
fake_freecad = types.ModuleType("FreeCAD")
|
||||
fake_freecad.Vector = Vector
|
||||
fake_freecad.Rotation = Rotation
|
||||
fake_freecad.Placement = Placement
|
||||
fake_freecad.ActiveDocument = None
|
||||
fake_freecad.Console = types.SimpleNamespace(
|
||||
PrintMessage=lambda *args, **kwargs: None,
|
||||
PrintWarning=lambda *args, **kwargs: None,
|
||||
PrintError=lambda *args, **kwargs: None,
|
||||
)
|
||||
sys.modules["FreeCAD"] = fake_freecad
|
||||
|
||||
|
||||
class FakeViewObject:
|
||||
def __init__(self):
|
||||
self.Visibility = True
|
||||
|
||||
|
||||
class FakeObject:
|
||||
def __init__(self, name, type_id):
|
||||
self.Name = name
|
||||
self.Label = name
|
||||
self.TypeId = type_id
|
||||
self.PropertiesList = []
|
||||
self.Group = []
|
||||
self.ViewObject = FakeViewObject()
|
||||
self.InList = []
|
||||
|
||||
def isDerivedFrom(self, type_name):
|
||||
return self.TypeId == type_name
|
||||
|
||||
def addProperty(self, prop_type, prop_name, group_name, description):
|
||||
if prop_name not in self.PropertiesList:
|
||||
self.PropertiesList.append(prop_name)
|
||||
|
||||
def addObject(self, child):
|
||||
if child not in self.Group:
|
||||
self.Group.append(child)
|
||||
if self not in child.InList:
|
||||
child.InList.append(self)
|
||||
|
||||
|
||||
class FakeDocument:
|
||||
def __init__(self):
|
||||
self.Objects = []
|
||||
self.Name = "FakeDoc"
|
||||
|
||||
def addObject(self, type_name, name):
|
||||
obj = FakeObject(name, type_name)
|
||||
self.Objects.append(obj)
|
||||
return obj
|
||||
|
||||
def getObject(self, name):
|
||||
for obj in self.Objects:
|
||||
if obj.Name == name:
|
||||
return obj
|
||||
return None
|
||||
|
||||
|
||||
def _reload_modules():
|
||||
for name in ["TerminalObjects", "WiringObjects", "WiringImport"]:
|
||||
sys.modules.pop(name, None)
|
||||
terminal_objects = importlib.import_module("TerminalObjects")
|
||||
wiring_objects = importlib.import_module("WiringObjects")
|
||||
wiring_import = importlib.import_module("WiringImport")
|
||||
return terminal_objects, wiring_objects, wiring_import
|
||||
|
||||
|
||||
class WiringImportTest(unittest.TestCase):
|
||||
def test_import_wire_tasks_creates_and_updates_qet_tasks(self):
|
||||
_install_fake_freecad()
|
||||
terminal_objects, _wiring_objects, wiring_import = _reload_modules()
|
||||
|
||||
doc = FakeDocument()
|
||||
terminal_objects.ensure_root_group(doc, "project-1")
|
||||
payload = {
|
||||
"project_uuid": "project-1",
|
||||
"wires": [
|
||||
{
|
||||
"wire_id": "wire-1",
|
||||
"net_uuid": "net-1",
|
||||
"group_uuid": "group-1",
|
||||
"wire_mark": "W001",
|
||||
"wire_mark_is_manual": True,
|
||||
"start_element_uuid": "device-a",
|
||||
"start_terminal_uuid": "terminal-a",
|
||||
"end_element_uuid": "device-b",
|
||||
"end_terminal_uuid": "terminal-b",
|
||||
"start_terminal_display": "A1",
|
||||
"end_terminal_display": "B1",
|
||||
"conductor_uuids": ["conductor-1"],
|
||||
}
|
||||
],
|
||||
}
|
||||
|
||||
report = wiring_import.import_wire_tasks_from_payload(payload, doc)
|
||||
|
||||
task_group = doc.getObject("QETWiring_01_Tasks")
|
||||
self.assertIsNotNone(task_group)
|
||||
self.assertEqual(1, report["imported_tasks"])
|
||||
self.assertEqual(0, report["updated_tasks"])
|
||||
self.assertEqual(1, len(task_group.Group))
|
||||
task = task_group.Group[0]
|
||||
self.assertEqual("wire-1", task.QetWireUuid)
|
||||
self.assertEqual("net-1", task.QetNetUuid)
|
||||
self.assertEqual("group-1", task.QetGroupUuid)
|
||||
self.assertEqual("W001", task.QetWireMark)
|
||||
self.assertTrue(task.QetWireMarkIsManual)
|
||||
self.assertEqual("terminal-a", task.QetStartTerminalUuid)
|
||||
self.assertEqual("terminal-b", task.QetEndTerminalUuid)
|
||||
self.assertEqual("device-a", task.QetStartElementUuid)
|
||||
self.assertEqual("device-b", task.QetEndElementUuid)
|
||||
self.assertEqual("A1", task.QetStartTerminalDisplay)
|
||||
self.assertEqual("B1", task.QetEndTerminalDisplay)
|
||||
self.assertIn("conductor-1", task.QetConductorUuidsJson)
|
||||
self.assertEqual("Task", task.RouteType)
|
||||
self.assertEqual("Task", task.RouteStatus)
|
||||
|
||||
payload["wires"][0]["wire_mark"] = "W001-updated"
|
||||
task.RouteStatus = "Routed"
|
||||
second_report = wiring_import.import_wire_tasks_from_payload(payload, doc)
|
||||
|
||||
self.assertEqual(0, second_report["imported_tasks"])
|
||||
self.assertEqual(1, second_report["updated_tasks"])
|
||||
self.assertEqual(1, len(task_group.Group))
|
||||
self.assertEqual("W001-updated", task_group.Group[0].QetWireMark)
|
||||
self.assertEqual("Routed", task_group.Group[0].RouteStatus)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
Loading…
Reference in New Issue