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
50 changes: 50 additions & 0 deletions ultrack/core/linking/_test/test_link_processing.py
Original file line number Diff line number Diff line change
@@ -1,3 +1,4 @@
from types import SimpleNamespace
from typing import Tuple

import numpy as np
Expand All @@ -7,7 +8,9 @@

from ultrack import link
from ultrack.config import MainConfig
from ultrack.config.config import LinkingConfig
from ultrack.core.database import LinkDB, NodeDB
from ultrack.core.linking.processing import compute_spatial_neighbors


@pytest.mark.parametrize(
Expand Down Expand Up @@ -63,3 +66,50 @@ def test_multiprocess_link(
last_t = nodes["t"].max()
assert np.all(nodes[nodes["t"] != last_t].index.isin(edges["source_id"]))
assert np.all(nodes[nodes["t"] != 0].index.isin(edges["target_id"]))


def _stub_node(z: int, y: int, x: int) -> SimpleNamespace:
"""Lightweight Node stand-in exposing just the `centroid` attribute.

`compute_spatial_neighbors` only reads `n.centroid` on the path exercised
here (it returns early before any other Node API is touched), so a real
`Node` — which requires a parent and database setup — is not needed.
"""
return SimpleNamespace(centroid=np.asarray([z, y, x], dtype=np.float32))


@pytest.mark.parametrize(
"empty_side, expected_msg",
[
("source", "No segments found at source t=3"),
("target", "No segments found at target t=4"),
("both", "No segments found at source t=3"),
],
)
def test_compute_spatial_neighbors_empty_frame_warns(
empty_side: str, expected_msg: str, tmp_path
) -> None:
"""Regression test for issue #274.

`compute_spatial_neighbors` used to raise `IndexError: tuple index out of
range` when either source or target nodes were empty, because
`np.asarray([]).shape[1]` is undefined. It should now warn (pointing at the
offending t-frame) and skip linking instead of crashing.
"""
nodes = [_stub_node(0, 0, 0), _stub_node(1, 1, 1)]
source_nodes = [] if empty_side in ("source", "both") else nodes
target_nodes = [] if empty_side in ("target", "both") else nodes
target_shift = np.zeros((len(target_nodes), 3), dtype=np.float32)

with pytest.warns(RuntimeWarning, match=expected_msg):
compute_spatial_neighbors(
time=3,
config=LinkingConfig(),
source_nodes=source_nodes,
target_nodes=target_nodes,
target_shift=target_shift,
scale=None,
table_name=LinkDB.__tablename__,
db_path=f"sqlite:///{tmp_path / 'unused.db'}",
images=[],
)
22 changes: 22 additions & 0 deletions ultrack/core/linking/processing.py
Original file line number Diff line number Diff line change
@@ -1,4 +1,5 @@
import logging
import warnings
from contextlib import nullcontext
from typing import Callable, List, Optional, Sequence

Expand Down Expand Up @@ -186,6 +187,27 @@ def compute_spatial_neighbors(
source_pos = np.asarray([n.centroid for n in source_nodes])
target_pos = np.asarray([n.centroid for n in target_nodes], dtype=np.float32)

# Guard against empty source/target frames (issue #274). `np.asarray([])`
# returns a 1D array of shape (0,), so accessing `shape[1]` would raise
# IndexError. Warn the user pinpointing which side and t-frame is empty
# so they can inspect their segmentation, then skip linking for this pair.
if source_pos.ndim < 2:
warnings.warn(
f"No segments found at source t={time}; "
f"skipping linking to t={time + 1}.",
RuntimeWarning,
stacklevel=2,
)
return
if target_pos.ndim < 2:
warnings.warn(
f"No segments found at target t={time + 1}; "
f"skipping linking from t={time}.",
RuntimeWarning,
stacklevel=2,
)
return

n_dim = target_pos.shape[1]
target_shift = target_shift[:, -n_dim:] # matching positions dimensions
target_pos += target_shift
Expand Down
Loading