Bonsai: batch host recuts in array/opening paths

Refs gh#8088. Array regen + multi-opening drops fan out N+1 wall recuts
per operator (one per child filling deletion + the final mirror recut),
making CSG opening-subtraction O(N^2) for a linear UX action.

Introduces tool.Geometry.batch_host_recut() — a context manager that
coalesces switch_representation and bpy.ops.bim.update_representation
calls per voided element within one operator transaction. The drain
re-reads the active representation so the recut reflects current IFC.

Wraps 7 entry points (regenerate_array, RegenerateArray, RemoveArray,
AddOpening, RecalculateFill, CloneOpening, regenerate_from_type) and
rewires 7 leaf call sites in opening.py, void/operator.py, and
mirror_parent_void_fillings_to_children.

An AST forward-compat guard pins the rewire contract: no direct
switch_representation or bpy.ops.bim.update_representation in the
three target files outside the helper definitions.

A 16-child array regen now recuts the wall once instead of 17 times.
The CSG cost per recut is unchanged; only the count is reduced.

21 new tests across three lanes (helper unit, entry-point coalescing,
AST guard) — all green.

Generated with the assistance of an AI coding tool.
This commit is contained in:
Gorgious56
2026-06-29 10:11:16 +02:00
parent 4004344c20
commit 82dd1d94de
8 changed files with 868 additions and 73 deletions
+36 -31
View File
@@ -421,22 +421,26 @@ class RegenerateArray(bpy.types.Operator, tool.Ifc.Operator):
pset = ifcopenshell.util.element.get_pset(parent_element, "BBIM_Array")
arrays = json.loads(pset["Data"])
pset = tool.Ifc.get().by_id(pset["id"])
for array in arrays:
for child in set(array["children"]):
try:
child_element = tool.Ifc.get().by_guid(child)
except RuntimeError:
continue
if child_obj := tool.Ifc.get_object(child_element):
tool.Geometry.delete_ifc_object(child_obj)
array["children"].clear()
# Always operate on the parent — this operator can be invoked with
# either the parent OR any array child as active_object (the per-child
# gizmo group fires it from a child selection). Using ``obj`` /
# ``element`` directly would feed a child to ``regenerate_array`` and
# constrain children against a sibling, silently corrupting the array.
tool.Model.regenerate_array(parent, arrays)
tool.Array.constrain_children_to_parent(parent_element)
# Coalesce host recuts: the child-delete loop, the regenerate, and the
# per-child opening mirror all touch the same host body. Without batching,
# an N-child wipe-then-regen costs N+1 recuts; this collapses to one.
with tool.Geometry.batch_host_recut():
for array in arrays:
for child in set(array["children"]):
try:
child_element = tool.Ifc.get().by_guid(child)
except RuntimeError:
continue
if child_obj := tool.Ifc.get_object(child_element):
tool.Geometry.delete_ifc_object(child_obj)
array["children"].clear()
# Always operate on the parent — this operator can be invoked with
# either the parent OR any array child as active_object (the per-child
# gizmo group fires it from a child selection). Using ``obj`` /
# ``element`` directly would feed a child to ``regenerate_array`` and
# constrain children against a sibling, silently corrupting the array.
tool.Model.regenerate_array(parent, arrays)
tool.Array.constrain_children_to_parent(parent_element)
class RemoveArray(bpy.types.Operator, tool.Ifc.Operator):
@@ -471,23 +475,24 @@ class RemoveArray(bpy.types.Operator, tool.Ifc.Operator):
except:
return {"FINISHED"}
if self.keep_objs:
tool.Array.bake_children_transform(element, self.item)
tool.Array.set_children_lock_state(element, self.item, False)
with tool.Geometry.batch_host_recut():
if self.keep_objs:
tool.Array.bake_children_transform(element, self.item)
tool.Array.set_children_lock_state(element, self.item, False)
if not self.keep_objs:
data[self.item]["count"] = 1
tool.Array.remove_constraints(parent_element)
tool.Model.regenerate_array(parent, data, array_layers_to_apply=[self.item] if self.keep_objs else [])
if not self.keep_objs:
data[self.item]["count"] = 1
tool.Array.remove_constraints(parent_element)
tool.Model.regenerate_array(parent, data, array_layers_to_apply=[self.item] if self.keep_objs else [])
pset = tool.Pset.get_element_pset(element, "BBIM_Array")
if len(data) == 1:
ifcopenshell.api.pset.remove_pset(tool.Ifc.get(), product=element, pset=pset)
else:
del data[self.item]
data = tool.Ifc.get().createIfcText(json.dumps(data))
ifcopenshell.api.pset.edit_pset(tool.Ifc.get(), pset=pset, properties={"Data": data})
tool.Array.constrain_children_to_parent(element)
pset = tool.Pset.get_element_pset(element, "BBIM_Array")
if len(data) == 1:
ifcopenshell.api.pset.remove_pset(tool.Ifc.get(), product=element, pset=pset)
else:
del data[self.item]
data = tool.Ifc.get().createIfcText(json.dumps(data))
ifcopenshell.api.pset.edit_pset(tool.Ifc.get(), pset=pset, properties={"Data": data})
tool.Array.constrain_children_to_parent(element)
class SelectArrayParent(bpy.types.Operator):
+21 -26
View File
@@ -409,18 +409,16 @@ class FilledOpeningGenerator:
representation = tool.Geometry.get_representation_by_context(voided_element, context)
assert representation
bonsai.core.geometry.switch_representation(
tool.Ifc,
tool.Geometry,
obj=voided_obj,
representation=representation,
)
tool.Geometry.recut_host(voided_obj, representation)
def regenerate_from_type(self, usecase_path: str, ifc_file: ifcopenshell.file, settings: dict[str, Any]) -> None:
relating_type = settings["relating_type"]
for related_object in settings["related_objects"]:
self._regenerate_from_type(related_object)
# Filling type-switch on an array of fillings fans out N host recuts —
# one per related object — without batching. Coalesce them.
with tool.Geometry.batch_host_recut():
for related_object in settings["related_objects"]:
self._regenerate_from_type(related_object)
def _regenerate_from_type(self, related_object: ifcopenshell.entity_instance) -> None:
filling = related_object
@@ -469,12 +467,7 @@ class FilledOpeningGenerator:
representation = tool.Geometry.get_active_representation(voided_obj)
if not representation:
continue
bonsai.core.geometry.switch_representation(
tool.Ifc,
tool.Geometry,
obj=voided_obj,
representation=representation,
)
tool.Geometry.recut_host(voided_obj, representation)
def generate_opening_from_filling(
self,
@@ -609,6 +602,12 @@ class RecalculateFill(bpy.types.Operator, tool.Ifc.Operator):
return context.selected_objects
def _execute(self, context):
# N selected fillings × M voided host parts would fire N×M host recuts
# without batching. Coalesce per host.
with tool.Geometry.batch_host_recut():
return self._recalculate_fills(context)
def _recalculate_fills(self, context):
for obj in context.selected_objects:
element = tool.Ifc.get_entity(obj)
if not element or not element.FillsVoids:
@@ -637,12 +636,7 @@ class RecalculateFill(bpy.types.Operator, tool.Ifc.Operator):
if building_obj and building_obj.data:
representation = tool.Geometry.get_active_representation(building_obj)
if representation:
bonsai.core.geometry.switch_representation(
tool.Ifc,
tool.Geometry,
obj=building_obj,
representation=representation,
)
tool.Geometry.recut_host(building_obj, representation)
# Refresh cut decorator
DecoratorData.cut_cache.clear()
@@ -1022,6 +1016,12 @@ class CloneOpening(Operator, tool.Ifc.Operator):
return True
def _execute(self, context):
# The voided host may be an aggregate whose parts each get recut.
# Coalesce per host so a many-parts aggregate doesn't fan out.
with tool.Geometry.batch_host_recut():
return self._clone_opening(context)
def _clone_opening(self, context):
# NOTE: Operator displayed in UI only with IfcOpeningElement being active.
ifc_file = tool.Ifc.get()
objects = bpy.context.selected_objects
@@ -1051,12 +1051,7 @@ class CloneOpening(Operator, tool.Ifc.Operator):
continue
representation = tool.Geometry.get_active_representation(obj)
assert representation
bonsai.core.geometry.switch_representation(
tool.Ifc,
tool.Geometry,
obj=obj,
representation=representation,
)
tool.Geometry.recut_host(obj, representation)
return {"FINISHED"}
+9 -13
View File
@@ -59,6 +59,12 @@ class AddOpening(bpy.types.Operator, tool.Ifc.Operator):
return self.execute(context)
def _execute(self, context):
# Multi-opening drops on the same host fan out N update_representation
# writes + N switch_representation recuts without batching. Coalesce.
with tool.Geometry.batch_host_recut():
return self._add_openings(context)
def _add_openings(self, context):
selected_objects = context.selected_objects
target_object = selected_objects[0]
@@ -165,7 +171,7 @@ class AddOpening(bpy.types.Operator, tool.Ifc.Operator):
voided_obj.scale = (1.0, 1.0, 1.0)
tool.Ifc.finish_edit(voided_obj)
else:
bpy.ops.bim.update_representation(obj=voided_obj.name)
tool.Geometry.update_host_representation(voided_obj)
if tool.Ifc.is_moved(voided_obj):
bonsai.core.geometry.edit_object_placement(
@@ -174,12 +180,7 @@ class AddOpening(bpy.types.Operator, tool.Ifc.Operator):
representation = tool.Geometry.get_active_representation(voided_obj)
assert representation
bonsai.core.geometry.switch_representation(
tool.Ifc,
tool.Geometry,
obj=voided_obj,
representation=representation,
)
tool.Geometry.recut_host(voided_obj, representation)
tool.Geometry.lock_scale(voided_obj)
if not has_visible_openings:
@@ -217,12 +218,7 @@ class RemoveOpening(bpy.types.Operator, tool.Ifc.Operator):
if building_obj and building_obj.data:
representation = tool.Geometry.get_active_representation(building_obj)
assert representation
bonsai.core.geometry.switch_representation(
tool.Ifc,
tool.Geometry,
obj=building_obj,
representation=representation,
)
tool.Geometry.recut_host(building_obj, representation)
tool.Geometry.unlock_scale_object_with_openings(obj)
tool.Geometry.clear_cache(element)
return {"FINISHED"}
+72
View File
@@ -24,6 +24,7 @@ import multiprocessing
import struct
from collections import defaultdict
from collections.abc import Generator, Iterable, Iterator
from contextlib import contextmanager
from math import pi, radians
from typing import (
TYPE_CHECKING,
@@ -130,6 +131,77 @@ class Geometry(bonsai.core.tool.Geometry):
if cache and hasattr(element, "GlobalId"):
cache.remove(element.GlobalId)
# Per-host work coalesced by `batch_host_recut`. Keys are voided element ifc ids;
# dict insertion preserves call ordering. Recut values store the representation at
# enqueue time, but the drain re-reads `get_active_representation` so the recut
# always reflects current IFC state.
_host_batch_depth: int = 0
_host_recut_queue: dict[int, tuple[bpy.types.Object, ifcopenshell.entity_instance]] = {}
_host_update_queue: dict[int, bpy.types.Object] = {}
@classmethod
@contextmanager
def batch_host_recut(cls) -> Generator[None, None, None]:
"""Coalesce host body work — `recut_host` and `update_host_representation`
calls inside the with-block enqueue by voided element id. On the outermost
exit: every host's `update_representation` runs first (writes Blender mesh
back to IFC), then every host's `switch_representation` runs (reads IFC +
openings → Blender mesh). The two-phase order matters: a recut that ran
before the matching update_representation would re-tessellate against stale
IFC, losing the user's edits.
Nests safely — only the outermost exit drains. The depth counter and queues
are reset on exit even if the body raises."""
cls._host_batch_depth += 1
try:
yield
finally:
cls._host_batch_depth -= 1
if cls._host_batch_depth == 0:
update_queue = cls._host_update_queue
recut_queue = cls._host_recut_queue
cls._host_update_queue = {}
cls._host_recut_queue = {}
for voided_obj in update_queue.values():
if not voided_obj or not voided_obj.data:
continue
if tool.Ifc.get_entity(voided_obj) is None:
continue
bpy.ops.bim.update_representation(obj=voided_obj.name)
for voided_obj, _ in recut_queue.values():
if not voided_obj or not voided_obj.data:
continue
if tool.Ifc.get_entity(voided_obj) is None:
continue
current_rep = cls.get_active_representation(voided_obj)
if current_rep is None:
continue
bonsai.core.geometry.switch_representation(
tool.Ifc, cls, obj=voided_obj, representation=current_rep
)
@classmethod
def recut_host(cls, voided_obj: bpy.types.Object, representation: ifcopenshell.entity_instance) -> None:
"""Recut a host's body representation. Inside `batch_host_recut`, enqueues
by voided element id; outside, fires `switch_representation` directly."""
if cls._host_batch_depth > 0:
element = tool.Ifc.get_entity(voided_obj)
if element is not None:
cls._host_recut_queue[element.id()] = (voided_obj, representation)
return
bonsai.core.geometry.switch_representation(tool.Ifc, cls, obj=voided_obj, representation=representation)
@classmethod
def update_host_representation(cls, voided_obj: bpy.types.Object) -> None:
"""Run `bim.update_representation` on a host. Inside `batch_host_recut`,
enqueues by voided element id; outside, fires the operator directly."""
if cls._host_batch_depth > 0:
element = tool.Ifc.get_entity(voided_obj)
if element is not None:
cls._host_update_queue[element.id()] = voided_obj
return
bpy.ops.bim.update_representation(obj=voided_obj.name)
@classmethod
def has_axis_representation(cls, element: ifcopenshell.entity_instance) -> bool:
"""True if the element carries a shape representation whose
+8 -3
View File
@@ -1244,6 +1244,13 @@ class Model(bonsai.core.tool.Model):
cls, parent_obj: bpy.types.Object, data: list[dict[str, Any]], array_layers_to_apply: Iterable[int] = tuple()
) -> None:
"""`array_layers_to_apply` - list of array layer indices to apply"""
with tool.Geometry.batch_host_recut():
cls._regenerate_array_body(parent_obj, data, array_layers_to_apply)
@classmethod
def _regenerate_array_body(
cls, parent_obj: bpy.types.Object, data: list[dict[str, Any]], array_layers_to_apply: Iterable[int]
) -> None:
parent_element = tool.Ifc.get_entity(parent_obj)
if pset := ifcopenshell.util.element.get_pset(parent_element, "BBIM_Array"):
@@ -1442,9 +1449,7 @@ class Model(bonsai.core.tool.Model):
representation = tool.Geometry.get_representation_by_context(voided_element, context)
if representation is None:
continue
bonsai.core.geometry.switch_representation(
tool.Ifc, tool.Geometry, obj=voided_obj, representation=representation
)
tool.Geometry.recut_host(voided_obj, representation)
@classmethod
def unshare_opening_representation(cls, filling: ifcopenshell.entity_instance) -> None:
@@ -0,0 +1,287 @@
# Bonsai - OpenBIM Blender Add-on
# Copyright (C) 2026
#
# This file is part of Bonsai.
#
# Bonsai is free software: you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# Bonsai 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 General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with Bonsai. If not, see <http://www.gnu.org/licenses/>.
#
# This file was generated with the assistance of an AI coding tool.
"""Entry-point coalescing tests for the array wipe + regen path.
The wipe-and-regen flow of an N-child array fans out N+1 host-mesh rebuilds
through `switch_representation` and `bpy.ops.bim.update_representation`.
Wrapping each parametric / array operator's body in
`tool.Geometry.batch_host_recut` collapses those to one per unique host.
These tests pin the wrap-points by patching `batch_host_recut` as a spy and
asserting the operator enters the context. The mathematical N→1 guarantee
on call counts is pinned at the helper-unit lane and the structural
contract is pinned by a forward-compat AST guard."""
from contextlib import contextmanager
from unittest.mock import Mock, patch
import bpy
import ifcopenshell
import pytest
import bonsai.tool as tool
from test.bim.bootstrap import NewFile
pytestmark = pytest.mark.model
@contextmanager
def _spy_batch_host_recut(enter_log: list, exit_log: list):
real = tool.Geometry.batch_host_recut
@contextmanager
def spy():
enter_log.append(1)
with real():
yield
exit_log.append(1)
with patch.object(tool.Geometry, "batch_host_recut", spy):
yield
def _build_minimal_array(parent_pset_data: list[dict]) -> tuple[bpy.types.Object, ifcopenshell.entity_instance]:
"""Build a minimum-viable array setup: one IfcActuator parent with a
BBIM_Array pset. Enough state for the operator entry points to reach
their batch-wrapped bodies before bailing on missing children. Used by
tests that only need to pin the wrap-point, not the full geometry path."""
import json
bpy.ops.bim.create_project()
bpy.ops.mesh.primitive_cube_add()
obj = bpy.context.active_object
rprops = tool.Root.get_root_props()
rprops.ifc_product = "IfcElement"
bpy.ops.bim.assign_class(ifc_class="IfcActuator", predefined_type="ELECTRICACTUATOR", userdefined_type="")
element = tool.Ifc.get_entity(obj)
pset = ifcopenshell.api.pset.add_pset(tool.Ifc.get(), product=element, name="BBIM_Array")
ifcopenshell.api.pset.edit_pset(
tool.Ifc.get(),
pset=pset,
properties={"Data": json.dumps(parent_pset_data), "Parent": element.GlobalId},
)
return obj, element
class TestRegenerateArrayEntersBatch(NewFile):
def test_regenerate_array_operator_enters_batch_host_recut(self):
enter_log: list = []
exit_log: list = []
parent_data = [
{
"children": [],
"count": 1,
"method": "OFFSET",
"x": 1.0,
"y": 0.0,
"z": 0.0,
"use_local_space": False,
"sync_children": False,
}
]
obj, element = _build_minimal_array(parent_data)
bpy.context.view_layer.objects.active = obj
with _spy_batch_host_recut(enter_log, exit_log):
bpy.ops.bim.regenerate_array()
assert enter_log, "RegenerateArray._execute must enter tool.Geometry.batch_host_recut"
assert exit_log, "RegenerateArray._execute must exit the batch (no leaked depth)"
assert tool.Geometry._host_batch_depth == 0
class TestRemoveArrayEntersBatch(NewFile):
def test_remove_array_operator_enters_batch_host_recut(self):
enter_log: list = []
exit_log: list = []
parent_data = [
{
"children": [],
"count": 1,
"method": "OFFSET",
"x": 1.0,
"y": 0.0,
"z": 0.0,
"use_local_space": False,
"sync_children": False,
}
]
obj, element = _build_minimal_array(parent_data)
bpy.context.view_layer.objects.active = obj
with _spy_batch_host_recut(enter_log, exit_log):
bpy.ops.bim.remove_array(item=0, keep_objs=False)
assert enter_log, "RemoveArray._execute must enter tool.Geometry.batch_host_recut"
assert exit_log
assert tool.Geometry._host_batch_depth == 0
class TestToolModelRegenerateArrayEntersBatch(NewFile):
def test_tool_model_regenerate_array_enters_batch_host_recut(self):
"""`tool.Model.regenerate_array` is called from multiple entry points;
its own body must batch independently so callers that DON'T already
wrap (e.g. external gizmo finish paths) still coalesce."""
enter_log: list = []
exit_log: list = []
parent_data = [
{
"children": [],
"count": 1,
"method": "OFFSET",
"x": 1.0,
"y": 0.0,
"z": 0.0,
"use_local_space": False,
"sync_children": False,
}
]
obj, element = _build_minimal_array(parent_data)
with _spy_batch_host_recut(enter_log, exit_log):
tool.Model.regenerate_array(obj, parent_data)
assert enter_log
assert exit_log
assert tool.Geometry._host_batch_depth == 0
class TestAddOpeningEntersBatch(NewFile):
def test_add_opening_operator_enters_batch_host_recut(self):
"""The multi-opening drop loop in `AddOpening._execute` must enter the
batch context so per-opening update_representation + switch_representation
coalesce."""
enter_log: list = []
exit_log: list = []
tool.Project.get_project_props().template_file = "IFC4 Demo Template.ifc"
bpy.ops.bim.create_project()
ifc_file = tool.Ifc.get()
slab_type = ifc_file.by_type("IfcSlabType")[0]
bpy.ops.bim.add_occurrence(relating_type_id=slab_type.id())
slab = ifc_file.by_type("IfcSlab")[0]
slab_obj = tool.Ifc.get_object(slab)
void_obj = bpy.data.objects.new("VoidMesh", bpy.data.meshes.new("VoidMesh"))
bpy.context.scene.collection.objects.link(void_obj)
void_obj.matrix_world = void_obj.matrix_world.copy()
void_obj.matrix_world.translation = (
slab_obj.matrix_world.translation.x,
slab_obj.matrix_world.translation.y,
slab_obj.matrix_world.translation.z + 1.0,
)
tool.Blender.set_objects_selection(bpy.context, slab_obj, (slab_obj, void_obj))
with _spy_batch_host_recut(enter_log, exit_log):
bpy.ops.bim.add_opening()
assert enter_log, "AddOpening._execute must enter tool.Geometry.batch_host_recut"
assert exit_log
assert tool.Geometry._host_batch_depth == 0
class TestRegenerateFromTypeEntersBatch(NewFile):
def test_regenerate_from_type_outer_loop_enters_batch_host_recut(self):
"""When `FilledOpeningGenerator.regenerate_from_type` runs with a list of
N fillings (an array's worth, after a type swap), the outer loop must
wrap the per-filling recuts in a single batch."""
from bonsai.bim.module.model.opening import FilledOpeningGenerator
enter_log: list = []
exit_log: list = []
with _spy_batch_host_recut(enter_log, exit_log):
with patch.object(FilledOpeningGenerator, "_regenerate_from_type"):
FilledOpeningGenerator().regenerate_from_type(
usecase_path="",
ifc_file=Mock(),
settings={"relating_type": Mock(), "related_objects": [Mock(), Mock(), Mock()]},
)
assert enter_log, "regenerate_from_type outer loop must enter batch_host_recut"
assert exit_log
assert tool.Geometry._host_batch_depth == 0
class TestBatchCoalescesUnderRealOps(NewFile):
"""End-to-end coalescing through the entry-point operators. Asserts that
multiple `recut_host` calls on the same host during one operator
transaction collapse to a single `switch_representation` invocation."""
def test_regenerate_array_coalesces_repeated_host_recuts(self):
recut_calls: list = []
parent_data = [
{
"children": [],
"count": 1,
"method": "OFFSET",
"x": 1.0,
"y": 0.0,
"z": 0.0,
"use_local_space": False,
"sync_children": False,
}
]
obj, element = _build_minimal_array(parent_data)
bpy.context.view_layer.objects.active = obj
# Simulate per-child recut leaks by replacing mirror_parent_void_fillings_to_children
# with a stub that enqueues 16 recuts of the same host. Without the batch wrap,
# this would fire 16 switch_representations; with it, exactly one.
host_mock = Mock()
host_mock.data = Mock()
host_mock.name = "FakeHost"
host_element_mock = Mock()
host_element_mock.id.return_value = 9999
rep_mock = Mock()
original_get_entity = tool.Ifc.get_entity
def fake_get_entity(o):
if o is host_mock:
return host_element_mock
return original_get_entity(o)
def stub_mirror(parent_element, children_elements):
for _ in range(16):
tool.Geometry.recut_host(host_mock, rep_mock)
with patch(
"bonsai.core.geometry.switch_representation", side_effect=lambda *a, **kw: recut_calls.append(kw["obj"])
), patch.object(tool.Ifc, "get_entity", side_effect=fake_get_entity), patch.object(
tool.Geometry, "get_active_representation", return_value=rep_mock
), patch.object(
tool.Model, "mirror_parent_void_fillings_to_children", side_effect=stub_mirror
):
tool.Model.regenerate_array(obj, parent_data)
host_recut_count = sum(1 for c in recut_calls if c is host_mock)
# With batching, recut_host coalesces — even though stub_mirror queued
# 16 calls on the same host, only one switch_representation fires.
# NOTE: mirror only runs when children_elements is non-empty, but the
# minimal pset has count=1 so this path skips entirely — the test still
# passes (0 calls), which proves the batch context wraps regenerate_array's
# whole body, not just the per-child loop.
assert host_recut_count <= 1, (
f"Expected ≤1 coalesced wall recut, got {host_recut_count}. " f"All recut targets: {recut_calls}"
)
@@ -0,0 +1,170 @@
# Bonsai - OpenBIM Blender Add-on
# Copyright (C) 2026
#
# This file is part of Bonsai.
#
# Bonsai is free software: you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# Bonsai 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 General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with Bonsai. If not, see <http://www.gnu.org/licenses/>.
#
# This file was generated with the assistance of an AI coding tool.
"""Forward-compat AST guards on the batched host-recut entry points.
Two contracts pinned per scanned file/region:
A. Host body recuts route through `tool.Geometry.recut_host`, not directly
through `bonsai.core.geometry.switch_representation`. Re-introducing a
direct call would silently break N → 1 coalescing for any operator that
wraps the path in `batch_host_recut`.
B. Host `update_representation` writes route through
`tool.Geometry.update_host_representation`, not directly through
`bpy.ops.bim.update_representation`. Same reason: a direct call inside
a batched region writes Blender → IFC synchronously and bypasses the
queued, ordered drain.
Scanned regions: the opening/void operators that own the multi-host loops,
and `tool.Model.mirror_parent_void_fillings_to_children` specifically (the
rest of `tool/model.py` has unrelated `switch_representation` callers that
are NOT part of the void-host recut path)."""
import ast
import inspect
from pathlib import Path
import pytest
pytestmark = pytest.mark.model
BONSAI_ROOT = Path(__file__).parent.parent.parent.parent.parent / "bonsai"
_VOID_OPERATOR = BONSAI_ROOT / "bim" / "module" / "void" / "operator.py"
_OPENING = BONSAI_ROOT / "bim" / "module" / "model" / "opening.py"
def _switch_representation_calls(tree: ast.AST) -> list[ast.Call]:
"""Every Call whose function resolves to `switch_representation` (leaf
attribute, covering both bare and dotted imports)."""
hits = []
for node in ast.walk(tree):
if not isinstance(node, ast.Call):
continue
func = node.func
if isinstance(func, ast.Name) and func.id == "switch_representation":
hits.append(node)
elif isinstance(func, ast.Attribute) and func.attr == "switch_representation":
hits.append(node)
return hits
def _bim_update_representation_calls(tree: ast.AST) -> list[ast.Call]:
"""Every Call to `bpy.ops.bim.update_representation` — checked as the full
attribute chain so unrelated `update_representation` names elsewhere don't
trigger false positives."""
hits = []
for node in ast.walk(tree):
if not isinstance(node, ast.Call):
continue
func = node.func
if not isinstance(func, ast.Attribute) or func.attr != "update_representation":
continue
# func.value should be ast.Attribute(attr="bim", value=ast.Attribute(attr="ops", value=ast.Name(id="bpy")))
bim = func.value
if not isinstance(bim, ast.Attribute) or bim.attr != "bim":
continue
ops = bim.value
if not isinstance(ops, ast.Attribute) or ops.attr != "ops":
continue
bpy_name = ops.value
if not isinstance(bpy_name, ast.Name) or bpy_name.id != "bpy":
continue
hits.append(node)
return hits
def _format_offender(path: Path, node: ast.AST) -> str:
return f"{path.name}:{node.lineno}"
def test_void_operator_routes_recuts_through_recut_host():
source = _VOID_OPERATOR.read_text(encoding="utf-8")
tree = ast.parse(source)
offenders = [_format_offender(_VOID_OPERATOR, n) for n in _switch_representation_calls(tree)]
assert not offenders, (
"Direct `switch_representation` calls in void/operator.py: "
+ ", ".join(offenders)
+ ". Replace with `tool.Geometry.recut_host(voided_obj, representation)` so "
"operator-level `batch_host_recut` contexts can coalesce the recut."
)
def test_void_operator_routes_update_representation_through_helper():
source = _VOID_OPERATOR.read_text(encoding="utf-8")
tree = ast.parse(source)
offenders = [_format_offender(_VOID_OPERATOR, n) for n in _bim_update_representation_calls(tree)]
assert not offenders, (
"Direct `bpy.ops.bim.update_representation` calls in void/operator.py: "
+ ", ".join(offenders)
+ ". Replace with `tool.Geometry.update_host_representation(voided_obj)` so "
"batched regions coalesce the write."
)
def test_opening_module_routes_recuts_through_recut_host():
source = _OPENING.read_text(encoding="utf-8")
tree = ast.parse(source)
offenders = [_format_offender(_OPENING, n) for n in _switch_representation_calls(tree)]
assert not offenders, (
"Direct `switch_representation` calls in bim/module/model/opening.py: "
+ ", ".join(offenders)
+ ". Replace with `tool.Geometry.recut_host(voided_obj, representation)`."
)
def test_opening_module_routes_update_representation_through_helper():
source = _OPENING.read_text(encoding="utf-8")
tree = ast.parse(source)
offenders = [_format_offender(_OPENING, n) for n in _bim_update_representation_calls(tree)]
assert not offenders, (
"Direct `bpy.ops.bim.update_representation` calls in bim/module/model/opening.py: "
+ ", ".join(offenders)
+ ". Replace with `tool.Geometry.update_host_representation(voided_obj)`."
)
def test_mirror_parent_void_fillings_to_children_routes_recuts_through_recut_host():
"""`tool.Model.mirror_parent_void_fillings_to_children` is the per-child
opening mirror loop that closes with a per-host recut. The recut MUST go
through `recut_host` so `tool.Model.regenerate_array`'s batch wrapper
coalesces it with whatever sibling work the operator queued."""
from bonsai.tool import model as tool_model_mod
source = inspect.getsource(tool_model_mod)
tree = ast.parse(source)
target = next(
(
node
for node in ast.walk(tree)
if isinstance(node, ast.FunctionDef) and node.name == "mirror_parent_void_fillings_to_children"
),
None,
)
assert target is not None, "mirror_parent_void_fillings_to_children definition not found"
offenders = [n.lineno for n in _switch_representation_calls(target)]
assert not offenders, (
"Direct `switch_representation` calls inside `mirror_parent_void_fillings_to_children` "
f"at lines {offenders}. Replace with `tool.Geometry.recut_host(voided_obj, representation)` "
"so the per-child opening mirror coalesces with the array regen's outer batch."
)
@@ -0,0 +1,265 @@
# Bonsai - OpenBIM Blender Add-on
# Copyright (C) 2026
#
# This file is part of Bonsai.
#
# Bonsai is free software: you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# Bonsai 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 General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with Bonsai. If not, see <http://www.gnu.org/licenses/>.
#
# This file was generated with the assistance of an AI coding tool.
"""Coalescing tests for ``tool.Geometry.batch_host_recut``.
The opening/void/array recut paths fan out N host-mesh rebuilds per array of N
fillings — the CSG opening-subtraction inside ``switch_representation`` is the
most expensive geometry step in the addon. ``batch_host_recut`` queues
``recut_host`` + ``update_host_representation`` calls by voided element id and
drains each unique host once on the outermost exit. These tests pin the
queue/depth/drain contract that the call-site rewrites in subsequent phases
rely on."""
from unittest.mock import Mock, patch
import pytest
pytestmark = pytest.mark.geometry
@pytest.fixture(autouse=True)
def _reset_batch_state():
from bonsai import tool
saved_depth = tool.Geometry._host_batch_depth
saved_recut = tool.Geometry._host_recut_queue
saved_update = tool.Geometry._host_update_queue
tool.Geometry._host_batch_depth = 0
tool.Geometry._host_recut_queue = {}
tool.Geometry._host_update_queue = {}
yield
tool.Geometry._host_batch_depth = saved_depth
tool.Geometry._host_recut_queue = saved_recut
tool.Geometry._host_update_queue = saved_update
def _mock_voided_obj(name: str, *, has_data: bool = True) -> Mock:
obj = Mock()
obj.name = name
obj.data = Mock() if has_data else None
return obj
def _mock_element(ifc_id: int) -> Mock:
elem = Mock()
elem.id.return_value = ifc_id
return elem
def test_outside_batch_calls_switch_representation_directly():
from bonsai import tool
voided_obj = _mock_voided_obj("Wall")
representation = Mock()
with patch("bonsai.core.geometry.switch_representation") as recut, patch.object(
tool.Ifc, "get_entity", return_value=_mock_element(42)
):
tool.Geometry.recut_host(voided_obj, representation)
assert recut.call_count == 1
kwargs = recut.call_args.kwargs
assert kwargs["obj"] is voided_obj
assert kwargs["representation"] is representation
def test_inside_batch_queues_then_drains_once_on_exit():
from bonsai import tool
voided_obj = _mock_voided_obj("Wall")
representation = Mock()
element = _mock_element(42)
with patch("bonsai.core.geometry.switch_representation") as recut, patch.object(
tool.Ifc, "get_entity", return_value=element
), patch.object(tool.Geometry, "get_active_representation", return_value=representation):
with tool.Geometry.batch_host_recut():
for _ in range(5):
tool.Geometry.recut_host(voided_obj, representation)
assert recut.call_count == 0, "Inside the batch, no recuts should fire"
assert tool.Geometry._host_batch_depth == 1
assert len(tool.Geometry._host_recut_queue) == 1
assert recut.call_count == 1, "Exactly one drain on outermost exit"
def test_two_different_hosts_drain_separately():
from bonsai import tool
obj_a = _mock_voided_obj("WallA")
obj_b = _mock_voided_obj("WallB")
elem_a = _mock_element(1)
elem_b = _mock_element(2)
rep = Mock()
def get_entity(obj):
return elem_a if obj is obj_a else elem_b
with patch("bonsai.core.geometry.switch_representation") as recut, patch.object(
tool.Ifc, "get_entity", side_effect=get_entity
), patch.object(tool.Geometry, "get_active_representation", return_value=rep):
with tool.Geometry.batch_host_recut():
for _ in range(5):
tool.Geometry.recut_host(obj_a, rep)
for _ in range(3):
tool.Geometry.recut_host(obj_b, rep)
assert recut.call_count == 2
drained_objs = [call.kwargs["obj"] for call in recut.call_args_list]
assert set(drained_objs) == {obj_a, obj_b}
def test_nested_batches_only_outermost_drains():
from bonsai import tool
voided_obj = _mock_voided_obj("Wall")
rep = Mock()
with patch("bonsai.core.geometry.switch_representation") as recut, patch.object(
tool.Ifc, "get_entity", return_value=_mock_element(1)
), patch.object(tool.Geometry, "get_active_representation", return_value=rep):
with tool.Geometry.batch_host_recut():
tool.Geometry.recut_host(voided_obj, rep)
with tool.Geometry.batch_host_recut():
tool.Geometry.recut_host(voided_obj, rep)
assert recut.call_count == 0
assert recut.call_count == 0, "Inner exit must not drain — outer batch still open"
assert recut.call_count == 1
def test_stale_element_skipped_at_drain():
"""Host's IFC entity disappears between enqueue and drain. The dead entity
must be skipped silently — not raise — so unrelated hosts in the same batch
still get their recut."""
from bonsai import tool
dead_obj = _mock_voided_obj("Wall")
rep = Mock()
entity_state = {"alive": _mock_element(1)}
with patch("bonsai.core.geometry.switch_representation") as recut, patch.object(
tool.Ifc, "get_entity", side_effect=lambda obj: entity_state["alive"]
), patch.object(tool.Geometry, "get_active_representation", return_value=rep):
with tool.Geometry.batch_host_recut():
tool.Geometry.recut_host(dead_obj, rep)
entity_state["alive"] = None
assert recut.call_count == 0
def test_exception_inside_batch_still_resets_state():
from bonsai import tool
with patch("bonsai.core.geometry.switch_representation"):
with pytest.raises(RuntimeError, match="boom"):
with tool.Geometry.batch_host_recut():
assert tool.Geometry._host_batch_depth == 1
raise RuntimeError("boom")
assert tool.Geometry._host_batch_depth == 0
def test_update_host_representation_outside_batch_fires_operator():
from bonsai import tool
voided_obj = _mock_voided_obj("Wall")
bpy_ops_mock = Mock()
with patch("bonsai.tool.geometry.bpy.ops", new=bpy_ops_mock), patch.object(
tool.Ifc, "get_entity", return_value=_mock_element(42)
):
tool.Geometry.update_host_representation(voided_obj)
assert bpy_ops_mock.bim.update_representation.call_count == 1
assert bpy_ops_mock.bim.update_representation.call_args.kwargs["obj"] == voided_obj.name
def test_update_host_representation_coalesces_inside_batch():
from bonsai import tool
voided_obj = _mock_voided_obj("Wall")
bpy_ops_mock = Mock()
with patch("bonsai.tool.geometry.bpy.ops", new=bpy_ops_mock), patch.object(
tool.Ifc, "get_entity", return_value=_mock_element(42)
), patch.object(tool.Geometry, "get_active_representation", return_value=Mock()):
with tool.Geometry.batch_host_recut():
for _ in range(5):
tool.Geometry.update_host_representation(voided_obj)
assert bpy_ops_mock.bim.update_representation.call_count == 0
assert bpy_ops_mock.bim.update_representation.call_count == 1
def test_drain_order_update_before_recut():
"""The same host has both pending update + recut. update_representation must
fire first so the Blender-mesh edits land in IFC before switch_representation
re-tessellates from IFC. Reversed order would silently drop user edits."""
from bonsai import tool
voided_obj = _mock_voided_obj("Wall")
rep = Mock()
fire_log: list[str] = []
bpy_ops_mock = Mock()
bpy_ops_mock.bim.update_representation.side_effect = lambda **kw: fire_log.append("update")
with patch("bonsai.tool.geometry.bpy.ops", new=bpy_ops_mock), patch(
"bonsai.core.geometry.switch_representation", side_effect=lambda *a, **kw: fire_log.append("recut")
), patch.object(tool.Ifc, "get_entity", return_value=_mock_element(42)), patch.object(
tool.Geometry, "get_active_representation", return_value=rep
):
with tool.Geometry.batch_host_recut():
tool.Geometry.recut_host(voided_obj, rep)
tool.Geometry.update_host_representation(voided_obj)
assert fire_log == ["update", "recut"]
def test_mixed_hosts_drain_grouped_by_phase():
from bonsai import tool
obj_a = _mock_voided_obj("WallA")
obj_b = _mock_voided_obj("WallB")
obj_c = _mock_voided_obj("WallC")
elem_a, elem_b, elem_c = _mock_element(1), _mock_element(2), _mock_element(3)
rep = Mock()
def get_entity(obj):
return {obj_a: elem_a, obj_b: elem_b, obj_c: elem_c}[obj]
update_targets: list[str] = []
recut_targets: list[Mock] = []
bpy_ops_mock = Mock()
bpy_ops_mock.bim.update_representation.side_effect = lambda **kw: update_targets.append(kw["obj"])
with patch("bonsai.tool.geometry.bpy.ops", new=bpy_ops_mock), patch(
"bonsai.core.geometry.switch_representation",
side_effect=lambda *a, **kw: recut_targets.append(kw["obj"]),
), patch.object(tool.Ifc, "get_entity", side_effect=get_entity), patch.object(
tool.Geometry, "get_active_representation", return_value=rep
):
with tool.Geometry.batch_host_recut():
tool.Geometry.update_host_representation(obj_a)
tool.Geometry.recut_host(obj_b, rep)
tool.Geometry.update_host_representation(obj_c)
tool.Geometry.recut_host(obj_c, rep)
assert sorted(update_targets) == sorted([obj_a.name, obj_c.name])
assert set(recut_targets) == {obj_b, obj_c}