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 N1 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}