feature/端子显示连线保存回写-zwl-0520
parent
79d39fcf2a
commit
73ce289d98
@ -0,0 +1,323 @@
|
|||||||
|
# FreeCADExchange write-back helpers.
|
||||||
|
|
||||||
|
import json
|
||||||
|
import os
|
||||||
|
from datetime import datetime
|
||||||
|
from pathlib import Path
|
||||||
|
|
||||||
|
import FreeCAD as App
|
||||||
|
|
||||||
|
import DeviceImport
|
||||||
|
import TerminalObjects as TerminalObjects
|
||||||
|
|
||||||
|
try:
|
||||||
|
import FreeCADGui as Gui
|
||||||
|
except ImportError:
|
||||||
|
Gui = None
|
||||||
|
|
||||||
|
|
||||||
|
STATE_WRITEBACK_OBSERVER = "_qet_exchange_writeback_observer"
|
||||||
|
|
||||||
|
|
||||||
|
class ExchangeWriteBackError(RuntimeError):
|
||||||
|
pass
|
||||||
|
|
||||||
|
|
||||||
|
def _append_debug_log(message):
|
||||||
|
try:
|
||||||
|
DeviceImport._append_debug_log(message)
|
||||||
|
except Exception:
|
||||||
|
pass
|
||||||
|
|
||||||
|
|
||||||
|
def _project_uuid_from_payload(payload):
|
||||||
|
if isinstance(payload, dict):
|
||||||
|
value = (payload.get("project_uuid") or "").strip()
|
||||||
|
if value:
|
||||||
|
return value
|
||||||
|
return ""
|
||||||
|
|
||||||
|
|
||||||
|
def _root_group(doc):
|
||||||
|
try:
|
||||||
|
return doc.getObject(TerminalObjects.ROOT_GROUP_NAME)
|
||||||
|
except Exception:
|
||||||
|
return None
|
||||||
|
|
||||||
|
|
||||||
|
def _is_device_group(obj):
|
||||||
|
if obj is None:
|
||||||
|
return False
|
||||||
|
try:
|
||||||
|
if not obj.Name.startswith(DeviceImport.DEVICE_GROUP_PREFIX):
|
||||||
|
return False
|
||||||
|
return "QetElementUuid" in getattr(obj, "PropertiesList", [])
|
||||||
|
except Exception:
|
||||||
|
return False
|
||||||
|
|
||||||
|
|
||||||
|
def _iter_device_groups(doc):
|
||||||
|
root = _root_group(doc)
|
||||||
|
if root is not None:
|
||||||
|
for child in list(getattr(root, "Group", []) or []):
|
||||||
|
if _is_device_group(child):
|
||||||
|
yield child
|
||||||
|
return
|
||||||
|
|
||||||
|
for obj in doc.Objects:
|
||||||
|
if _is_device_group(obj):
|
||||||
|
yield obj
|
||||||
|
|
||||||
|
|
||||||
|
def _iter_terminal_objects(device_group):
|
||||||
|
terminal_container = TerminalObjects.find_child_group_by_kind(
|
||||||
|
device_group,
|
||||||
|
TerminalObjects.TERMINAL_GROUP_KIND,
|
||||||
|
)
|
||||||
|
if terminal_container is None:
|
||||||
|
return []
|
||||||
|
return TerminalObjects.collect_terminal_objects(terminal_container)
|
||||||
|
|
||||||
|
|
||||||
|
def _scene_path_from_doc(doc, scene_path=""):
|
||||||
|
candidate = (scene_path or "").strip()
|
||||||
|
if candidate:
|
||||||
|
return candidate
|
||||||
|
|
||||||
|
env_scene = os.environ.get("QET_FREECAD_SCENE_FILE", "").strip()
|
||||||
|
if env_scene:
|
||||||
|
return env_scene
|
||||||
|
|
||||||
|
file_name = getattr(doc, "FileName", "").strip()
|
||||||
|
if file_name:
|
||||||
|
return file_name
|
||||||
|
|
||||||
|
return ""
|
||||||
|
|
||||||
|
|
||||||
|
def _output_path_for_scene(scene_path):
|
||||||
|
scene_path = (scene_path or "").strip()
|
||||||
|
if not scene_path:
|
||||||
|
return ""
|
||||||
|
|
||||||
|
path = Path(scene_path)
|
||||||
|
if path.suffix.lower() == ".fcstd":
|
||||||
|
return str(path.with_name("3d_to_2d.json"))
|
||||||
|
if path.is_dir():
|
||||||
|
return str(path / "3d_to_2d.json")
|
||||||
|
if path.name.lower().endswith(".fcstd"):
|
||||||
|
return str(path.with_name("3d_to_2d.json"))
|
||||||
|
return str(path.parent / "3d_to_2d.json")
|
||||||
|
|
||||||
|
|
||||||
|
def _format_timestamp():
|
||||||
|
return datetime.now().astimezone().isoformat(timespec="seconds")
|
||||||
|
|
||||||
|
|
||||||
|
def _collect_instance_bindings(doc):
|
||||||
|
bindings = []
|
||||||
|
seen = set()
|
||||||
|
for device_group in _iter_device_groups(doc):
|
||||||
|
element_uuid = getattr(device_group, "QetElementUuid", "").strip()
|
||||||
|
instance_id = getattr(device_group, "QetInstanceId", "").strip()
|
||||||
|
if not element_uuid or not instance_id:
|
||||||
|
continue
|
||||||
|
key = (element_uuid, instance_id)
|
||||||
|
if key in seen:
|
||||||
|
continue
|
||||||
|
seen.add(key)
|
||||||
|
bindings.append(
|
||||||
|
{
|
||||||
|
"element_uuid": element_uuid,
|
||||||
|
"instance_id": instance_id,
|
||||||
|
}
|
||||||
|
)
|
||||||
|
return bindings
|
||||||
|
|
||||||
|
|
||||||
|
def _collect_terminal_bindings(doc):
|
||||||
|
bindings = []
|
||||||
|
seen = set()
|
||||||
|
for device_group in _iter_device_groups(doc):
|
||||||
|
instance_id = getattr(device_group, "QetInstanceId", "").strip()
|
||||||
|
for terminal_obj in _iter_terminal_objects(device_group):
|
||||||
|
terminal_uuid = getattr(terminal_obj, "QetTerminalUuid", "").strip()
|
||||||
|
terminal_instance_id = getattr(terminal_obj, "QetInstanceId", "").strip() or instance_id
|
||||||
|
if not terminal_uuid or not terminal_instance_id:
|
||||||
|
continue
|
||||||
|
key = (terminal_uuid, terminal_instance_id)
|
||||||
|
if key in seen:
|
||||||
|
continue
|
||||||
|
seen.add(key)
|
||||||
|
bindings.append(
|
||||||
|
{
|
||||||
|
"terminal_uuid": terminal_uuid,
|
||||||
|
"instance_id": terminal_instance_id,
|
||||||
|
}
|
||||||
|
)
|
||||||
|
return bindings
|
||||||
|
|
||||||
|
|
||||||
|
def _project_uuid_from_doc(doc, payload=None):
|
||||||
|
root = _root_group(doc)
|
||||||
|
if root is not None:
|
||||||
|
project_uuid = getattr(root, "QetProjectUuid", "").strip()
|
||||||
|
if project_uuid:
|
||||||
|
return project_uuid
|
||||||
|
return _project_uuid_from_payload(payload)
|
||||||
|
|
||||||
|
|
||||||
|
def write_back_document(doc=None, scene_path="", payload=None):
|
||||||
|
if doc is None:
|
||||||
|
doc = App.ActiveDocument
|
||||||
|
if doc is None:
|
||||||
|
raise ExchangeWriteBackError("No active FreeCAD document is available.")
|
||||||
|
|
||||||
|
scene_path = _scene_path_from_doc(doc, scene_path)
|
||||||
|
output_path = _output_path_for_scene(scene_path)
|
||||||
|
if not output_path:
|
||||||
|
raise ExchangeWriteBackError(
|
||||||
|
"Cannot determine the 3d_to_2d.json output path."
|
||||||
|
)
|
||||||
|
|
||||||
|
project_uuid = _project_uuid_from_doc(doc, payload)
|
||||||
|
if not project_uuid:
|
||||||
|
raise ExchangeWriteBackError(
|
||||||
|
"Cannot determine project_uuid for write-back."
|
||||||
|
)
|
||||||
|
|
||||||
|
report = {
|
||||||
|
"schema_version": "1.0",
|
||||||
|
"project_uuid": project_uuid,
|
||||||
|
"generated_at": _format_timestamp(),
|
||||||
|
"instances": _collect_instance_bindings(doc),
|
||||||
|
"terminals": _collect_terminal_bindings(doc),
|
||||||
|
"output_path": output_path,
|
||||||
|
}
|
||||||
|
|
||||||
|
output_dir = str(Path(output_path).parent)
|
||||||
|
os.makedirs(output_dir, exist_ok=True)
|
||||||
|
Path(output_path).write_text(
|
||||||
|
json.dumps(
|
||||||
|
{
|
||||||
|
"schema_version": report["schema_version"],
|
||||||
|
"project_uuid": report["project_uuid"],
|
||||||
|
"generated_at": report["generated_at"],
|
||||||
|
"instances": report["instances"],
|
||||||
|
"terminals": report["terminals"],
|
||||||
|
},
|
||||||
|
ensure_ascii=False,
|
||||||
|
indent=2,
|
||||||
|
),
|
||||||
|
encoding="utf-8",
|
||||||
|
)
|
||||||
|
|
||||||
|
_append_debug_log(
|
||||||
|
"write_back_document completed: instances={0}, terminals={1}, path={2}".format(
|
||||||
|
len(report["instances"]),
|
||||||
|
len(report["terminals"]),
|
||||||
|
output_path,
|
||||||
|
)
|
||||||
|
)
|
||||||
|
try:
|
||||||
|
App.Console.PrintMessage(
|
||||||
|
"[FreeCADExchange] Wrote 3d_to_2d.json to {0}\n".format(output_path)
|
||||||
|
)
|
||||||
|
except Exception:
|
||||||
|
pass
|
||||||
|
return report
|
||||||
|
|
||||||
|
|
||||||
|
def _is_exchange_document(doc):
|
||||||
|
if doc is None:
|
||||||
|
return False
|
||||||
|
|
||||||
|
if _root_group(doc) is not None:
|
||||||
|
return True
|
||||||
|
|
||||||
|
for obj in doc.Objects:
|
||||||
|
if _is_device_group(obj):
|
||||||
|
return True
|
||||||
|
return False
|
||||||
|
|
||||||
|
|
||||||
|
class _WriteBackObserver:
|
||||||
|
def slotFinishSaveDocument(self, doc, name):
|
||||||
|
if not _is_exchange_document(doc):
|
||||||
|
return
|
||||||
|
try:
|
||||||
|
write_back_document(doc, scene_path=name)
|
||||||
|
except Exception as exc:
|
||||||
|
_append_debug_log("write-back after save failed: {0}".format(exc))
|
||||||
|
try:
|
||||||
|
App.Console.PrintError(
|
||||||
|
"[FreeCADExchange] write-back after save failed: {0}\n".format(exc)
|
||||||
|
)
|
||||||
|
except Exception:
|
||||||
|
pass
|
||||||
|
|
||||||
|
|
||||||
|
def ensure_document_observer_installed():
|
||||||
|
if getattr(App, STATE_WRITEBACK_OBSERVER, None) is not None:
|
||||||
|
return getattr(App, STATE_WRITEBACK_OBSERVER)
|
||||||
|
|
||||||
|
observer = _WriteBackObserver()
|
||||||
|
try:
|
||||||
|
App.addDocumentObserver(observer)
|
||||||
|
except Exception as exc:
|
||||||
|
_append_debug_log("failed to add write-back observer: {0}".format(exc))
|
||||||
|
return None
|
||||||
|
|
||||||
|
setattr(App, STATE_WRITEBACK_OBSERVER, observer)
|
||||||
|
return observer
|
||||||
|
|
||||||
|
|
||||||
|
class CommandWriteBack:
|
||||||
|
def GetResources(self):
|
||||||
|
return {
|
||||||
|
"MenuText": "Write Back 3D Binding",
|
||||||
|
"ToolTip": "Generate 3d_to_2d.json from the current FreeCAD document",
|
||||||
|
}
|
||||||
|
|
||||||
|
def IsActive(self):
|
||||||
|
return App.ActiveDocument is not None
|
||||||
|
|
||||||
|
def Activated(self):
|
||||||
|
try:
|
||||||
|
report = write_back_document(App.ActiveDocument)
|
||||||
|
try:
|
||||||
|
App.Console.PrintMessage(
|
||||||
|
"[FreeCADExchange] Write-back completed: {0} instances, {1} terminals\n".format(
|
||||||
|
len(report["instances"]),
|
||||||
|
len(report["terminals"]),
|
||||||
|
)
|
||||||
|
)
|
||||||
|
except Exception:
|
||||||
|
pass
|
||||||
|
except Exception as exc:
|
||||||
|
try:
|
||||||
|
App.Console.PrintError(
|
||||||
|
"[FreeCADExchange] Write-back failed: {0}\n".format(exc)
|
||||||
|
)
|
||||||
|
except Exception:
|
||||||
|
pass
|
||||||
|
|
||||||
|
|
||||||
|
_COMMANDS_REGISTERED = False
|
||||||
|
|
||||||
|
|
||||||
|
def register_commands():
|
||||||
|
global _COMMANDS_REGISTERED
|
||||||
|
if _COMMANDS_REGISTERED:
|
||||||
|
return
|
||||||
|
if Gui is None:
|
||||||
|
return
|
||||||
|
try:
|
||||||
|
Gui.addCommand("QET_Exchange_WriteBack", CommandWriteBack())
|
||||||
|
_COMMANDS_REGISTERED = True
|
||||||
|
except Exception as exc:
|
||||||
|
_append_debug_log("failed to register write-back command: {0}".format(exc))
|
||||||
|
|
||||||
|
|
||||||
|
register_commands()
|
||||||
|
ensure_document_observer_installed()
|
||||||
@ -0,0 +1,236 @@
|
|||||||
|
# FreeCADExchange manual wiring helpers.
|
||||||
|
|
||||||
|
import FreeCAD as App
|
||||||
|
|
||||||
|
try:
|
||||||
|
import FreeCADGui as Gui
|
||||||
|
except ImportError:
|
||||||
|
Gui = None
|
||||||
|
|
||||||
|
import DeviceImport
|
||||||
|
import TerminalObjects as TerminalObjects
|
||||||
|
|
||||||
|
|
||||||
|
class ManualWiringError(RuntimeError):
|
||||||
|
pass
|
||||||
|
|
||||||
|
|
||||||
|
def _append_debug_log(message):
|
||||||
|
try:
|
||||||
|
DeviceImport._append_debug_log(message)
|
||||||
|
except Exception:
|
||||||
|
pass
|
||||||
|
|
||||||
|
|
||||||
|
def _terminal_points(start_terminal, end_terminal, waypoints=None):
|
||||||
|
points = [TerminalObjects.terminal_origin(start_terminal)]
|
||||||
|
for point in waypoints or []:
|
||||||
|
if isinstance(point, App.Vector):
|
||||||
|
points.append(point)
|
||||||
|
elif isinstance(point, (list, tuple)) and len(point) >= 3:
|
||||||
|
points.append(App.Vector(float(point[0]), float(point[1]), float(point[2])))
|
||||||
|
points.append(TerminalObjects.terminal_origin(end_terminal))
|
||||||
|
return points
|
||||||
|
|
||||||
|
|
||||||
|
def _wire_object_name(start_terminal, end_terminal):
|
||||||
|
start_uuid = TerminalObjects.safe_token(getattr(start_terminal, "QetTerminalUuid", ""))
|
||||||
|
end_uuid = TerminalObjects.safe_token(getattr(end_terminal, "QetTerminalUuid", ""))
|
||||||
|
return "QETWire_{0}_{1}".format(start_uuid, end_uuid)
|
||||||
|
|
||||||
|
|
||||||
|
def _set_wire_properties(obj, project_uuid, start_terminal, end_terminal):
|
||||||
|
TerminalObjects.ensure_string_property(
|
||||||
|
obj,
|
||||||
|
"QetProjectUuid",
|
||||||
|
"QET Exchange",
|
||||||
|
"Project UUID for this wire",
|
||||||
|
project_uuid,
|
||||||
|
)
|
||||||
|
TerminalObjects.ensure_string_property(
|
||||||
|
obj,
|
||||||
|
"QetStartTerminalUuid",
|
||||||
|
"QET Exchange",
|
||||||
|
"Start terminal UUID",
|
||||||
|
getattr(start_terminal, "QetTerminalUuid", "").strip(),
|
||||||
|
)
|
||||||
|
TerminalObjects.ensure_string_property(
|
||||||
|
obj,
|
||||||
|
"QetEndTerminalUuid",
|
||||||
|
"QET Exchange",
|
||||||
|
"End terminal UUID",
|
||||||
|
getattr(end_terminal, "QetTerminalUuid", "").strip(),
|
||||||
|
)
|
||||||
|
TerminalObjects.ensure_string_property(
|
||||||
|
obj,
|
||||||
|
"QetStartInstanceId",
|
||||||
|
"QET Exchange",
|
||||||
|
"Start device instance ID",
|
||||||
|
getattr(start_terminal, "QetInstanceId", "").strip(),
|
||||||
|
)
|
||||||
|
TerminalObjects.ensure_string_property(
|
||||||
|
obj,
|
||||||
|
"QetEndInstanceId",
|
||||||
|
"QET Exchange",
|
||||||
|
"End device instance ID",
|
||||||
|
getattr(end_terminal, "QetInstanceId", "").strip(),
|
||||||
|
)
|
||||||
|
TerminalObjects.ensure_string_property(
|
||||||
|
obj,
|
||||||
|
"RouteType",
|
||||||
|
"QET Exchange",
|
||||||
|
"Wire route type",
|
||||||
|
"Manual",
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
|
def _wire_parent_group(doc, project_uuid, start_terminal, end_terminal, fallback_group=None):
|
||||||
|
for terminal in (start_terminal, end_terminal):
|
||||||
|
element_uuid = getattr(terminal, "QetElementUuid", "").strip()
|
||||||
|
if not element_uuid:
|
||||||
|
continue
|
||||||
|
|
||||||
|
device_group = TerminalObjects.find_device_group(doc, element_uuid)
|
||||||
|
if device_group is None:
|
||||||
|
continue
|
||||||
|
|
||||||
|
device_instance_id = getattr(device_group, "QetInstanceId", "").strip()
|
||||||
|
return TerminalObjects.ensure_wire_group(
|
||||||
|
doc,
|
||||||
|
device_group,
|
||||||
|
project_uuid=project_uuid,
|
||||||
|
instance_id=device_instance_id,
|
||||||
|
)
|
||||||
|
|
||||||
|
return fallback_group
|
||||||
|
|
||||||
|
|
||||||
|
def create_manual_wire(doc, start_terminal, end_terminal, waypoints=None, parent_group=None):
|
||||||
|
if not TerminalObjects.is_terminal_object(start_terminal):
|
||||||
|
raise ManualWiringError("The start selection is not a valid terminal.")
|
||||||
|
if not TerminalObjects.is_terminal_object(end_terminal):
|
||||||
|
raise ManualWiringError("The end selection is not a valid terminal.")
|
||||||
|
if start_terminal == end_terminal:
|
||||||
|
raise ManualWiringError("The start and end terminal must be different.")
|
||||||
|
|
||||||
|
project_uuid = (
|
||||||
|
getattr(start_terminal, "QetProjectUuid", "").strip()
|
||||||
|
or getattr(end_terminal, "QetProjectUuid", "").strip()
|
||||||
|
or getattr(DeviceImport._ensure_root_group(doc), "QetProjectUuid", "").strip()
|
||||||
|
)
|
||||||
|
if not project_uuid:
|
||||||
|
raise ManualWiringError("A project UUID is required to create a wire.")
|
||||||
|
|
||||||
|
wire_base_name = TerminalObjects.safe_token(
|
||||||
|
_wire_object_name(start_terminal, end_terminal)
|
||||||
|
)
|
||||||
|
wire_name = wire_base_name
|
||||||
|
suffix = 1
|
||||||
|
while doc.getObject(wire_name) is not None:
|
||||||
|
wire_name = "{0}_{1}".format(wire_base_name, suffix)
|
||||||
|
suffix += 1
|
||||||
|
|
||||||
|
wire_obj = doc.addObject("Part::Feature", wire_name)
|
||||||
|
wire_obj.Label = "QET Manual Wire"
|
||||||
|
|
||||||
|
points = _terminal_points(start_terminal, end_terminal, waypoints=waypoints)
|
||||||
|
if len(points) < 2:
|
||||||
|
raise ManualWiringError("A wire requires at least two points.")
|
||||||
|
|
||||||
|
import Part
|
||||||
|
|
||||||
|
wire_obj.Shape = Part.makePolygon(points)
|
||||||
|
_set_wire_properties(
|
||||||
|
wire_obj,
|
||||||
|
project_uuid,
|
||||||
|
start_terminal,
|
||||||
|
end_terminal,
|
||||||
|
)
|
||||||
|
|
||||||
|
if parent_group is None:
|
||||||
|
try:
|
||||||
|
parent_group = _wire_parent_group(
|
||||||
|
doc,
|
||||||
|
project_uuid,
|
||||||
|
start_terminal,
|
||||||
|
end_terminal,
|
||||||
|
fallback_group=DeviceImport._ensure_root_group(doc, project_uuid),
|
||||||
|
)
|
||||||
|
except Exception:
|
||||||
|
parent_group = None
|
||||||
|
if parent_group is not None and wire_obj not in getattr(parent_group, "Group", []):
|
||||||
|
parent_group.addObject(wire_obj)
|
||||||
|
|
||||||
|
try:
|
||||||
|
wire_obj.ViewObject.LineWidth = 2.0
|
||||||
|
except Exception:
|
||||||
|
pass
|
||||||
|
try:
|
||||||
|
wire_obj.ViewObject.LineColor = (0.0, 0.6, 1.0)
|
||||||
|
except Exception:
|
||||||
|
pass
|
||||||
|
|
||||||
|
doc.recompute()
|
||||||
|
return wire_obj
|
||||||
|
|
||||||
|
|
||||||
|
class CommandCreateManualWire:
|
||||||
|
def GetResources(self):
|
||||||
|
return {
|
||||||
|
"MenuText": "Create Manual Wire",
|
||||||
|
"ToolTip": "Create a manual wire between two selected terminals",
|
||||||
|
}
|
||||||
|
|
||||||
|
def IsActive(self):
|
||||||
|
return App.ActiveDocument is not None and Gui is not None
|
||||||
|
|
||||||
|
def Activated(self):
|
||||||
|
if Gui is None:
|
||||||
|
return
|
||||||
|
|
||||||
|
selection = [
|
||||||
|
obj
|
||||||
|
for obj in Gui.Selection.getSelection()
|
||||||
|
if TerminalObjects.is_terminal_object(obj)
|
||||||
|
]
|
||||||
|
if len(selection) != 2:
|
||||||
|
try:
|
||||||
|
App.Console.PrintWarning(
|
||||||
|
"Select exactly two valid terminals before creating a wire.\n"
|
||||||
|
)
|
||||||
|
except Exception:
|
||||||
|
pass
|
||||||
|
return
|
||||||
|
|
||||||
|
try:
|
||||||
|
create_manual_wire(App.ActiveDocument, selection[0], selection[1])
|
||||||
|
try:
|
||||||
|
Gui.SendMsgToActiveView("ViewFit")
|
||||||
|
except Exception:
|
||||||
|
pass
|
||||||
|
except Exception as exc:
|
||||||
|
try:
|
||||||
|
App.Console.PrintError(
|
||||||
|
"[FreeCADExchange] manual wire creation failed: {0}\n".format(exc)
|
||||||
|
)
|
||||||
|
except Exception:
|
||||||
|
pass
|
||||||
|
|
||||||
|
|
||||||
|
_COMMANDS_REGISTERED = False
|
||||||
|
|
||||||
|
|
||||||
|
def register_commands():
|
||||||
|
global _COMMANDS_REGISTERED
|
||||||
|
if _COMMANDS_REGISTERED:
|
||||||
|
return
|
||||||
|
if Gui is None:
|
||||||
|
return
|
||||||
|
try:
|
||||||
|
Gui.addCommand("QET_Exchange_CreateManualWire", CommandCreateManualWire())
|
||||||
|
_COMMANDS_REGISTERED = True
|
||||||
|
except Exception as exc:
|
||||||
|
_append_debug_log("failed to register manual wiring command: {0}".format(exc))
|
||||||
|
|
||||||
|
|
||||||
|
register_commands()
|
||||||
@ -0,0 +1,363 @@
|
|||||||
|
# FreeCADExchange template semantics helpers.
|
||||||
|
|
||||||
|
import json
|
||||||
|
from pathlib import Path
|
||||||
|
|
||||||
|
import FreeCAD as App
|
||||||
|
|
||||||
|
import TerminalObjects as TerminalObjects
|
||||||
|
|
||||||
|
|
||||||
|
def _sidecar_candidates(model_path):
|
||||||
|
native = TerminalObjects.native_path(model_path)
|
||||||
|
if not native:
|
||||||
|
return []
|
||||||
|
|
||||||
|
path = Path(native)
|
||||||
|
base = path.stem
|
||||||
|
parent = path.parent
|
||||||
|
return [
|
||||||
|
parent / (base + ".terminals.json"),
|
||||||
|
parent / (base + ".qet_terminals.json"),
|
||||||
|
parent / (base + ".terminal_slots.json"),
|
||||||
|
parent / (base + ".qet_template.json"),
|
||||||
|
]
|
||||||
|
|
||||||
|
|
||||||
|
def _load_json_file(path):
|
||||||
|
try:
|
||||||
|
raw_text = path.read_text(encoding="utf-8")
|
||||||
|
except OSError:
|
||||||
|
return None
|
||||||
|
|
||||||
|
try:
|
||||||
|
payload = json.loads(raw_text)
|
||||||
|
except json.JSONDecodeError:
|
||||||
|
return None
|
||||||
|
|
||||||
|
if not isinstance(payload, dict):
|
||||||
|
return None
|
||||||
|
return payload
|
||||||
|
|
||||||
|
|
||||||
|
def _vector_from_value(value):
|
||||||
|
if isinstance(value, App.Vector):
|
||||||
|
return value
|
||||||
|
|
||||||
|
if isinstance(value, dict):
|
||||||
|
if {"x", "y", "z"}.issubset(value.keys()):
|
||||||
|
return App.Vector(
|
||||||
|
float(value["x"]),
|
||||||
|
float(value["y"]),
|
||||||
|
float(value["z"]),
|
||||||
|
)
|
||||||
|
|
||||||
|
if isinstance(value, (list, tuple)) and len(value) >= 3:
|
||||||
|
return App.Vector(float(value[0]), float(value[1]), float(value[2]))
|
||||||
|
|
||||||
|
return None
|
||||||
|
|
||||||
|
|
||||||
|
def _rotation_from_value(value):
|
||||||
|
if not isinstance(value, dict):
|
||||||
|
return None
|
||||||
|
|
||||||
|
axis = _vector_from_value(value.get("axis"))
|
||||||
|
angle = value.get("angle")
|
||||||
|
if axis is None or angle is None:
|
||||||
|
return None
|
||||||
|
|
||||||
|
try:
|
||||||
|
angle = float(angle)
|
||||||
|
except (TypeError, ValueError):
|
||||||
|
return None
|
||||||
|
|
||||||
|
return {
|
||||||
|
"axis": axis,
|
||||||
|
"angle": angle,
|
||||||
|
}
|
||||||
|
|
||||||
|
|
||||||
|
def _rotation_from_object(source_object):
|
||||||
|
if source_object is None:
|
||||||
|
return None
|
||||||
|
|
||||||
|
rotation = None
|
||||||
|
placement = getattr(source_object, "Placement", None)
|
||||||
|
if placement is not None:
|
||||||
|
rotation = getattr(placement, "Rotation", None)
|
||||||
|
if rotation is None:
|
||||||
|
rotation = getattr(source_object, "Rotation", None)
|
||||||
|
if rotation is None:
|
||||||
|
return None
|
||||||
|
|
||||||
|
axis = getattr(rotation, "Axis", None)
|
||||||
|
if axis is None:
|
||||||
|
axis = getattr(rotation, "axis", None)
|
||||||
|
angle = getattr(rotation, "Angle", None)
|
||||||
|
if angle is None:
|
||||||
|
angle = getattr(rotation, "angle", None)
|
||||||
|
|
||||||
|
axis = _vector_from_value(axis)
|
||||||
|
if axis is None or angle is None:
|
||||||
|
return None
|
||||||
|
|
||||||
|
try:
|
||||||
|
angle = float(angle)
|
||||||
|
except (TypeError, ValueError):
|
||||||
|
return None
|
||||||
|
|
||||||
|
return {
|
||||||
|
"axis": axis,
|
||||||
|
"angle": angle,
|
||||||
|
}
|
||||||
|
|
||||||
|
|
||||||
|
def _slot_from_payload(item, source, index, source_object=None):
|
||||||
|
if not isinstance(item, dict):
|
||||||
|
return None
|
||||||
|
|
||||||
|
name = (item.get("name") or item.get("slot_name") or item.get("label") or "").strip()
|
||||||
|
if not name:
|
||||||
|
name = "SLOT_{0}".format(index + 1)
|
||||||
|
|
||||||
|
label = (item.get("label") or item.get("display_tag") or name).strip()
|
||||||
|
base = None
|
||||||
|
placement = item.get("placement")
|
||||||
|
if isinstance(placement, dict):
|
||||||
|
base = _vector_from_value(placement.get("base"))
|
||||||
|
if base is None:
|
||||||
|
base = _vector_from_value(item.get("base"))
|
||||||
|
if base is None:
|
||||||
|
base = _vector_from_value(item.get("position"))
|
||||||
|
if base is None:
|
||||||
|
base = _vector_from_value(item.get("origin"))
|
||||||
|
if base is None:
|
||||||
|
base = _vector_from_value(item.get("point"))
|
||||||
|
|
||||||
|
if base is None:
|
||||||
|
x = item.get("x")
|
||||||
|
y = item.get("y")
|
||||||
|
z = item.get("z")
|
||||||
|
if x is not None and y is not None and z is not None:
|
||||||
|
base = App.Vector(float(x), float(y), float(z))
|
||||||
|
|
||||||
|
if base is None:
|
||||||
|
return None
|
||||||
|
|
||||||
|
rotation = None
|
||||||
|
placement_rotation = None
|
||||||
|
placement = item.get("placement")
|
||||||
|
if isinstance(placement, dict):
|
||||||
|
placement_rotation = _rotation_from_value(placement.get("rotation"))
|
||||||
|
rotation = _rotation_from_value(item.get("rotation"))
|
||||||
|
if rotation is None:
|
||||||
|
rotation = placement_rotation
|
||||||
|
if rotation is None:
|
||||||
|
rotation = _rotation_from_object(source_object)
|
||||||
|
|
||||||
|
slot = {
|
||||||
|
"name": name,
|
||||||
|
"label": label,
|
||||||
|
"base": base,
|
||||||
|
"source": source,
|
||||||
|
"source_object": source_object,
|
||||||
|
}
|
||||||
|
if rotation is not None:
|
||||||
|
slot["rotation"] = rotation
|
||||||
|
return slot
|
||||||
|
|
||||||
|
|
||||||
|
def _is_child_group(obj, group_kind):
|
||||||
|
try:
|
||||||
|
return (
|
||||||
|
obj is not None
|
||||||
|
and obj.isDerivedFrom("App::DocumentObjectGroup")
|
||||||
|
and getattr(obj, "QetGroupKind", "").strip() == group_kind
|
||||||
|
)
|
||||||
|
except Exception:
|
||||||
|
return False
|
||||||
|
|
||||||
|
|
||||||
|
def collect_terminal_hints(container):
|
||||||
|
hints = []
|
||||||
|
if container is None:
|
||||||
|
return hints
|
||||||
|
|
||||||
|
for child in list(getattr(container, "Group", []) or []):
|
||||||
|
if TerminalObjects.is_terminal_hint_object(child) and not TerminalObjects.is_terminal_object(child):
|
||||||
|
hint = _slot_from_payload(
|
||||||
|
{
|
||||||
|
"name": getattr(child, "Name", ""),
|
||||||
|
"label": getattr(child, "Label", "") or getattr(child, "Name", ""),
|
||||||
|
"base": [
|
||||||
|
getattr(child.Placement.Base, "x", 0.0),
|
||||||
|
getattr(child.Placement.Base, "y", 0.0),
|
||||||
|
getattr(child.Placement.Base, "z", 0.0),
|
||||||
|
],
|
||||||
|
},
|
||||||
|
"model",
|
||||||
|
len(hints),
|
||||||
|
source_object=child,
|
||||||
|
)
|
||||||
|
if hint is not None:
|
||||||
|
hints.append(hint)
|
||||||
|
continue
|
||||||
|
|
||||||
|
if _is_child_group(child, TerminalObjects.TERMINAL_GROUP_KIND):
|
||||||
|
continue
|
||||||
|
if _is_child_group(child, TerminalObjects.WIRE_GROUP_KIND):
|
||||||
|
continue
|
||||||
|
|
||||||
|
if hasattr(child, "Group"):
|
||||||
|
hints.extend(collect_terminal_hints(child))
|
||||||
|
|
||||||
|
hints.sort(key=lambda item: (item.get("label", ""), item.get("name", "")))
|
||||||
|
return hints
|
||||||
|
|
||||||
|
|
||||||
|
def collect_model_bounding_box(container):
|
||||||
|
boxes = []
|
||||||
|
|
||||||
|
def walk(obj):
|
||||||
|
if obj is None:
|
||||||
|
return
|
||||||
|
|
||||||
|
if TerminalObjects.is_terminal_hint_object(obj):
|
||||||
|
return
|
||||||
|
|
||||||
|
if _is_child_group(obj, TerminalObjects.TERMINAL_GROUP_KIND):
|
||||||
|
return
|
||||||
|
if _is_child_group(obj, TerminalObjects.WIRE_GROUP_KIND):
|
||||||
|
return
|
||||||
|
|
||||||
|
shape = getattr(obj, "Shape", None)
|
||||||
|
if shape is not None:
|
||||||
|
try:
|
||||||
|
if not shape.isNull():
|
||||||
|
boxes.append(shape.BoundBox)
|
||||||
|
except Exception:
|
||||||
|
pass
|
||||||
|
|
||||||
|
for child in list(getattr(obj, "Group", []) or []):
|
||||||
|
walk(child)
|
||||||
|
|
||||||
|
walk(container)
|
||||||
|
|
||||||
|
if not boxes:
|
||||||
|
return None
|
||||||
|
|
||||||
|
x_min = min(box.XMin for box in boxes)
|
||||||
|
y_min = min(box.YMin for box in boxes)
|
||||||
|
z_min = min(box.ZMin for box in boxes)
|
||||||
|
x_max = max(box.XMax for box in boxes)
|
||||||
|
y_max = max(box.YMax for box in boxes)
|
||||||
|
z_max = max(box.ZMax for box in boxes)
|
||||||
|
|
||||||
|
return {
|
||||||
|
"x_min": x_min,
|
||||||
|
"y_min": y_min,
|
||||||
|
"z_min": z_min,
|
||||||
|
"x_max": x_max,
|
||||||
|
"y_max": y_max,
|
||||||
|
"z_max": z_max,
|
||||||
|
"x_span": x_max - x_min,
|
||||||
|
"y_span": y_max - y_min,
|
||||||
|
"z_span": z_max - z_min,
|
||||||
|
"x_center": (x_min + x_max) * 0.5,
|
||||||
|
"y_center": (y_min + y_max) * 0.5,
|
||||||
|
"z_center": (z_min + z_max) * 0.5,
|
||||||
|
}
|
||||||
|
|
||||||
|
|
||||||
|
def _fallback_slot_base(bbox, index, count):
|
||||||
|
if bbox is None:
|
||||||
|
step = 8.0
|
||||||
|
return App.Vector(20.0, float(index) * step, 0.0)
|
||||||
|
|
||||||
|
span_x = max(bbox["x_span"], 1.0)
|
||||||
|
span_y = max(bbox["y_span"], 1.0)
|
||||||
|
span_z = max(bbox["z_span"], 1.0)
|
||||||
|
offset = max(8.0, max(span_x, span_y, span_z) * 0.15)
|
||||||
|
|
||||||
|
if count <= 1:
|
||||||
|
y_pos = bbox["y_center"]
|
||||||
|
z_pos = bbox["z_center"]
|
||||||
|
else:
|
||||||
|
if span_y >= span_z:
|
||||||
|
step = span_y / float(count + 1)
|
||||||
|
y_pos = bbox["y_min"] + step * float(index + 1)
|
||||||
|
z_pos = bbox["z_center"]
|
||||||
|
else:
|
||||||
|
step = span_z / float(count + 1)
|
||||||
|
y_pos = bbox["y_center"]
|
||||||
|
z_pos = bbox["z_min"] + step * float(index + 1)
|
||||||
|
|
||||||
|
return App.Vector(bbox["x_max"] + offset, y_pos, z_pos)
|
||||||
|
|
||||||
|
|
||||||
|
def build_fallback_terminal_slots(container, count):
|
||||||
|
bbox = collect_model_bounding_box(container)
|
||||||
|
slots = []
|
||||||
|
for index in range(max(0, count)):
|
||||||
|
slots.append(
|
||||||
|
{
|
||||||
|
"name": "SLOT_{0}".format(index + 1),
|
||||||
|
"label": "SLOT_{0}".format(index + 1),
|
||||||
|
"base": _fallback_slot_base(bbox, index, count),
|
||||||
|
"source": "fallback",
|
||||||
|
"source_object": None,
|
||||||
|
}
|
||||||
|
)
|
||||||
|
return slots
|
||||||
|
|
||||||
|
|
||||||
|
def load_sidecar_terminal_slots(model_path):
|
||||||
|
for sidecar_path in _sidecar_candidates(model_path):
|
||||||
|
if not sidecar_path.is_file():
|
||||||
|
continue
|
||||||
|
|
||||||
|
payload = _load_json_file(sidecar_path)
|
||||||
|
if payload is None:
|
||||||
|
continue
|
||||||
|
|
||||||
|
entries = payload.get("terminal_slots")
|
||||||
|
if entries is None:
|
||||||
|
entries = payload.get("terminals")
|
||||||
|
if not isinstance(entries, list):
|
||||||
|
continue
|
||||||
|
|
||||||
|
slots = []
|
||||||
|
for index, item in enumerate(entries):
|
||||||
|
slot = _slot_from_payload(item, "sidecar", index)
|
||||||
|
if slot is not None:
|
||||||
|
slots.append(slot)
|
||||||
|
|
||||||
|
if slots:
|
||||||
|
slots.sort(key=lambda item: (item.get("label", ""), item.get("name", "")))
|
||||||
|
return slots
|
||||||
|
return []
|
||||||
|
|
||||||
|
|
||||||
|
def resolve_terminal_slots(device_group, model_path, desired_count):
|
||||||
|
desired_count = max(0, int(desired_count or 0))
|
||||||
|
hints = collect_terminal_hints(device_group)
|
||||||
|
sidecar_slots = load_sidecar_terminal_slots(model_path)
|
||||||
|
|
||||||
|
slots = []
|
||||||
|
slots.extend(hints[:desired_count])
|
||||||
|
|
||||||
|
if len(slots) < desired_count and sidecar_slots:
|
||||||
|
for slot in sidecar_slots:
|
||||||
|
if len(slots) >= desired_count:
|
||||||
|
break
|
||||||
|
slots.append(slot)
|
||||||
|
|
||||||
|
if len(slots) < desired_count:
|
||||||
|
fallback_slots = build_fallback_terminal_slots(device_group, desired_count)
|
||||||
|
for slot in fallback_slots:
|
||||||
|
if len(slots) >= desired_count:
|
||||||
|
break
|
||||||
|
slots.append(slot)
|
||||||
|
|
||||||
|
return slots[:desired_count]
|
||||||
@ -0,0 +1,325 @@
|
|||||||
|
# FreeCADExchange terminal import helpers.
|
||||||
|
|
||||||
|
from collections import OrderedDict
|
||||||
|
|
||||||
|
import FreeCAD as App
|
||||||
|
import FreeCADGui as Gui
|
||||||
|
|
||||||
|
import DeviceImport
|
||||||
|
import TerminalObjects as TerminalObjects
|
||||||
|
import TemplateSemantics
|
||||||
|
|
||||||
|
|
||||||
|
class TerminalImportError(RuntimeError):
|
||||||
|
pass
|
||||||
|
|
||||||
|
|
||||||
|
def _append_debug_log(message):
|
||||||
|
try:
|
||||||
|
DeviceImport._append_debug_log(message)
|
||||||
|
except Exception:
|
||||||
|
pass
|
||||||
|
|
||||||
|
|
||||||
|
def _normalize_terminal_entry(item, index):
|
||||||
|
if not isinstance(item, dict):
|
||||||
|
raise TerminalImportError(
|
||||||
|
"Terminal entry #{0} must be an object.".format(index)
|
||||||
|
)
|
||||||
|
|
||||||
|
terminal_uuid = (item.get("terminal_uuid") or "").strip()
|
||||||
|
if not terminal_uuid:
|
||||||
|
raise TerminalImportError(
|
||||||
|
"Terminal entry #{0} is missing terminal_uuid.".format(index)
|
||||||
|
)
|
||||||
|
|
||||||
|
instance_id = (item.get("instance_id") or "").strip()
|
||||||
|
element_uuid = (item.get("element_uuid") or "").strip()
|
||||||
|
|
||||||
|
return {
|
||||||
|
"terminal_uuid": terminal_uuid,
|
||||||
|
"instance_id": instance_id,
|
||||||
|
"element_uuid": element_uuid,
|
||||||
|
}
|
||||||
|
|
||||||
|
|
||||||
|
def _ensure_visible(obj):
|
||||||
|
try:
|
||||||
|
if getattr(obj, "ViewObject", None) is not None:
|
||||||
|
obj.ViewObject.Visibility = True
|
||||||
|
except Exception:
|
||||||
|
pass
|
||||||
|
|
||||||
|
|
||||||
|
def _hide_object(obj):
|
||||||
|
try:
|
||||||
|
if getattr(obj, "ViewObject", None) is not None:
|
||||||
|
obj.ViewObject.Visibility = False
|
||||||
|
except Exception:
|
||||||
|
pass
|
||||||
|
|
||||||
|
|
||||||
|
def _terminal_existing_index(container):
|
||||||
|
index = OrderedDict()
|
||||||
|
for obj in TerminalObjects.collect_terminal_objects(container):
|
||||||
|
terminal_uuid = getattr(obj, "QetTerminalUuid", "").strip()
|
||||||
|
if terminal_uuid and terminal_uuid not in index:
|
||||||
|
index[terminal_uuid] = obj
|
||||||
|
return index
|
||||||
|
|
||||||
|
|
||||||
|
def _device_key(device_group):
|
||||||
|
return getattr(device_group, "QetInstanceId", "").strip() or getattr(
|
||||||
|
device_group, "QetElementUuid", ""
|
||||||
|
).strip()
|
||||||
|
|
||||||
|
|
||||||
|
def _locate_device_group(doc, entry):
|
||||||
|
instance_id = entry["instance_id"]
|
||||||
|
element_uuid = entry["element_uuid"]
|
||||||
|
|
||||||
|
device_group = None
|
||||||
|
if instance_id:
|
||||||
|
device_group = TerminalObjects.find_device_group_by_instance_id(doc, instance_id)
|
||||||
|
if device_group is None and element_uuid:
|
||||||
|
device_group = DeviceImport._find_device_group(doc, element_uuid)
|
||||||
|
|
||||||
|
return device_group
|
||||||
|
|
||||||
|
|
||||||
|
def _terminal_container_for_device(doc, device_group, project_uuid):
|
||||||
|
device_instance_id = getattr(device_group, "QetInstanceId", "").strip()
|
||||||
|
return TerminalObjects.ensure_terminal_group(
|
||||||
|
doc,
|
||||||
|
device_group,
|
||||||
|
project_uuid=project_uuid,
|
||||||
|
instance_id=device_instance_id,
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
|
def _is_device_group(obj):
|
||||||
|
try:
|
||||||
|
return (
|
||||||
|
obj is not None
|
||||||
|
and getattr(obj, "Name", "").startswith(DeviceImport.DEVICE_GROUP_PREFIX)
|
||||||
|
and "QetElementUuid" in getattr(obj, "PropertiesList", [])
|
||||||
|
)
|
||||||
|
except Exception:
|
||||||
|
return False
|
||||||
|
|
||||||
|
|
||||||
|
def _terminal_slot_label(slot, terminal_uuid):
|
||||||
|
label = (slot.get("label") or "").strip()
|
||||||
|
if label:
|
||||||
|
return label
|
||||||
|
return terminal_uuid
|
||||||
|
|
||||||
|
|
||||||
|
def _slot_base(slot):
|
||||||
|
base = slot.get("base")
|
||||||
|
if isinstance(base, App.Vector):
|
||||||
|
return base
|
||||||
|
return App.Vector(0, 0, 0)
|
||||||
|
|
||||||
|
|
||||||
|
def _slot_placement(slot):
|
||||||
|
base = _slot_base(slot)
|
||||||
|
rotation = App.Rotation()
|
||||||
|
|
||||||
|
rotation_value = slot.get("rotation")
|
||||||
|
if isinstance(rotation_value, dict):
|
||||||
|
axis = rotation_value.get("axis")
|
||||||
|
angle = rotation_value.get("angle")
|
||||||
|
if isinstance(axis, App.Vector) and angle is not None:
|
||||||
|
try:
|
||||||
|
rotation = App.Rotation(axis, float(angle))
|
||||||
|
except Exception:
|
||||||
|
rotation = App.Rotation()
|
||||||
|
|
||||||
|
return App.Placement(base, rotation)
|
||||||
|
|
||||||
|
|
||||||
|
def _create_terminal_object(doc, terminal_uuid, slot, terminal_group, project_uuid, element_uuid, instance_id):
|
||||||
|
name_hint = "QETTerminal_{0}".format(TerminalObjects.safe_token(terminal_uuid))
|
||||||
|
terminal_obj = TerminalObjects.create_lcs_object(
|
||||||
|
doc,
|
||||||
|
name_hint,
|
||||||
|
placement=_slot_placement(slot),
|
||||||
|
label=_terminal_slot_label(slot, terminal_uuid),
|
||||||
|
)
|
||||||
|
terminal_group.addObject(terminal_obj)
|
||||||
|
TerminalObjects.set_terminal_semantics(
|
||||||
|
terminal_obj,
|
||||||
|
project_uuid,
|
||||||
|
element_uuid,
|
||||||
|
terminal_uuid,
|
||||||
|
instance_id,
|
||||||
|
label=_terminal_slot_label(slot, terminal_uuid),
|
||||||
|
slot_name=slot.get("name", ""),
|
||||||
|
)
|
||||||
|
_ensure_visible(terminal_obj)
|
||||||
|
return terminal_obj
|
||||||
|
|
||||||
|
|
||||||
|
def import_terminals_from_payload(payload, scene_path=""):
|
||||||
|
_append_debug_log("TerminalImport.import_terminals_from_payload entered")
|
||||||
|
|
||||||
|
if not isinstance(payload, dict):
|
||||||
|
raise TerminalImportError("Exchange payload must be an object.")
|
||||||
|
|
||||||
|
project_uuid = (payload.get("project_uuid") or "").strip()
|
||||||
|
if not project_uuid:
|
||||||
|
raise TerminalImportError("Field 'project_uuid' is required for terminal import.")
|
||||||
|
|
||||||
|
doc = DeviceImport._ensure_document(scene_path)
|
||||||
|
root_group = DeviceImport._ensure_root_group(doc, project_uuid)
|
||||||
|
_ = root_group
|
||||||
|
|
||||||
|
terminal_entries = payload.get("terminals", [])
|
||||||
|
if not isinstance(terminal_entries, list):
|
||||||
|
raise TerminalImportError("Field 'terminals' must be a list.")
|
||||||
|
|
||||||
|
report = {
|
||||||
|
"document_name": doc.Name,
|
||||||
|
"scene_path": scene_path or "",
|
||||||
|
"project_uuid": project_uuid,
|
||||||
|
"total_terminals": 0,
|
||||||
|
"imported_terminals": 0,
|
||||||
|
"updated_terminals": 0,
|
||||||
|
"removed_terminals": 0,
|
||||||
|
"reused_template_hints": 0,
|
||||||
|
"skipped_missing_device": 0,
|
||||||
|
"skipped_invalid_entry": 0,
|
||||||
|
"warnings": [],
|
||||||
|
}
|
||||||
|
|
||||||
|
grouped = OrderedDict()
|
||||||
|
for index, item in enumerate(terminal_entries):
|
||||||
|
report["total_terminals"] += 1
|
||||||
|
try:
|
||||||
|
entry = _normalize_terminal_entry(item, index)
|
||||||
|
except TerminalImportError as exc:
|
||||||
|
report["skipped_invalid_entry"] += 1
|
||||||
|
report["warnings"].append(str(exc))
|
||||||
|
continue
|
||||||
|
|
||||||
|
device_group = _locate_device_group(doc, entry)
|
||||||
|
if device_group is None:
|
||||||
|
report["skipped_missing_device"] += 1
|
||||||
|
report["warnings"].append(
|
||||||
|
"Terminal {0} could not find its parent device.".format(
|
||||||
|
entry["terminal_uuid"]
|
||||||
|
)
|
||||||
|
)
|
||||||
|
continue
|
||||||
|
|
||||||
|
key = device_group.Name
|
||||||
|
if key not in grouped:
|
||||||
|
grouped[key] = {"device_group": device_group, "entries": []}
|
||||||
|
grouped[key]["entries"].append(entry)
|
||||||
|
|
||||||
|
device_candidates = []
|
||||||
|
if root_group is not None:
|
||||||
|
device_candidates.extend(list(getattr(root_group, "Group", []) or []))
|
||||||
|
if not device_candidates:
|
||||||
|
device_candidates.extend(doc.Objects)
|
||||||
|
|
||||||
|
for device_group in device_candidates:
|
||||||
|
if not _is_device_group(device_group):
|
||||||
|
continue
|
||||||
|
|
||||||
|
item = grouped.get(device_group.Name, {"entries": []})
|
||||||
|
entries = item["entries"]
|
||||||
|
device_element_uuid = getattr(device_group, "QetElementUuid", "").strip()
|
||||||
|
device_instance_id = getattr(device_group, "QetInstanceId", "").strip()
|
||||||
|
resolved_model_path = getattr(device_group, "QetResolvedModelPath", "").strip()
|
||||||
|
|
||||||
|
terminal_group = _terminal_container_for_device(doc, device_group, project_uuid)
|
||||||
|
existing_by_uuid = _terminal_existing_index(terminal_group)
|
||||||
|
used_uuids = set()
|
||||||
|
slots = TemplateSemantics.resolve_terminal_slots(
|
||||||
|
device_group,
|
||||||
|
resolved_model_path,
|
||||||
|
len(entries),
|
||||||
|
)
|
||||||
|
|
||||||
|
for index, entry in enumerate(entries):
|
||||||
|
terminal_uuid = entry["terminal_uuid"]
|
||||||
|
payload_instance_id = entry["instance_id"]
|
||||||
|
if payload_instance_id and payload_instance_id != device_instance_id:
|
||||||
|
report["warnings"].append(
|
||||||
|
"Terminal {0} references instance_id {1} but device {2} uses {3}. The device value was kept."
|
||||||
|
.format(terminal_uuid, payload_instance_id, device_element_uuid, device_instance_id)
|
||||||
|
)
|
||||||
|
|
||||||
|
slot = slots[index] if index < len(slots) else {
|
||||||
|
"name": "SLOT_{0}".format(index + 1),
|
||||||
|
"label": terminal_uuid,
|
||||||
|
"base": App.Vector(0, 0, 0),
|
||||||
|
"source": "fallback",
|
||||||
|
"source_object": None,
|
||||||
|
}
|
||||||
|
|
||||||
|
terminal_obj = existing_by_uuid.get(terminal_uuid)
|
||||||
|
if terminal_obj is None:
|
||||||
|
terminal_obj = _create_terminal_object(
|
||||||
|
doc,
|
||||||
|
terminal_uuid,
|
||||||
|
slot,
|
||||||
|
terminal_group,
|
||||||
|
project_uuid,
|
||||||
|
device_element_uuid,
|
||||||
|
device_instance_id,
|
||||||
|
)
|
||||||
|
report["imported_terminals"] += 1
|
||||||
|
else:
|
||||||
|
TerminalObjects.set_terminal_semantics(
|
||||||
|
terminal_obj,
|
||||||
|
project_uuid,
|
||||||
|
device_element_uuid,
|
||||||
|
terminal_uuid,
|
||||||
|
device_instance_id,
|
||||||
|
label=_terminal_slot_label(slot, terminal_uuid),
|
||||||
|
slot_name=slot.get("name", ""),
|
||||||
|
)
|
||||||
|
try:
|
||||||
|
terminal_obj.Placement = _slot_placement(slot)
|
||||||
|
except Exception:
|
||||||
|
pass
|
||||||
|
_ensure_visible(terminal_obj)
|
||||||
|
report["updated_terminals"] += 1
|
||||||
|
|
||||||
|
if terminal_obj not in getattr(terminal_group, "Group", []):
|
||||||
|
terminal_group.addObject(terminal_obj)
|
||||||
|
|
||||||
|
used_uuids.add(terminal_uuid)
|
||||||
|
source_obj = slot.get("source_object")
|
||||||
|
if source_obj is not None:
|
||||||
|
_hide_object(source_obj)
|
||||||
|
report["reused_template_hints"] += 1
|
||||||
|
|
||||||
|
for terminal_uuid, terminal_obj in list(existing_by_uuid.items()):
|
||||||
|
if terminal_uuid in used_uuids:
|
||||||
|
continue
|
||||||
|
report["warnings"].append(
|
||||||
|
"Removed stale terminal {0} from device {1}.".format(
|
||||||
|
terminal_uuid, device_element_uuid
|
||||||
|
)
|
||||||
|
)
|
||||||
|
TerminalObjects.remove_object_tree(doc, terminal_obj)
|
||||||
|
report["removed_terminals"] += 1
|
||||||
|
|
||||||
|
doc.recompute()
|
||||||
|
try:
|
||||||
|
Gui.SendMsgToActiveView("ViewFit")
|
||||||
|
except Exception:
|
||||||
|
pass
|
||||||
|
|
||||||
|
_append_debug_log(
|
||||||
|
"TerminalImport finished: imported={0}, updated={1}, removed={2}".format(
|
||||||
|
report["imported_terminals"],
|
||||||
|
report["updated_terminals"],
|
||||||
|
report["removed_terminals"],
|
||||||
|
)
|
||||||
|
)
|
||||||
|
return report
|
||||||
@ -0,0 +1,386 @@
|
|||||||
|
# FreeCADExchange terminal and object helpers.
|
||||||
|
|
||||||
|
import json
|
||||||
|
import os
|
||||||
|
from pathlib import Path
|
||||||
|
|
||||||
|
import FreeCAD as App
|
||||||
|
|
||||||
|
|
||||||
|
ROOT_GROUP_NAME = "QETExchangeDevices"
|
||||||
|
ROOT_GROUP_LABEL = "QET Exchange Devices"
|
||||||
|
DEVICE_GROUP_PREFIX = "QETDevice_"
|
||||||
|
TERMINAL_GROUP_PREFIX = "QETTerminals_"
|
||||||
|
WIRE_GROUP_PREFIX = "QETWires_"
|
||||||
|
TERMINAL_GROUP_KIND = "Terminals"
|
||||||
|
WIRE_GROUP_KIND = "Wires"
|
||||||
|
TERMINAL_ROLE = "Terminal"
|
||||||
|
|
||||||
|
|
||||||
|
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 native_path(value):
|
||||||
|
text = (value or "").strip()
|
||||||
|
if not text:
|
||||||
|
return ""
|
||||||
|
return os.path.normpath(os.path.expandvars(os.path.expanduser(text)))
|
||||||
|
|
||||||
|
|
||||||
|
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 _unique_object_name(doc, base_name):
|
||||||
|
name = safe_token(base_name) or "QETObject"
|
||||||
|
if doc.getObject(name) is None:
|
||||||
|
return name
|
||||||
|
|
||||||
|
suffix = 1
|
||||||
|
while doc.getObject("{0}_{1}".format(name, suffix)) is not None:
|
||||||
|
suffix += 1
|
||||||
|
return "{0}_{1}".format(name, suffix)
|
||||||
|
|
||||||
|
|
||||||
|
def _group_kind(obj):
|
||||||
|
return (getattr(obj, "QetGroupKind", "") or "").strip()
|
||||||
|
|
||||||
|
|
||||||
|
def _is_group_candidate(obj):
|
||||||
|
try:
|
||||||
|
return bool(obj and obj.isDerivedFrom("App::DocumentObjectGroup"))
|
||||||
|
except Exception:
|
||||||
|
return False
|
||||||
|
|
||||||
|
|
||||||
|
def ensure_root_group(doc, project_uuid=""):
|
||||||
|
root = doc.getObject(ROOT_GROUP_NAME)
|
||||||
|
if root is None:
|
||||||
|
root = doc.addObject("App::DocumentObjectGroup", ROOT_GROUP_NAME)
|
||||||
|
root.Label = ROOT_GROUP_LABEL
|
||||||
|
project_uuid = (project_uuid or "").strip() or getattr(root, "QetProjectUuid", "").strip()
|
||||||
|
ensure_string_property(
|
||||||
|
root,
|
||||||
|
"QetProjectUuid",
|
||||||
|
"QET Exchange",
|
||||||
|
"Project UUID for the exchange document",
|
||||||
|
project_uuid,
|
||||||
|
)
|
||||||
|
return root
|
||||||
|
|
||||||
|
|
||||||
|
def ensure_named_child_group(
|
||||||
|
doc,
|
||||||
|
parent_group,
|
||||||
|
name_prefix,
|
||||||
|
label,
|
||||||
|
group_kind,
|
||||||
|
project_uuid="",
|
||||||
|
element_uuid="",
|
||||||
|
instance_id="",
|
||||||
|
):
|
||||||
|
target_element_uuid = (element_uuid or "").strip()
|
||||||
|
preferred_name = name_prefix + safe_token(target_element_uuid)
|
||||||
|
group = doc.getObject(preferred_name)
|
||||||
|
|
||||||
|
if group is None and _is_group_candidate(parent_group):
|
||||||
|
for candidate in getattr(parent_group, "Group", []) or []:
|
||||||
|
if _group_kind(candidate) != group_kind:
|
||||||
|
continue
|
||||||
|
if target_element_uuid and getattr(candidate, "QetElementUuid", "").strip() != target_element_uuid:
|
||||||
|
continue
|
||||||
|
group = candidate
|
||||||
|
break
|
||||||
|
|
||||||
|
if group is None:
|
||||||
|
group = doc.addObject(
|
||||||
|
"App::DocumentObjectGroup",
|
||||||
|
_unique_object_name(doc, preferred_name),
|
||||||
|
)
|
||||||
|
|
||||||
|
if parent_group is not None and 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 for the exchange document",
|
||||||
|
project_uuid,
|
||||||
|
)
|
||||||
|
ensure_string_property(
|
||||||
|
group,
|
||||||
|
"QetElementUuid",
|
||||||
|
"QET Exchange",
|
||||||
|
"Parent element UUID for the exchange group",
|
||||||
|
element_uuid,
|
||||||
|
)
|
||||||
|
ensure_string_property(
|
||||||
|
group,
|
||||||
|
"QetInstanceId",
|
||||||
|
"QET Exchange",
|
||||||
|
"Parent instance UUID for the exchange group",
|
||||||
|
instance_id,
|
||||||
|
)
|
||||||
|
return group
|
||||||
|
|
||||||
|
|
||||||
|
def ensure_terminal_group(doc, device_group, project_uuid="", instance_id=""):
|
||||||
|
element_uuid = getattr(device_group, "QetElementUuid", "").strip()
|
||||||
|
label = "QET Terminals"
|
||||||
|
return ensure_named_child_group(
|
||||||
|
doc,
|
||||||
|
device_group,
|
||||||
|
TERMINAL_GROUP_PREFIX,
|
||||||
|
label,
|
||||||
|
TERMINAL_GROUP_KIND,
|
||||||
|
project_uuid=project_uuid,
|
||||||
|
element_uuid=element_uuid,
|
||||||
|
instance_id=instance_id,
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
|
def ensure_wire_group(doc, device_group, project_uuid="", instance_id=""):
|
||||||
|
element_uuid = getattr(device_group, "QetElementUuid", "").strip()
|
||||||
|
label = "QET Wires"
|
||||||
|
return ensure_named_child_group(
|
||||||
|
doc,
|
||||||
|
device_group,
|
||||||
|
WIRE_GROUP_PREFIX,
|
||||||
|
label,
|
||||||
|
WIRE_GROUP_KIND,
|
||||||
|
project_uuid=project_uuid,
|
||||||
|
element_uuid=element_uuid,
|
||||||
|
instance_id=instance_id,
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
|
def find_child_group_by_kind(parent_group, group_kind):
|
||||||
|
if parent_group is None:
|
||||||
|
return None
|
||||||
|
for candidate in getattr(parent_group, "Group", []) or []:
|
||||||
|
if _group_kind(candidate) == group_kind:
|
||||||
|
return candidate
|
||||||
|
return None
|
||||||
|
|
||||||
|
|
||||||
|
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:
|
||||||
|
return obj
|
||||||
|
|
||||||
|
for candidate in doc.Objects:
|
||||||
|
if DEVICE_GROUP_PREFIX not in getattr(candidate, "Name", ""):
|
||||||
|
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 "QetInstanceId" in getattr(candidate, "PropertiesList", []):
|
||||||
|
if getattr(candidate, "QetInstanceId", "").strip() == target_instance_id:
|
||||||
|
if getattr(candidate, "Name", "").startswith(DEVICE_GROUP_PREFIX):
|
||||||
|
return candidate
|
||||||
|
return None
|
||||||
|
|
||||||
|
|
||||||
|
def is_lcs_like(obj):
|
||||||
|
if obj is None:
|
||||||
|
return False
|
||||||
|
try:
|
||||||
|
if obj.isDerivedFrom("App::LocalCoordinateSystem"):
|
||||||
|
return True
|
||||||
|
except Exception:
|
||||||
|
pass
|
||||||
|
return getattr(obj, "TypeId", "") in {
|
||||||
|
"Part::LocalCoordinateSystem",
|
||||||
|
"PartDesign::CoordinateSystem",
|
||||||
|
}
|
||||||
|
|
||||||
|
|
||||||
|
def is_terminal_hint_object(obj):
|
||||||
|
if not is_lcs_like(obj):
|
||||||
|
return False
|
||||||
|
role = getattr(obj, "Role", "")
|
||||||
|
return isinstance(role, str) and role.strip() == TERMINAL_ROLE
|
||||||
|
|
||||||
|
|
||||||
|
def is_terminal_object(obj):
|
||||||
|
if not is_terminal_hint_object(obj):
|
||||||
|
return False
|
||||||
|
|
||||||
|
if "QetTerminalUuid" not in getattr(obj, "PropertiesList", []):
|
||||||
|
return False
|
||||||
|
if "CanWire" not in getattr(obj, "PropertiesList", []):
|
||||||
|
return False
|
||||||
|
return bool(getattr(obj, "CanWire", False))
|
||||||
|
|
||||||
|
|
||||||
|
def terminal_origin(obj):
|
||||||
|
try:
|
||||||
|
placement = getattr(obj, "Placement", None)
|
||||||
|
if placement is not None:
|
||||||
|
return App.Vector(placement.Base.x, placement.Base.y, placement.Base.z)
|
||||||
|
except Exception:
|
||||||
|
pass
|
||||||
|
return App.Vector(0, 0, 0)
|
||||||
|
|
||||||
|
|
||||||
|
def create_lcs_object(doc, name_hint, placement=None, label=None):
|
||||||
|
base_name = safe_token(name_hint) or "QETTerminal"
|
||||||
|
object_name = _unique_object_name(doc, base_name)
|
||||||
|
lcs = None
|
||||||
|
for type_name in ("Part::LocalCoordinateSystem", "PartDesign::CoordinateSystem"):
|
||||||
|
try:
|
||||||
|
lcs = doc.addObject(type_name, object_name)
|
||||||
|
break
|
||||||
|
except Exception:
|
||||||
|
lcs = None
|
||||||
|
if lcs is None:
|
||||||
|
raise RuntimeError("FreeCAD does not provide a usable LCS object type.")
|
||||||
|
|
||||||
|
if label:
|
||||||
|
lcs.Label = label
|
||||||
|
elif name_hint:
|
||||||
|
lcs.Label = name_hint
|
||||||
|
|
||||||
|
if placement is not None:
|
||||||
|
try:
|
||||||
|
lcs.Placement = placement
|
||||||
|
except Exception:
|
||||||
|
pass
|
||||||
|
return lcs
|
||||||
|
|
||||||
|
|
||||||
|
def set_terminal_semantics(
|
||||||
|
obj,
|
||||||
|
project_uuid,
|
||||||
|
element_uuid,
|
||||||
|
terminal_uuid,
|
||||||
|
instance_id,
|
||||||
|
label="",
|
||||||
|
slot_name="",
|
||||||
|
):
|
||||||
|
ensure_string_property(
|
||||||
|
obj,
|
||||||
|
"QetProjectUuid",
|
||||||
|
"QET Exchange",
|
||||||
|
"Project UUID for this terminal",
|
||||||
|
project_uuid,
|
||||||
|
)
|
||||||
|
ensure_string_property(
|
||||||
|
obj,
|
||||||
|
"QetElementUuid",
|
||||||
|
"QET Exchange",
|
||||||
|
"Parent element UUID for this terminal",
|
||||||
|
element_uuid,
|
||||||
|
)
|
||||||
|
ensure_string_property(
|
||||||
|
obj,
|
||||||
|
"QetTerminalUuid",
|
||||||
|
"QET Exchange",
|
||||||
|
"Terminal UUID from QET",
|
||||||
|
terminal_uuid,
|
||||||
|
)
|
||||||
|
ensure_string_property(
|
||||||
|
obj,
|
||||||
|
"QetInstanceId",
|
||||||
|
"QET Exchange",
|
||||||
|
"Parent instance UUID for this terminal",
|
||||||
|
instance_id,
|
||||||
|
)
|
||||||
|
ensure_string_property(
|
||||||
|
obj,
|
||||||
|
"Role",
|
||||||
|
"QET Exchange",
|
||||||
|
"Terminal role marker",
|
||||||
|
TERMINAL_ROLE,
|
||||||
|
)
|
||||||
|
ensure_bool_property(
|
||||||
|
obj,
|
||||||
|
"CanWire",
|
||||||
|
"QET Exchange",
|
||||||
|
"Whether the terminal can be used for wiring",
|
||||||
|
True,
|
||||||
|
)
|
||||||
|
if slot_name:
|
||||||
|
ensure_string_property(
|
||||||
|
obj,
|
||||||
|
"QetTemplateSlotName",
|
||||||
|
"QET Exchange",
|
||||||
|
"Template slot name",
|
||||||
|
slot_name,
|
||||||
|
)
|
||||||
|
|
||||||
|
terminal_label = (label or "").strip() or terminal_uuid or "QET Terminal"
|
||||||
|
obj.Label = terminal_label
|
||||||
|
return obj
|
||||||
|
|
||||||
|
|
||||||
|
def remove_object_tree(doc, obj):
|
||||||
|
if obj is None:
|
||||||
|
return
|
||||||
|
|
||||||
|
children = list(getattr(obj, "Group", []) or [])
|
||||||
|
for child in children:
|
||||||
|
remove_object_tree(doc, child)
|
||||||
|
|
||||||
|
if doc.getObject(obj.Name) is not None:
|
||||||
|
doc.removeObject(obj.Name)
|
||||||
|
|
||||||
|
|
||||||
|
def collect_terminal_objects(container):
|
||||||
|
result = []
|
||||||
|
if container is None:
|
||||||
|
return result
|
||||||
|
|
||||||
|
children = list(getattr(container, "Group", []) or [])
|
||||||
|
for child in children:
|
||||||
|
if is_terminal_object(child):
|
||||||
|
result.append(child)
|
||||||
|
continue
|
||||||
|
if _is_group_candidate(child):
|
||||||
|
result.extend(collect_terminal_objects(child))
|
||||||
|
return result
|
||||||
@ -0,0 +1,192 @@
|
|||||||
|
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:
|
||||||
|
def __init__(self, axis=None, angle=None):
|
||||||
|
self.Axis = axis
|
||||||
|
self.Angle = angle
|
||||||
|
|
||||||
|
class Placement:
|
||||||
|
def __init__(self, base=None, rotation=None):
|
||||||
|
self.Base = base
|
||||||
|
self.Rotation = rotation
|
||||||
|
|
||||||
|
fake_freecad = types.ModuleType("FreeCAD")
|
||||||
|
fake_freecad.Vector = Vector
|
||||||
|
fake_freecad.Rotation = Rotation
|
||||||
|
fake_freecad.Placement = Placement
|
||||||
|
fake_freecad.Console = types.SimpleNamespace(
|
||||||
|
PrintMessage=lambda *args, **kwargs: None,
|
||||||
|
PrintWarning=lambda *args, **kwargs: None,
|
||||||
|
PrintError=lambda *args, **kwargs: None,
|
||||||
|
)
|
||||||
|
fake_freecad.ActiveDocument = None
|
||||||
|
sys.modules["FreeCAD"] = fake_freecad
|
||||||
|
|
||||||
|
fake_freecadgui = types.ModuleType("FreeCADGui")
|
||||||
|
fake_freecadgui.SendMsgToActiveView = lambda *args, **kwargs: None
|
||||||
|
fake_freecadgui.addCommand = lambda *args, **kwargs: None
|
||||||
|
fake_freecadgui.Selection = types.SimpleNamespace(getSelection=lambda: [])
|
||||||
|
sys.modules["FreeCADGui"] = fake_freecadgui
|
||||||
|
|
||||||
|
fake_importgui = types.ModuleType("ImportGui")
|
||||||
|
fake_importgui.insert = lambda *args, **kwargs: None
|
||||||
|
sys.modules["ImportGui"] = fake_importgui
|
||||||
|
|
||||||
|
fake_part = types.ModuleType("Part")
|
||||||
|
fake_part.makePolygon = lambda points: tuple(points)
|
||||||
|
sys.modules["Part"] = fake_part
|
||||||
|
|
||||||
|
|
||||||
|
class FakeViewObject:
|
||||||
|
def __init__(self):
|
||||||
|
self.Visibility = True
|
||||||
|
self.LineWidth = None
|
||||||
|
self.LineColor = None
|
||||||
|
|
||||||
|
|
||||||
|
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.Shape = None
|
||||||
|
|
||||||
|
def isDerivedFrom(self, type_name):
|
||||||
|
if self.TypeId == type_name:
|
||||||
|
return True
|
||||||
|
if type_name == "App::DocumentObjectGroup":
|
||||||
|
return self.TypeId == "App::DocumentObjectGroup"
|
||||||
|
if type_name == "App::LocalCoordinateSystem":
|
||||||
|
return self.TypeId in {"Part::LocalCoordinateSystem", "PartDesign::CoordinateSystem"}
|
||||||
|
return False
|
||||||
|
|
||||||
|
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)
|
||||||
|
|
||||||
|
|
||||||
|
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 removeObject(self, name):
|
||||||
|
self.Objects = [obj for obj in self.Objects if obj.Name != name]
|
||||||
|
|
||||||
|
def recompute(self):
|
||||||
|
return None
|
||||||
|
|
||||||
|
|
||||||
|
def _reload_modules():
|
||||||
|
for name in [
|
||||||
|
"TerminalObjects",
|
||||||
|
"DeviceImport",
|
||||||
|
"ManualWiring",
|
||||||
|
]:
|
||||||
|
sys.modules.pop(name, None)
|
||||||
|
import DeviceImport
|
||||||
|
import ManualWiring
|
||||||
|
import TerminalObjects
|
||||||
|
|
||||||
|
return DeviceImport, ManualWiring, TerminalObjects
|
||||||
|
|
||||||
|
|
||||||
|
class ManualWiringGroupTest(unittest.TestCase):
|
||||||
|
def test_manual_wire_is_added_to_device_wire_group(self):
|
||||||
|
_install_fake_freecad()
|
||||||
|
device_import, manual_wiring, terminal_objects = _reload_modules()
|
||||||
|
|
||||||
|
doc = FakeDocument()
|
||||||
|
root = terminal_objects.ensure_root_group(doc, "project-1")
|
||||||
|
device_group = doc.addObject("App::DocumentObjectGroup", "QETDevice_device_a")
|
||||||
|
root.addObject(device_group)
|
||||||
|
terminal_objects.ensure_string_property(
|
||||||
|
device_group,
|
||||||
|
"QetElementUuid",
|
||||||
|
"QET Exchange",
|
||||||
|
"Element UUID",
|
||||||
|
"device-a",
|
||||||
|
)
|
||||||
|
terminal_objects.ensure_string_property(
|
||||||
|
device_group,
|
||||||
|
"QetInstanceId",
|
||||||
|
"QET Exchange",
|
||||||
|
"Instance ID",
|
||||||
|
"instance-a",
|
||||||
|
)
|
||||||
|
terminal_objects.ensure_string_property(
|
||||||
|
device_group,
|
||||||
|
"QetProjectUuid",
|
||||||
|
"QET Exchange",
|
||||||
|
"Project UUID",
|
||||||
|
"project-1",
|
||||||
|
)
|
||||||
|
|
||||||
|
start_terminal = FakeObject("TerminalStart", "Part::LocalCoordinateSystem")
|
||||||
|
terminal_objects.set_terminal_semantics(
|
||||||
|
start_terminal,
|
||||||
|
"project-1",
|
||||||
|
"device-a",
|
||||||
|
"terminal-start",
|
||||||
|
"instance-a",
|
||||||
|
label="Start",
|
||||||
|
)
|
||||||
|
end_terminal = FakeObject("TerminalEnd", "Part::LocalCoordinateSystem")
|
||||||
|
terminal_objects.set_terminal_semantics(
|
||||||
|
end_terminal,
|
||||||
|
"project-1",
|
||||||
|
"device-a",
|
||||||
|
"terminal-end",
|
||||||
|
"instance-a",
|
||||||
|
label="End",
|
||||||
|
)
|
||||||
|
|
||||||
|
wire = manual_wiring.create_manual_wire(doc, start_terminal, end_terminal)
|
||||||
|
|
||||||
|
wire_group = terminal_objects.find_child_group_by_kind(
|
||||||
|
device_group,
|
||||||
|
terminal_objects.WIRE_GROUP_KIND,
|
||||||
|
)
|
||||||
|
|
||||||
|
self.assertIsNotNone(wire_group)
|
||||||
|
self.assertIn(wire, wire_group.Group)
|
||||||
|
self.assertNotIn(wire, root.Group)
|
||||||
|
|
||||||
|
|
||||||
|
if __name__ == "__main__":
|
||||||
|
unittest.main()
|
||||||
@ -0,0 +1,162 @@
|
|||||||
|
import importlib
|
||||||
|
import json
|
||||||
|
import sys
|
||||||
|
import tempfile
|
||||||
|
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:
|
||||||
|
def __init__(self, axis=None, angle=None):
|
||||||
|
self.axis = axis
|
||||||
|
self.angle = angle
|
||||||
|
|
||||||
|
class Placement:
|
||||||
|
def __init__(self, base=None, rotation=None):
|
||||||
|
self.Base = base
|
||||||
|
self.Rotation = rotation
|
||||||
|
|
||||||
|
fake_freecad = types.ModuleType("FreeCAD")
|
||||||
|
fake_freecad.Vector = Vector
|
||||||
|
fake_freecad.Rotation = Rotation
|
||||||
|
fake_freecad.Placement = Placement
|
||||||
|
fake_freecad.Console = types.SimpleNamespace(
|
||||||
|
PrintMessage=lambda *args, **kwargs: None,
|
||||||
|
PrintWarning=lambda *args, **kwargs: None,
|
||||||
|
PrintError=lambda *args, **kwargs: None,
|
||||||
|
)
|
||||||
|
fake_freecad.ActiveDocument = None
|
||||||
|
fake_freecad.newDocument = lambda name: types.SimpleNamespace(Name=name, Objects=[], getObject=lambda item: None)
|
||||||
|
sys.modules["FreeCAD"] = fake_freecad
|
||||||
|
|
||||||
|
fake_freecadgui = types.ModuleType("FreeCADGui")
|
||||||
|
fake_freecadgui.SendMsgToActiveView = lambda *args, **kwargs: None
|
||||||
|
fake_freecadgui.addCommand = lambda *args, **kwargs: None
|
||||||
|
fake_freecadgui.Selection = types.SimpleNamespace(getSelection=lambda: [])
|
||||||
|
sys.modules["FreeCADGui"] = fake_freecadgui
|
||||||
|
|
||||||
|
fake_importgui = types.ModuleType("ImportGui")
|
||||||
|
fake_importgui.insert = lambda *args, **kwargs: None
|
||||||
|
sys.modules["ImportGui"] = fake_importgui
|
||||||
|
|
||||||
|
|
||||||
|
def _reload_exchange_modules():
|
||||||
|
for name in [
|
||||||
|
"TerminalObjects",
|
||||||
|
"TemplateSemantics",
|
||||||
|
"DeviceImport",
|
||||||
|
"TerminalImport",
|
||||||
|
]:
|
||||||
|
sys.modules.pop(name, None)
|
||||||
|
template_semantics = importlib.import_module("TemplateSemantics")
|
||||||
|
terminal_import = importlib.import_module("TerminalImport")
|
||||||
|
return template_semantics, terminal_import
|
||||||
|
|
||||||
|
|
||||||
|
class TemplateSemanticsRotationTest(unittest.TestCase):
|
||||||
|
def test_terminal_hint_keeps_source_object_rotation(self):
|
||||||
|
_install_fake_freecad()
|
||||||
|
template_semantics, _ = _reload_exchange_modules()
|
||||||
|
|
||||||
|
fake_lcs = types.SimpleNamespace(
|
||||||
|
Name="TerminalA1",
|
||||||
|
Label="Terminal A1",
|
||||||
|
TypeId="Part::LocalCoordinateSystem",
|
||||||
|
Role="Terminal",
|
||||||
|
Placement=types.SimpleNamespace(
|
||||||
|
Base=sys.modules["FreeCAD"].Vector(4, 5, 6),
|
||||||
|
Rotation=sys.modules["FreeCAD"].Rotation(
|
||||||
|
sys.modules["FreeCAD"].Vector(0, 1, 0),
|
||||||
|
37.5,
|
||||||
|
),
|
||||||
|
),
|
||||||
|
)
|
||||||
|
container = types.SimpleNamespace(Group=[fake_lcs])
|
||||||
|
|
||||||
|
hints = template_semantics.collect_terminal_hints(container)
|
||||||
|
|
||||||
|
self.assertEqual(1, len(hints))
|
||||||
|
self.assertIn("rotation", hints[0])
|
||||||
|
self.assertEqual(37.5, hints[0]["rotation"]["angle"])
|
||||||
|
self.assertEqual(0.0, hints[0]["rotation"]["axis"].x)
|
||||||
|
self.assertEqual(1.0, hints[0]["rotation"]["axis"].y)
|
||||||
|
self.assertEqual(0.0, hints[0]["rotation"]["axis"].z)
|
||||||
|
|
||||||
|
def test_sidecar_rotation_is_normalized_from_payload(self):
|
||||||
|
_install_fake_freecad()
|
||||||
|
template_semantics, _ = _reload_exchange_modules()
|
||||||
|
|
||||||
|
with tempfile.TemporaryDirectory() as temp_dir:
|
||||||
|
model_path = Path(temp_dir) / "Relay.step"
|
||||||
|
model_path.write_text("", encoding="utf-8")
|
||||||
|
sidecar_path = Path(temp_dir) / "Relay.qet_template.json"
|
||||||
|
sidecar_path.write_text(
|
||||||
|
json.dumps(
|
||||||
|
{
|
||||||
|
"terminal_slots": [
|
||||||
|
{
|
||||||
|
"name": "A1",
|
||||||
|
"label": "A1",
|
||||||
|
"position": {"x": 10, "y": 20, "z": 30},
|
||||||
|
"rotation": {
|
||||||
|
"axis": {"x": 0, "y": 0, "z": 1},
|
||||||
|
"angle": 90,
|
||||||
|
},
|
||||||
|
}
|
||||||
|
]
|
||||||
|
}
|
||||||
|
),
|
||||||
|
encoding="utf-8",
|
||||||
|
)
|
||||||
|
|
||||||
|
slots = template_semantics.load_sidecar_terminal_slots(str(model_path))
|
||||||
|
|
||||||
|
self.assertEqual(1, len(slots))
|
||||||
|
self.assertIn("rotation", slots[0])
|
||||||
|
self.assertEqual(90.0, slots[0]["rotation"]["angle"])
|
||||||
|
self.assertEqual(0.0, slots[0]["rotation"]["axis"].x)
|
||||||
|
self.assertEqual(0.0, slots[0]["rotation"]["axis"].y)
|
||||||
|
self.assertEqual(1.0, slots[0]["rotation"]["axis"].z)
|
||||||
|
|
||||||
|
|
||||||
|
class TerminalPlacementTest(unittest.TestCase):
|
||||||
|
def test_slot_placement_uses_rotation_metadata(self):
|
||||||
|
_install_fake_freecad()
|
||||||
|
_, terminal_import = _reload_exchange_modules()
|
||||||
|
|
||||||
|
slot = {
|
||||||
|
"base": sys.modules["FreeCAD"].Vector(1, 2, 3),
|
||||||
|
"rotation": {
|
||||||
|
"axis": sys.modules["FreeCAD"].Vector(0, 0, 1),
|
||||||
|
"angle": 45.0,
|
||||||
|
},
|
||||||
|
}
|
||||||
|
|
||||||
|
placement = terminal_import._slot_placement(slot)
|
||||||
|
|
||||||
|
self.assertEqual(1.0, placement.Base.x)
|
||||||
|
self.assertEqual(2.0, placement.Base.y)
|
||||||
|
self.assertEqual(3.0, placement.Base.z)
|
||||||
|
self.assertIsNotNone(placement.Rotation)
|
||||||
|
self.assertEqual(45.0, placement.Rotation.angle)
|
||||||
|
self.assertEqual(0.0, placement.Rotation.axis.x)
|
||||||
|
self.assertEqual(0.0, placement.Rotation.axis.y)
|
||||||
|
self.assertEqual(1.0, placement.Rotation.axis.z)
|
||||||
|
|
||||||
|
|
||||||
|
if __name__ == "__main__":
|
||||||
|
unittest.main()
|
||||||
Loading…
Reference in New Issue