From c299f0b191e4efa127946a1ea50bd10e57ee5e68 Mon Sep 17 00:00:00 2001 From: Gorgious56 Date: Wed, 8 Jul 2026 13:19:26 +0200 Subject: [PATCH] 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. --- src/bonsai/bonsai/tool/duplicate.py | 113 +++++----- src/bonsai/bonsai/tool/root.py | 52 +++-- .../model/test_array_duplicate_batched.py | 213 ++++++++++++++++++ 3 files changed, 294 insertions(+), 84 deletions(-) diff --git a/src/bonsai/bonsai/tool/duplicate.py b/src/bonsai/bonsai/tool/duplicate.py index eb3630ccf1..fd8258672b 100644 --- a/src/bonsai/bonsai/tool/duplicate.py +++ b/src/bonsai/bonsai/tool/duplicate.py @@ -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}") diff --git a/src/bonsai/bonsai/tool/root.py b/src/bonsai/bonsai/tool/root.py index d7b7e0596f..ef2464d964 100644 --- a/src/bonsai/bonsai/tool/root.py +++ b/src/bonsai/bonsai/tool/root.py @@ -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 diff --git a/src/bonsai/test/bim/module/model/test_array_duplicate_batched.py b/src/bonsai/test/bim/module/model/test_array_duplicate_batched.py index 285eccab39..a3696fba9f 100644 --- a/src/bonsai/test/bim/module/model/test_array_duplicate_batched.py +++ b/src/bonsai/test/bim/module/model/test_array_duplicate_batched.py @@ -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