From fd8aa6d14221815e7051f248722d9b74493a0fd2 Mon Sep 17 00:00:00 2001 From: Dion Moult Date: Tue, 1 Feb 2022 16:51:42 +1100 Subject: [PATCH] IfcOpenShell API now supports port connections and disconnections. --- .../ifcopenshell/api/system/connect_port.py | 122 ++++++++++++++++++ .../api/system/disconnect_port.py | 39 ++++++ .../test/api/system/test_connect_port.py | 115 +++++++++++++++++ .../test/api/system/test_disconnect_port.py | 32 +++++ 4 files changed, 308 insertions(+) create mode 100644 src/ifcopenshell-python/ifcopenshell/api/system/connect_port.py create mode 100644 src/ifcopenshell-python/ifcopenshell/api/system/disconnect_port.py create mode 100644 src/ifcopenshell-python/test/api/system/test_connect_port.py create mode 100644 src/ifcopenshell-python/test/api/system/test_disconnect_port.py diff --git a/src/ifcopenshell-python/ifcopenshell/api/system/connect_port.py b/src/ifcopenshell-python/ifcopenshell/api/system/connect_port.py new file mode 100644 index 0000000000..79af2319e8 --- /dev/null +++ b/src/ifcopenshell-python/ifcopenshell/api/system/connect_port.py @@ -0,0 +1,122 @@ +# IfcOpenShell - IFC toolkit and geometry engine +# Copyright (C) 2022 Dion Moult +# +# This file is part of IfcOpenShell. +# +# IfcOpenShell is free software: you can redistribute it and/or modify +# it under the terms of the GNU Lesser General Public License as published by +# the Free Software Foundation, either version 3 of the License, or +# (at your option) any later version. +# +# IfcOpenShell is distributed in the hope that it will be useful, +# but WITHOUT ANY WARRANTY; without even the implied warranty of +# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +# GNU Lesser General Public License for more details. +# +# You should have received a copy of the GNU Lesser General Public License +# along with IfcOpenShell. If not, see . + +import ifcopenshell +import ifcopenshell.api + + +class Usecase: + def __init__(self, file, **settings): + self.file = file + self.settings = { + "port1": None, + "port2": None, + "direction": "NOTDEFINED", + "element": None, + } + for key, value in settings.items(): + self.settings[key] = value + + def execute(self): + # Note: there are a number of ambiguities with port connectivity. We + # assume system topology is represented by a directed graph. In other + # words, SOURCEANDSINK and NOTDEFINED implies a two way connection, with + # two IfcRelConnectsPorts. SOURCE or SINK by itself implies a one way + # connection. NOTDEFINED semantically implies that although you may + # traverse the graph either direction, the direction has not been + # determined yet by the engineer. None is not allowed as a direction as + # we assume None means that no connection is made. + + if self.settings["port1"] == self.settings["port2"]: + return + + self.purge_existing_connections_to_other_ports() + + if self.settings["direction"] == "SOURCE": + self.settings["port1"].FlowDirection = "SOURCE" + self.settings["port2"].FlowDirection = "SINK" + elif self.settings["direction"] == "SINK": + self.settings["port1"].FlowDirection = "SINK" + self.settings["port2"].FlowDirection = "SOURCE" + else: + self.settings["port1"].FlowDirection = self.settings["direction"] + self.settings["port2"].FlowDirection = self.settings["direction"] + + if self.settings["direction"] in ["SOURCE", "SOURCEANDSINK", "NOTDEFINED"]: + self.set_connected_to() + else: + self.purge_connected_to() + + if self.settings["direction"] in ["SINK", "SOURCEANDSINK", "NOTDEFINED"]: + self.set_connected_from() + else: + self.purge_connected_from() + + self.set_realising_element() + + def purge_existing_connections_to_other_ports(self): + for rel in self.settings["port1"].ConnectedTo or []: + if rel.RelatedPort != self.settings["port2"]: + self.file.remove(rel) + for rel in self.settings["port1"].ConnectedFrom or []: + if rel.RelatingPort != self.settings["port2"]: + self.file.remove(rel) + for rel in self.settings["port2"].ConnectedTo or []: + if rel.RelatedPort != self.settings["port1"]: + self.file.remove(rel) + for rel in self.settings["port2"].ConnectedFrom or []: + if rel.RelatingPort != self.settings["port1"]: + self.file.remove(rel) + + def set_connected_to(self): + if self.settings["port1"].ConnectedTo: + return + + self.file.create_entity( + "IfcRelConnectsPorts", + GlobalId=ifcopenshell.guid.new(), + OwnerHistory=ifcopenshell.api.run("owner.create_owner_history", self.file), + RelatingPort=self.settings["port1"], + RelatedPort=self.settings["port2"], + ) + + def set_connected_from(self): + if self.settings["port1"].ConnectedFrom: + return + + self.file.create_entity( + "IfcRelConnectsPorts", + GlobalId=ifcopenshell.guid.new(), + OwnerHistory=ifcopenshell.api.run("owner.create_owner_history", self.file), + RelatingPort=self.settings["port2"], + RelatedPort=self.settings["port1"], + ) + + def purge_connected_to(self): + for rel in self.settings["port1"].ConnectedTo or []: + self.file.remove(rel) + + def purge_connected_from(self): + for rel in self.settings["port1"].ConnectedFrom or []: + self.file.remove(rel) + + def set_realising_element(self): + for rel in self.settings["port1"].ConnectedTo or []: + rel.RealizingElement = self.settings["element"] + for rel in self.settings["port1"].ConnectedFrom or []: + rel.RealizingElement = self.settings["element"] diff --git a/src/ifcopenshell-python/ifcopenshell/api/system/disconnect_port.py b/src/ifcopenshell-python/ifcopenshell/api/system/disconnect_port.py new file mode 100644 index 0000000000..35add85315 --- /dev/null +++ b/src/ifcopenshell-python/ifcopenshell/api/system/disconnect_port.py @@ -0,0 +1,39 @@ +# IfcOpenShell - IFC toolkit and geometry engine +# Copyright (C) 2022 Dion Moult +# +# This file is part of IfcOpenShell. +# +# IfcOpenShell is free software: you can redistribute it and/or modify +# it under the terms of the GNU Lesser General Public License as published by +# the Free Software Foundation, either version 3 of the License, or +# (at your option) any later version. +# +# IfcOpenShell is distributed in the hope that it will be useful, +# but WITHOUT ANY WARRANTY; without even the implied warranty of +# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +# GNU Lesser General Public License for more details. +# +# You should have received a copy of the GNU Lesser General Public License +# along with IfcOpenShell. If not, see . + +import ifcopenshell +import ifcopenshell.api + + +class Usecase: + def __init__(self, file, **settings): + self.file = file + self.settings = { + "port": None, + } + for key, value in settings.items(): + self.settings[key] = value + + def execute(self): + rels = self.settings["port"].ConnectedTo or [] + rels += self.settings["port"].ConnectedFrom or [] + + for rel in rels: + rel.RelatingPort.FlowDirection = None + rel.RelatedPort.FlowDirection = None + self.file.remove(rel) diff --git a/src/ifcopenshell-python/test/api/system/test_connect_port.py b/src/ifcopenshell-python/test/api/system/test_connect_port.py new file mode 100644 index 0000000000..ab1325389a --- /dev/null +++ b/src/ifcopenshell-python/test/api/system/test_connect_port.py @@ -0,0 +1,115 @@ +# IfcOpenShell - IFC toolkit and geometry engine +# Copyright (C) 2022 Dion Moult +# +# This file is part of IfcOpenShell. +# +# IfcOpenShell is free software: you can redistribute it and/or modify +# it under the terms of the GNU Lesser General Public License as published by +# the Free Software Foundation, either version 3 of the License, or +# (at your option) any later version. +# +# IfcOpenShell is distributed in the hope that it will be useful, +# but WITHOUT ANY WARRANTY; without even the implied warranty of +# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +# GNU Lesser General Public License for more details. +# +# You should have received a copy of the GNU Lesser General Public License +# along with IfcOpenShell. If not, see . + +import test.bootstrap +import ifcopenshell.api +import ifcopenshell.util.system + + +class TestConnectPort(test.bootstrap.IFC4): + def test_connecting_a_port(self): + port = ifcopenshell.api.run("system.add_port", self.file) + port2 = ifcopenshell.api.run("system.add_port", self.file) + ifcopenshell.api.run("system.connect_port", self.file, port1=port, port2=port2) + assert port.ConnectedTo[0].RelatedPort == port2 + assert port2.ConnectedFrom[0].RelatingPort == port + assert port2.ConnectedTo[0].RelatedPort == port + assert port.ConnectedFrom[0].RelatingPort == port2 + assert port.FlowDirection == "NOTDEFINED" + assert port2.FlowDirection == "NOTDEFINED" + assert len(self.file.by_type("IfcRelConnectsPorts")) == 2 + + def test_not_connecting_a_port_twice(self): + port = ifcopenshell.api.run("system.add_port", self.file) + port2 = ifcopenshell.api.run("system.add_port", self.file) + ifcopenshell.api.run("system.connect_port", self.file, port1=port, port2=port2) + ifcopenshell.api.run("system.connect_port", self.file, port1=port, port2=port2) + ifcopenshell.api.run("system.connect_port", self.file, port1=port2, port2=port) + assert port.ConnectedTo[0].RelatedPort == port2 + assert port2.ConnectedFrom[0].RelatingPort == port + assert port2.ConnectedTo[0].RelatedPort == port + assert port.ConnectedFrom[0].RelatingPort == port2 + assert port.FlowDirection == "NOTDEFINED" + assert port2.FlowDirection == "NOTDEFINED" + assert len(self.file.by_type("IfcRelConnectsPorts")) == 2 + + def test_connecting_a_port_from_source_to_sink(self): + port = ifcopenshell.api.run("system.add_port", self.file) + port2 = ifcopenshell.api.run("system.add_port", self.file) + ifcopenshell.api.run("system.connect_port", self.file, port1=port, port2=port2, direction="SOURCE") + assert port.ConnectedTo[0].RelatedPort == port2 + assert port2.ConnectedFrom[0].RelatingPort == port + assert port.FlowDirection == "SOURCE" + assert port2.FlowDirection == "SINK" + assert len(self.file.by_type("IfcRelConnectsPorts")) == 1 + + def test_connecting_a_port_from_sink_to_source(self): + port = ifcopenshell.api.run("system.add_port", self.file) + port2 = ifcopenshell.api.run("system.add_port", self.file) + ifcopenshell.api.run("system.connect_port", self.file, port1=port, port2=port2, direction="SINK") + assert port2.ConnectedTo[0].RelatedPort == port + assert port.ConnectedFrom[0].RelatingPort == port2 + assert port.FlowDirection == "SINK" + assert port2.FlowDirection == "SOURCE" + assert len(self.file.by_type("IfcRelConnectsPorts")) == 1 + + def test_connecting_a_port_as_both_source_and_sink(self): + port = ifcopenshell.api.run("system.add_port", self.file) + port2 = ifcopenshell.api.run("system.add_port", self.file) + ifcopenshell.api.run("system.connect_port", self.file, port1=port, port2=port2, direction="SOURCEANDSINK") + assert port.ConnectedTo[0].RelatedPort == port2 + assert port2.ConnectedFrom[0].RelatingPort == port + assert port2.ConnectedTo[0].RelatedPort == port + assert port.ConnectedFrom[0].RelatingPort == port2 + assert port.FlowDirection == "SOURCEANDSINK" + assert port2.FlowDirection == "SOURCEANDSINK" + assert len(self.file.by_type("IfcRelConnectsPorts")) == 2 + + def test_ports_can_only_connect_to_one_port_at_a_time(self): + port = ifcopenshell.api.run("system.add_port", self.file) + port2 = ifcopenshell.api.run("system.add_port", self.file) + port3 = ifcopenshell.api.run("system.add_port", self.file) + ifcopenshell.api.run("system.connect_port", self.file, port1=port, port2=port3, direction="SOURCEANDSINK") + ifcopenshell.api.run("system.connect_port", self.file, port1=port, port2=port2, direction="SOURCE") + assert port.ConnectedTo[0].RelatedPort == port2 + assert port2.ConnectedFrom[0].RelatingPort == port + assert port.FlowDirection == "SOURCE" + assert port2.FlowDirection == "SINK" + assert len(self.file.by_type("IfcRelConnectsPorts")) == 1 + assert not port3.ConnectedTo + assert not port3.ConnectedFrom + + def test_changing_a_port_from_source_and_sink_to_only_source(self): + port = ifcopenshell.api.run("system.add_port", self.file) + port2 = ifcopenshell.api.run("system.add_port", self.file) + ifcopenshell.api.run("system.connect_port", self.file, port1=port, port2=port2, direction="SOURCEANDSINK") + ifcopenshell.api.run("system.connect_port", self.file, port1=port, port2=port2, direction="SOURCE") + assert port.ConnectedTo[0].RelatedPort == port2 + assert port2.ConnectedFrom[0].RelatingPort == port + assert port.FlowDirection == "SOURCE" + assert port2.FlowDirection == "SINK" + assert len(self.file.by_type("IfcRelConnectsPorts")) == 1 + + def test_connecting_ports_with_a_realising_element(self): + port = ifcopenshell.api.run("system.add_port", self.file) + port2 = ifcopenshell.api.run("system.add_port", self.file) + element = ifcopenshell.api.run("root.create_entity", self.file, ifc_class="IfcDuctFitting") + ifcopenshell.api.run("system.connect_port", self.file, port1=port, port2=port2, element=element) + assert self.file.by_type("IfcRelConnectsPorts")[0].RealizingElement == element + ifcopenshell.api.run("system.connect_port", self.file, port1=port, port2=port2) + assert self.file.by_type("IfcRelConnectsPorts")[0].RealizingElement is None diff --git a/src/ifcopenshell-python/test/api/system/test_disconnect_port.py b/src/ifcopenshell-python/test/api/system/test_disconnect_port.py new file mode 100644 index 0000000000..8b2ade0310 --- /dev/null +++ b/src/ifcopenshell-python/test/api/system/test_disconnect_port.py @@ -0,0 +1,32 @@ +# IfcOpenShell - IFC toolkit and geometry engine +# Copyright (C) 2022 Dion Moult +# +# This file is part of IfcOpenShell. +# +# IfcOpenShell is free software: you can redistribute it and/or modify +# it under the terms of the GNU Lesser General Public License as published by +# the Free Software Foundation, either version 3 of the License, or +# (at your option) any later version. +# +# IfcOpenShell is distributed in the hope that it will be useful, +# but WITHOUT ANY WARRANTY; without even the implied warranty of +# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +# GNU Lesser General Public License for more details. +# +# You should have received a copy of the GNU Lesser General Public License +# along with IfcOpenShell. If not, see . + +import test.bootstrap +import ifcopenshell.api +import ifcopenshell.util.system + + +class TestConnectPort(test.bootstrap.IFC4): + def test_disconnecting_a_port(self): + port = ifcopenshell.api.run("system.add_port", self.file) + port2 = ifcopenshell.api.run("system.add_port", self.file) + ifcopenshell.api.run("system.connect_port", self.file, port1=port, port2=port2, direction="NOTDEFINED") + ifcopenshell.api.run("system.disconnect_port", self.file, port=port) + assert port.FlowDirection == None + assert port2.FlowDirection == None + assert len(self.file.by_type("IfcRelConnectsPorts")) == 0