Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
51 changes: 36 additions & 15 deletions openhcs/runtime/napari_viewer_server.py
Original file line number Diff line number Diff line change
Expand Up @@ -3241,8 +3241,6 @@ def select_indices(
"""Select one exact, validated set of native feature rows."""

indices = tuple(sorted(set(data_indices)))
if not indices:
raise ValueError("Napari result selection must contain at least one row.")
for data_index in indices:
if isinstance(data_index, bool) or not isinstance(data_index, Integral):
raise TypeError("Napari result selection indices must be integers.")
Expand All @@ -3256,6 +3254,25 @@ def select_indices(
)
return observed

@classmethod
def displayed_indices(
cls,
layer: NapariLayerHandle,
data_indices: Iterable[int],
) -> tuple[int, ...]:
"""Project a logical subject onto Napari's actual current slice.

Subject membership does not imply native displayed geometry. In
particular, Shapes cannot construct a multi-member interaction box
when none of its selected members has displayed vertices.
"""
indices = tuple(sorted(set(data_indices)))
for data_index in indices:
cls.require_data_index(layer, data_index)
native_layer = cast(napari.layers.Shapes | napari.layers.Points, layer)
displayed = set(native_layer._indices_view)
return tuple(index for index in indices if index in displayed)


class NapariResultSelectionController:
"""Turn native result selection into an unambiguous visible viewer state."""
Expand Down Expand Up @@ -3537,7 +3554,6 @@ def _queue_selection(self, layer: NapariLayerHandle) -> None:
data_index = (
newly_selected[-1] if newly_selected else state.selected_data_indices[-1]
)
self._synchronize_linked_group(layer, data_index)
layer_reference = weakref.ref(layer)
QTimer.singleShot(
0,
Expand Down Expand Up @@ -3571,14 +3587,23 @@ def _synchronize_linked_group(
layer: NapariLayerHandle,
data_index: int,
) -> tuple[tuple[NapariLayerHandle, tuple[int, ...]], ...]:
linked = self._linked_group_members(layer, data_index)
linked = tuple(
(candidate, NapariResultElementSelectionAuthority.displayed_indices(
candidate, member_indices,
))
for candidate, member_indices in self._linked_group_members(layer, data_index)
)
if not any(candidate is layer and data_index in indices for candidate, indices in linked):
raise ValueError(
f"Viewer data_index {data_index} has no displayed geometry on the current slice."
)
self._synchronizing_group_selection = True
try:
for candidate, member_indices in linked:
cast(NapariShapesLayerHandle, candidate).selected_data = set(
member_indices
observed = NapariResultElementSelectionAuthority.select_indices(
candidate, member_indices,
)
self._observed_indices[candidate] = member_indices
self._observed_indices[candidate] = observed.selected_data_indices
finally:
self._synchronizing_group_selection = False
self._notify_selection_observers()
Expand Down Expand Up @@ -3645,15 +3670,11 @@ def _apply_selection(
),
)
self.server.viewer.dims.current_step = viewer_step
# Slice first, then derive the visible members of each linked
# subject. Never expand an off-slice singleton before navigation.
self._synchronize_linked_group(native_layer, data_index)
finally:
# Napari clears Shapes.selected_data when a dims change slices the
# selected geometry out of view. Re-assert the same authoritative
# member even if navigation fails so a UI interaction cannot erase
# an otherwise valid native selection.
try:
self._synchronize_linked_group(native_layer, data_index)
finally:
self._refreshing_highlights = False
self._refreshing_highlights = False
except Exception:
logger.exception(
"Failed to navigate to selected Napari result element %d on %s",
Expand Down
75 changes: 69 additions & 6 deletions tests/unit/test_viewer_result_element_selection.py
Original file line number Diff line number Diff line change
Expand Up @@ -187,17 +187,76 @@ def test_native_navigation_acknowledges_exact_bound_linked_selection(qtbot):
np.testing.assert_array_equal(before, after)


def test_singleton_selection_stays_strict_with_bound_group(qtbot):
def test_singleton_selection_expands_only_after_queued_navigation(qtbot):
from napari.components import ViewerModel

viewer = ViewerModel()
paths = [np.asarray([[i, i], [i + 1, i + 1]], dtype=float) for i in range(3)]
layer = viewer.add_shapes(paths, shape_type="path", features={"owner": [8, 8, 9]})
binding = NapariResultSelectionGroupBinding("native-test-subject", "owner")
_viewer_server(viewer, layer, group_binding=binding)
with pytest.raises(RuntimeError, match="requested native data selection"):
NapariResultElementSelectionAuthority.select(layer, 1)
assert layer.selected_data == {0, 1}
state = NapariResultElementSelectionAuthority.select(layer, 1)
assert state.selected_data_indices == (1,)
assert layer.selected_data == {1}
qtbot.waitUntil(lambda: layer.selected_data == {0, 1}, timeout=1000)


def test_linked_selection_projects_off_slice_members_without_changing_subject(qtbot):
from napari.components import ViewerModel

viewer = ViewerModel()
viewer.add_image(np.zeros((2, 8, 8), dtype=np.uint8))
paths = [np.array([[0, i, i], [0, i + 1, i + 1]], dtype=float) for i in range(2)]
layer = viewer.add_shapes(paths, shape_type="path", features={"owner": [8, 8]})
linked_paths = [coordinates + [1, 0, 0] for coordinates in paths]
linked = viewer.add_shapes(linked_paths, shape_type="path", features={"owner": [8, 8]})
viewer.dims.current_step = (0, 0, 0)
binding = NapariResultSelectionGroupBinding("native-test-subject", "owner")
server, _, _, _ = _viewer_server(viewer, layer, group_binding=binding)
server.result_selection_controller.bind(linked, group_binding=binding)
original_data = [coordinates.copy() for coordinates in linked.data]

# Original native behaviour reproduces the retained failure signature.
with pytest.raises(ValueError, match="zero-size array.*minimum"):
linked.selected_data = {0, 1}
linked.selected_data = set()

state = server.result_selection_controller.select(layer, 1)

assert state.selected_data_indices == (0, 1)
assert linked.selected_data == set()
assert server.result_selection_controller._linked_group_members(layer, 1) == (
(layer, (0, 1)), (linked, (0, 1)),
)
assert viewer.dims.current_step == (0, 0, 0)
for before, after in zip(original_data, linked.data, strict=True):
np.testing.assert_array_equal(before, after)


def test_linked_points_use_the_same_native_slice_projection(qtbot):
from napari.components import ViewerModel

viewer = ViewerModel()
viewer.add_image(np.zeros((2, 8, 8), dtype=np.uint8))
layer = viewer.add_shapes(
[np.array([[0, 0, 0], [0, 1, 1]], dtype=float)],
shape_type="path", features={"owner": [8]},
)
linked = viewer.add_points(
np.array([[0, 2, 2], [1, 3, 3]], dtype=float), features={"owner": [8, 8]},
)
viewer.dims.current_step = (0, 0, 0)
binding = NapariResultSelectionGroupBinding("native-test-subject", "owner")
server, _, _, _ = _viewer_server(viewer, layer, group_binding=binding)
server.result_selection_controller.bind(linked, group_binding=binding)

state = server.result_selection_controller.select(layer, 0)

assert state.selected_data_indices == (0,)
assert linked.selected_data == {0}
assert server.result_selection_controller._linked_group_members(layer, 0) == (
(layer, (0,)), (linked, (0, 1)),
)


def test_bound_selection_rejects_unrelated_native_members(qtbot):
Expand All @@ -221,7 +280,7 @@ def add_unrelated_member(_event):
mutating = False

layer.events.highlight.connect(add_unrelated_member)
with pytest.raises(RuntimeError, match="canonical linked result selection"):
with pytest.raises(RuntimeError, match="requested native data selection"):
server.result_selection_controller.select(layer, 1)


Expand Down Expand Up @@ -450,13 +509,14 @@ def test_napari_navigation_moves_to_selected_roi_component_slice(qtbot) -> None:
),
],
shape_type=["polygon", "polygon"],
features={"label": [11, 12], "area": [3.0, 4.0]},
features={"label": [11, 12], "area": [3.0, 4.0], "owner": [8, 8]},
ndim=4,
name="Result ROIs",
)
server, _overlay, _result_selection_dock, _qt_window = _viewer_server(
viewer,
layer,
group_binding=NapariResultSelectionGroupBinding("native-test-subject", "owner"),
)
projection = ViewerLayerAxisProjection(
projected_axis_components=("channel", "z"),
Expand Down Expand Up @@ -503,6 +563,9 @@ def test_napari_navigation_moves_to_selected_roi_component_slice(qtbot) -> None:
assert response["status"] == "success"
assert viewer.dims.current_step[:2] == (2, 3)
assert layer.selected_data == {1}
assert server.result_selection_controller._linked_group_members(layer, 1) == (
(layer, (0, 1)),
)

conflicting_response = NapariNavigationControlMessageAction().handle(
server,
Expand Down
Loading