mirror of
https://github.com/IfcOpenShell/IfcOpenShell.git
synced 2026-08-09 17:31:45 +00:00
Bonsai: iterate every duplicated entity in relationship recreation
Sweep [0]-indexing in recreate_aggregate, recreate_connections, and recreate_port_connections so batched N-child duplicates recreate relationships on every new child, not just the first. Single-source callers unaffected (loop collapses to one iteration on 1-element lists). Relates to #8088. Generated with the assistance of an AI coding tool.
This commit is contained in:
Gorgious56
@@ -248,32 +248,30 @@ class Duplicate(bonsai.core.tool.Duplicate):
|
||||
old_to_new: dict[ifcopenshell.entity_instance, list[ifcopenshell.entity_instance]],
|
||||
) -> None:
|
||||
for element, data in relationship.items():
|
||||
try:
|
||||
new_relating_element = old_to_new.get(data.relating_element)[0]
|
||||
new_related_element = old_to_new.get(data.related_element)[0]
|
||||
except (KeyError, IndexError, TypeError):
|
||||
continue
|
||||
new_rel = tool.Ifc.run(
|
||||
"geometry.connect_path",
|
||||
relating_element=new_relating_element,
|
||||
related_element=new_related_element,
|
||||
relating_connection=data.relating_connection_type,
|
||||
related_connection=data.related_connection_type,
|
||||
)
|
||||
new_relating_elements = old_to_new.get(data.relating_element) or []
|
||||
new_related_elements = old_to_new.get(data.related_element) or []
|
||||
# connect_path hardcodes priorities to []; restore them post-hoc.
|
||||
priority_attrs: dict[str, Any] = {}
|
||||
if data.relating_priorities:
|
||||
priority_attrs["RelatingPriorities"] = data.relating_priorities
|
||||
if data.related_priorities:
|
||||
priority_attrs["RelatedPriorities"] = data.related_priorities
|
||||
if new_rel is not None and priority_attrs:
|
||||
try:
|
||||
tool.Ifc.run("attribute.edit_attributes", product=new_rel, attributes=priority_attrs)
|
||||
except (RuntimeError, ifcopenshell.Error) as e:
|
||||
cls._emit_warning(
|
||||
f"connection priority restore failed for {new_rel}; "
|
||||
f"duplicate has empty RelatingPriorities/RelatedPriorities: {e}"
|
||||
)
|
||||
for new_relating_element, new_related_element in zip(new_relating_elements, new_related_elements):
|
||||
new_rel = tool.Ifc.run(
|
||||
"geometry.connect_path",
|
||||
relating_element=new_relating_element,
|
||||
related_element=new_related_element,
|
||||
relating_connection=data.relating_connection_type,
|
||||
related_connection=data.related_connection_type,
|
||||
)
|
||||
if new_rel is not None and priority_attrs:
|
||||
try:
|
||||
tool.Ifc.run("attribute.edit_attributes", product=new_rel, attributes=priority_attrs)
|
||||
except (RuntimeError, ifcopenshell.Error) as e:
|
||||
cls._emit_warning(
|
||||
f"connection priority restore failed for {new_rel}; "
|
||||
f"duplicate has empty RelatingPriorities/RelatedPriorities: {e}"
|
||||
)
|
||||
|
||||
@classmethod
|
||||
def recreate_port_connections(
|
||||
@@ -283,46 +281,43 @@ class Duplicate(bonsai.core.tool.Duplicate):
|
||||
) -> None:
|
||||
"""Recreate ``IfcRelConnectsPorts`` between duplicates; skip records whose duplicate's port count diverges from the snapshot."""
|
||||
for relating_element, records in snapshot.by_element.items():
|
||||
new_relatings = old_to_new.get(relating_element) or []
|
||||
expected_relating = snapshot.port_counts.get(relating_element)
|
||||
for record in records:
|
||||
related_element = record.related_element
|
||||
try:
|
||||
new_relating = old_to_new[relating_element][0]
|
||||
new_related = old_to_new[related_element][0]
|
||||
except (KeyError, IndexError):
|
||||
continue
|
||||
|
||||
new_relating_ports = tool.System.get_ports(new_relating)
|
||||
new_related_ports = tool.System.get_ports(new_related)
|
||||
|
||||
expected_relating = snapshot.port_counts.get(relating_element)
|
||||
if expected_relating is not None and len(new_relating_ports) != expected_relating:
|
||||
cls._emit_warning(
|
||||
f"port reconnect skipped — duplicate has {len(new_relating_ports)} ports, "
|
||||
f"snapshot had {expected_relating}"
|
||||
)
|
||||
continue
|
||||
new_relateds = old_to_new.get(related_element) or []
|
||||
expected_related = snapshot.port_counts.get(related_element)
|
||||
if expected_related is not None and len(new_related_ports) != expected_related:
|
||||
cls._emit_warning(
|
||||
f"port reconnect skipped — duplicate has {len(new_related_ports)} ports, "
|
||||
f"snapshot had {expected_related}"
|
||||
)
|
||||
continue
|
||||
for new_relating, new_related in zip(new_relatings, new_relateds):
|
||||
new_relating_ports = tool.System.get_ports(new_relating)
|
||||
new_related_ports = tool.System.get_ports(new_related)
|
||||
|
||||
try:
|
||||
new_port_a = new_relating_ports[record.relating_port_index]
|
||||
new_port_b = new_related_ports[record.related_port_index]
|
||||
except IndexError:
|
||||
cls._emit_warning(
|
||||
f"port reconnect skipped — record references port index past the duplicate's port list"
|
||||
)
|
||||
continue
|
||||
try:
|
||||
tool.Ifc.run(
|
||||
"system.connect_port",
|
||||
port1=new_port_a,
|
||||
port2=new_port_b,
|
||||
direction=record.direction or "NOTDEFINED",
|
||||
)
|
||||
except (RuntimeError, ifcopenshell.Error) as e:
|
||||
cls._emit_warning(f"port reconnect failed between duplicates: {e}")
|
||||
if expected_relating is not None and len(new_relating_ports) != expected_relating:
|
||||
cls._emit_warning(
|
||||
f"port reconnect skipped — duplicate has {len(new_relating_ports)} ports, "
|
||||
f"snapshot had {expected_relating}"
|
||||
)
|
||||
continue
|
||||
if expected_related is not None and len(new_related_ports) != expected_related:
|
||||
cls._emit_warning(
|
||||
f"port reconnect skipped — duplicate has {len(new_related_ports)} ports, "
|
||||
f"snapshot had {expected_related}"
|
||||
)
|
||||
continue
|
||||
|
||||
try:
|
||||
new_port_a = new_relating_ports[record.relating_port_index]
|
||||
new_port_b = new_related_ports[record.related_port_index]
|
||||
except IndexError:
|
||||
cls._emit_warning(
|
||||
f"port reconnect skipped — record references port index past the duplicate's port list"
|
||||
)
|
||||
continue
|
||||
try:
|
||||
tool.Ifc.run(
|
||||
"system.connect_port",
|
||||
port1=new_port_a,
|
||||
port2=new_port_b,
|
||||
direction=record.direction or "NOTDEFINED",
|
||||
)
|
||||
except (RuntimeError, ifcopenshell.Error) as e:
|
||||
cls._emit_warning(f"port reconnect failed between duplicates: {e}")
|
||||
|
||||
@@ -373,35 +373,37 @@ class Root(bonsai.core.tool.Root):
|
||||
try:
|
||||
new_aggregate = old_to_new[old_aggregate]
|
||||
except:
|
||||
bonsai.core.aggregate.unassign_object(
|
||||
tool.Ifc,
|
||||
tool.Aggregate,
|
||||
tool.Collector,
|
||||
relating_obj=tool.Ifc.get_object(old_aggregate),
|
||||
related_obj=tool.Ifc.get_object(new[0]),
|
||||
)
|
||||
continue
|
||||
|
||||
bonsai.core.aggregate.assign_object(
|
||||
tool.Ifc,
|
||||
tool.Aggregate,
|
||||
tool.Collector,
|
||||
relating_obj=tool.Ifc.get_object(new_aggregate[0]),
|
||||
related_obj=tool.Ifc.get_object(new[0]),
|
||||
)
|
||||
|
||||
# Make sure that the array children also get reassigned to the correct aggregate
|
||||
pset = ifcopenshell.util.element.get_pset(new[0], "BBIM_Array")
|
||||
if pset:
|
||||
array_children = tool.Array.get_all_children_objects(new[0])
|
||||
for obj in array_children:
|
||||
bonsai.core.aggregate.assign_object(
|
||||
for new_entity in new:
|
||||
bonsai.core.aggregate.unassign_object(
|
||||
tool.Ifc,
|
||||
tool.Aggregate,
|
||||
tool.Collector,
|
||||
relating_obj=tool.Ifc.get_object(new_aggregate[0]),
|
||||
related_obj=tool.Ifc.get_object(tool.Ifc.get_entity(obj)),
|
||||
relating_obj=tool.Ifc.get_object(old_aggregate),
|
||||
related_obj=tool.Ifc.get_object(new_entity),
|
||||
)
|
||||
continue
|
||||
|
||||
for new_entity in new:
|
||||
bonsai.core.aggregate.assign_object(
|
||||
tool.Ifc,
|
||||
tool.Aggregate,
|
||||
tool.Collector,
|
||||
relating_obj=tool.Ifc.get_object(new_aggregate[0]),
|
||||
related_obj=tool.Ifc.get_object(new_entity),
|
||||
)
|
||||
|
||||
# Make sure that the array children also get reassigned to the correct aggregate
|
||||
pset = ifcopenshell.util.element.get_pset(new_entity, "BBIM_Array")
|
||||
if pset:
|
||||
array_children = tool.Array.get_all_children_objects(new_entity)
|
||||
for obj in array_children:
|
||||
bonsai.core.aggregate.assign_object(
|
||||
tool.Ifc,
|
||||
tool.Aggregate,
|
||||
tool.Collector,
|
||||
relating_obj=tool.Ifc.get_object(new_aggregate[0]),
|
||||
related_obj=tool.Ifc.get_object(tool.Ifc.get_entity(obj)),
|
||||
)
|
||||
|
||||
if new_aggregate is None:
|
||||
return
|
||||
|
||||
@@ -374,3 +374,216 @@ class TestOrphanArrayChildPrune(NewFile):
|
||||
assert child_obj is not None, "every surviving child must have a live Blender object"
|
||||
|
||||
|
||||
class TestRecreateAggregateIteratesAllNew(NewFile):
|
||||
"""Pins the [0]-indexing sweep in tool/root.py recreate_aggregate. When the
|
||||
new-list has N>1 entries (the batched-duplicate shape), every entry must be
|
||||
aggregate-assigned, not just new[0]."""
|
||||
|
||||
def test_iterates_assign_object_per_new_entity_when_old_has_aggregate(self):
|
||||
from unittest.mock import Mock
|
||||
|
||||
old_assembly = Mock()
|
||||
old_assembly.is_a = lambda c: c == "IfcElementAssembly"
|
||||
old_parent_aggregate = Mock()
|
||||
old_parent_aggregate.is_a = lambda c: False
|
||||
|
||||
new_assemblies = [Mock(), Mock(), Mock()]
|
||||
new_parent_aggregate = [Mock()]
|
||||
|
||||
old_to_new = {old_assembly: new_assemblies, old_parent_aggregate: new_parent_aggregate}
|
||||
|
||||
with patch(
|
||||
"ifcopenshell.util.element.get_aggregate",
|
||||
side_effect=lambda e: old_parent_aggregate if e is old_assembly else None,
|
||||
), patch("bonsai.core.aggregate.assign_object") as assign_mock, patch(
|
||||
"ifcopenshell.util.element.get_pset", return_value=None
|
||||
), patch.object(
|
||||
tool.Ifc, "get_object", side_effect=lambda e: Mock()
|
||||
), patch.object(
|
||||
tool.Blender, "select_and_activate_single_object"
|
||||
):
|
||||
tool.Root.recreate_aggregate(old_to_new)
|
||||
|
||||
assert (
|
||||
assign_mock.call_count == 3
|
||||
), f"recreate_aggregate must assign each of N new entities (not just new[0]); got {assign_mock.call_count}"
|
||||
|
||||
def test_iterates_unassign_object_per_new_entity_when_aggregate_missing(self):
|
||||
from unittest.mock import Mock
|
||||
|
||||
old_assembly = Mock()
|
||||
old_assembly.is_a = lambda c: c == "IfcElementAssembly"
|
||||
old_parent_aggregate = Mock()
|
||||
|
||||
new_assemblies = [Mock(), Mock(), Mock()]
|
||||
old_to_new = {old_assembly: new_assemblies} # parent aggregate NOT in old_to_new
|
||||
|
||||
with patch(
|
||||
"ifcopenshell.util.element.get_aggregate",
|
||||
side_effect=lambda e: old_parent_aggregate if e is old_assembly else None,
|
||||
), patch("bonsai.core.aggregate.unassign_object") as unassign_mock, patch.object(
|
||||
tool.Ifc, "get_object", side_effect=lambda e: Mock()
|
||||
):
|
||||
tool.Root.recreate_aggregate(old_to_new)
|
||||
|
||||
assert unassign_mock.call_count == 3, (
|
||||
f"recreate_aggregate must unassign each of N new entities when parent aggregate is missing; "
|
||||
f"got {unassign_mock.call_count}"
|
||||
)
|
||||
|
||||
|
||||
class TestRecreateConnectionsZipsPairs(NewFile):
|
||||
"""Pins the [0]-indexing sweep in tool/duplicate.py recreate_connections. When
|
||||
both sides of a connection are duplicated N times, zip-pair the N new
|
||||
relating with N new related; when only one side is duplicated, skip."""
|
||||
|
||||
def _make_connection_data(self):
|
||||
from unittest.mock import Mock
|
||||
|
||||
data = Mock()
|
||||
data.relating_element = Mock()
|
||||
data.related_element = Mock()
|
||||
data.relating_connection_type = "ATSTART"
|
||||
data.related_connection_type = "ATEND"
|
||||
data.relating_priorities = []
|
||||
data.related_priorities = []
|
||||
return data
|
||||
|
||||
def test_zips_n_pairs_when_both_sides_duplicated(self):
|
||||
from unittest.mock import Mock
|
||||
|
||||
data = self._make_connection_data()
|
||||
old_to_new = {
|
||||
data.relating_element: [Mock(), Mock(), Mock()],
|
||||
data.related_element: [Mock(), Mock(), Mock()],
|
||||
}
|
||||
relationship = {Mock(): data}
|
||||
|
||||
with patch.object(tool.Ifc, "run", return_value=None) as run_mock:
|
||||
tool.Duplicate.recreate_connections(relationship, old_to_new)
|
||||
|
||||
connect_calls = [c for c in run_mock.call_args_list if c.args and c.args[0] == "geometry.connect_path"]
|
||||
assert (
|
||||
len(connect_calls) == 3
|
||||
), f"zip-pair must create 3 connect_path calls for 3-vs-3 batched duplicate; got {len(connect_calls)}"
|
||||
|
||||
def test_skips_when_other_side_not_duplicated(self):
|
||||
from unittest.mock import Mock
|
||||
|
||||
data = self._make_connection_data()
|
||||
# Only relating side is in old_to_new; related side was NOT duplicated.
|
||||
old_to_new = {data.relating_element: [Mock(), Mock(), Mock()]}
|
||||
relationship = {Mock(): data}
|
||||
|
||||
with patch.object(tool.Ifc, "run", return_value=None) as run_mock:
|
||||
tool.Duplicate.recreate_connections(relationship, old_to_new)
|
||||
|
||||
connect_calls = [c for c in run_mock.call_args_list if c.args and c.args[0] == "geometry.connect_path"]
|
||||
assert (
|
||||
connect_calls == []
|
||||
), "when only one side of a connection is in old_to_new, no connections should be recreated"
|
||||
|
||||
def test_single_pair_case_unchanged(self):
|
||||
"""Pre-sweep behavior (1 source -> 1 new) must still work — zip with two 1-element lists."""
|
||||
from unittest.mock import Mock
|
||||
|
||||
data = self._make_connection_data()
|
||||
old_to_new = {
|
||||
data.relating_element: [Mock()],
|
||||
data.related_element: [Mock()],
|
||||
}
|
||||
relationship = {Mock(): data}
|
||||
|
||||
with patch.object(tool.Ifc, "run", return_value=None) as run_mock:
|
||||
tool.Duplicate.recreate_connections(relationship, old_to_new)
|
||||
|
||||
connect_calls = [c for c in run_mock.call_args_list if c.args and c.args[0] == "geometry.connect_path"]
|
||||
assert len(connect_calls) == 1
|
||||
|
||||
|
||||
class TestRecreatePortConnectionsZipsPairs(NewFile):
|
||||
"""Pins the [0]-indexing sweep in tool/duplicate.py recreate_port_connections.
|
||||
When both sides of a port-to-port connection are duplicated N times, the
|
||||
connection must be recreated on every pair of new siblings — not just the
|
||||
first. Matters for arrayed MEP segments (pipes / ducts / cables) where each
|
||||
child in the array should stay connected to its neighbour after regen."""
|
||||
|
||||
def _make_snapshot(self, relating_element, records, port_counts):
|
||||
from bonsai.tool.duplicate import PortConnectionSnapshot
|
||||
|
||||
return PortConnectionSnapshot(
|
||||
by_element={relating_element: records},
|
||||
port_counts=port_counts,
|
||||
)
|
||||
|
||||
def _make_record(self, related_element, relating_port_index=0, related_port_index=0, direction="SOURCE"):
|
||||
from bonsai.tool.duplicate import PortConnectionRecord
|
||||
|
||||
return PortConnectionRecord(
|
||||
relating_port_index=relating_port_index,
|
||||
related_element=related_element,
|
||||
related_port_index=related_port_index,
|
||||
direction=direction,
|
||||
)
|
||||
|
||||
def test_zips_n_pairs_when_both_sides_duplicated(self):
|
||||
from unittest.mock import Mock
|
||||
|
||||
relating_old = Mock()
|
||||
related_old = Mock()
|
||||
record = self._make_record(related_old)
|
||||
snapshot = self._make_snapshot(relating_old, [record], port_counts={})
|
||||
|
||||
old_to_new = {
|
||||
relating_old: [Mock(), Mock(), Mock()],
|
||||
related_old: [Mock(), Mock(), Mock()],
|
||||
}
|
||||
|
||||
fake_ports = [Mock(), Mock()]
|
||||
with patch.object(tool.System, "get_ports", return_value=fake_ports), patch.object(
|
||||
tool.Ifc, "run", return_value=None
|
||||
) as run_mock:
|
||||
tool.Duplicate.recreate_port_connections(snapshot, old_to_new)
|
||||
|
||||
connect_calls = [c for c in run_mock.call_args_list if c.args and c.args[0] == "system.connect_port"]
|
||||
assert (
|
||||
len(connect_calls) == 3
|
||||
), f"zip-pair must create 3 connect_port calls for 3-vs-3 batched MEP duplicate; got {len(connect_calls)}"
|
||||
|
||||
def test_skips_when_other_side_not_duplicated(self):
|
||||
from unittest.mock import Mock
|
||||
|
||||
relating_old = Mock()
|
||||
related_old = Mock()
|
||||
record = self._make_record(related_old)
|
||||
snapshot = self._make_snapshot(relating_old, [record], port_counts={})
|
||||
|
||||
# Only relating side is in old_to_new.
|
||||
old_to_new = {relating_old: [Mock(), Mock(), Mock()]}
|
||||
|
||||
with patch.object(tool.System, "get_ports", return_value=[Mock()]), patch.object(
|
||||
tool.Ifc, "run", return_value=None
|
||||
) as run_mock:
|
||||
tool.Duplicate.recreate_port_connections(snapshot, old_to_new)
|
||||
|
||||
connect_calls = [c for c in run_mock.call_args_list if c.args and c.args[0] == "system.connect_port"]
|
||||
assert connect_calls == [], "when only one side is in old_to_new, no port connections should be recreated"
|
||||
|
||||
def test_single_pair_case_unchanged(self):
|
||||
"""Pre-sweep behavior (1 source -> 1 new) must still work — zip with two 1-element lists."""
|
||||
from unittest.mock import Mock
|
||||
|
||||
relating_old = Mock()
|
||||
related_old = Mock()
|
||||
record = self._make_record(related_old)
|
||||
snapshot = self._make_snapshot(relating_old, [record], port_counts={})
|
||||
|
||||
old_to_new = {relating_old: [Mock()], related_old: [Mock()]}
|
||||
|
||||
with patch.object(tool.System, "get_ports", return_value=[Mock()]), patch.object(
|
||||
tool.Ifc, "run", return_value=None
|
||||
) as run_mock:
|
||||
tool.Duplicate.recreate_port_connections(snapshot, old_to_new)
|
||||
|
||||
connect_calls = [c for c in run_mock.call_args_list if c.args and c.args[0] == "system.connect_port"]
|
||||
assert len(connect_calls) == 1
|
||||
|
||||
Reference in New Issue
Block a user