Skip to content

funtracks.data_model.tracks_controller

Classes:

TracksController 

TracksController(tracks: SolutionTracks)

A set of high level functions to change the data model. All changes to the data should go through this API.

Methods:

  • add_edges –

    Add edges to the graph. Also update the track ids and

  • add_nodes –

    Calls the _add_nodes function to add nodes. Calls the refresh signal when finished.

  • delete_edges –

    Delete edges from the graph.

  • delete_nodes –

    Calls the _delete_nodes function and then emits the refresh signal

  • is_valid –

    Check if this edge is valid.

  • redo –

    Obtain the action to redo from the history

  • undo –

    Obtain the action to undo from the history, and invert.

  • update_node_attrs –

    Update the user provided node attributes (not the managed attributes).

  • update_segmentations –

    Handle a change in the segmentation mask, checking for node addition, deletion, and attribute updates.

Source code in src/funtracks/data_model/tracks_controller.py 
35
36
37
38
def __init__(self, tracks: SolutionTracks):
    self.tracks = tracks
    self.action_history = ActionHistory()
    self.node_id_counter = 1

add_edges 

add_edges(edges: Iterable[Edge]) -> None

Add edges to the graph. Also update the track ids and corresponding segmentations if applicable

Parameters:

  • edges

    (Iterable[Edge]) –

    An iterable of edges, each with source and target node ids

Source code in src/funtracks/data_model/tracks_controller.py 
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
def add_edges(self, edges: Iterable[Edge]) -> None:
    """Add edges to the graph. Also update the track ids and
    corresponding segmentations if applicable

    Args:
        edges (Iterable[Edge]): An iterable of edges, each with source and target
            node ids
    """
    make_valid_actions = []
    for edge in edges:
        is_valid, valid_action = self.is_valid(edge)
        if not is_valid:
            # warning was printed with details in is_valid call
            return
        if valid_action is not None:
            make_valid_actions.append(valid_action)
    main_action = self._add_edges(edges)
    action: TracksAction
    if len(make_valid_actions) > 0:
        make_valid_actions.append(main_action)
        action = ActionGroup(self.tracks, make_valid_actions)
    else:
        action = main_action
    self.action_history.add_new_action(action)
    self.tracks.refresh.emit()

add_nodes 

add_nodes(attributes: Attrs, pixels: list[SegMask] | None = None) -> None

Calls the _add_nodes function to add nodes. Calls the refresh signal when finished.

Parameters:

  • attributes

    (Attrs) –

    dictionary containing at least time and position attributes

  • pixels

    (list[SegMask] | None, default: None ) –

    The pixels associated with each node, if a segmentation is present. Defaults to None.

Source code in src/funtracks/data_model/tracks_controller.py 
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
def add_nodes(
    self,
    attributes: Attrs,
    pixels: list[SegMask] | None = None,
) -> None:
    """Calls the _add_nodes function to add nodes. Calls the refresh signal when finished.

    Args:
        attributes (Attrs): dictionary containing at least time and position attributes
        pixels (list[SegMask] | None, optional): The pixels associated with each node,
            if a segmentation is present. Defaults to None.
    """
    result = self._add_nodes(attributes, pixels)
    if result is not None:
        action, nodes = result
        self.action_history.add_new_action(action)
        self.tracks.refresh.emit(nodes[0] if nodes else None)

delete_edges 

delete_edges(edges: Iterable[Edge])

Delete edges from the graph.

Parameters:

  • edges

    (Iterable[Edge]) –

    The Nx2 array of edges to be deleted

Source code in src/funtracks/data_model/tracks_controller.py 
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
def delete_edges(self, edges: Iterable[Edge]):
    """Delete edges from the graph.

    Args:
        edges (Iterable[Edge]): The Nx2 array of edges to be deleted
    """

    for edge in edges:
        # First check if the to be deleted edges exist
        if not self.tracks.graph.has_edge(edge[0], edge[1]):
            show_warning("Cannot delete non-existing edge!")
            return
    action = self._delete_edges(edges)
    self.action_history.add_new_action(action)
    self.tracks.refresh.emit()

delete_nodes 

delete_nodes(nodes: Iterable[Node]) -> None

Calls the _delete_nodes function and then emits the refresh signal

Parameters:

  • nodes

    (Iterable[Node]) –

    array of node_ids to be deleted

Source code in src/funtracks/data_model/tracks_controller.py 
189
190
191
192
193
194
195
196
197
198
def delete_nodes(self, nodes: Iterable[Node]) -> None:
    """Calls the _delete_nodes function and then emits the refresh signal

    Args:
        nodes (Iterable[Node]): array of node_ids to be deleted
    """

    action = self._delete_nodes(nodes)
    self.action_history.add_new_action(action)
    self.tracks.refresh.emit()

is_valid 

is_valid(edge: Edge) -> tuple[bool, TracksAction | None]

Check if this edge is valid. Criteria: - not horizontal - not existing yet - no merges - no triple divisions - new edge should be the shortest possible connection between two nodes, given their track_ids. (no skipping/bypassing any nodes of the same track_id). Check if there are any nodes of the same source or target track_id between source and target

Parameters:

  • edge

    (ndarray[int, int]) –

    edge to be validated

Returns: True if the edge is valid, false if invalid

Source code in src/funtracks/data_model/tracks_controller.py 
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
def is_valid(self, edge: Edge) -> tuple[bool, TracksAction | None]:
    """Check if this edge is valid.
    Criteria:
    - not horizontal
    - not existing yet
    - no merges
    - no triple divisions
    - new edge should be the shortest possible connection between two nodes, given their track_ids.
    (no skipping/bypassing any nodes of the same track_id). Check if there are any nodes of the same source or target track_id between source and target

    Args:
        edge (np.ndarray[(int, int)]: edge to be validated
    Returns:
        True if the edge is valid, false if invalid"""

    # make sure that the node2 is downstream of node1
    time1 = self.tracks.get_time(edge[0])
    time2 = self.tracks.get_time(edge[1])

    if time1 > time2:
        edge = (edge[1], edge[0])
        time1, time2 = time2, time1
    action = None
    # do all checks
    # reject if edge already exists
    if self.tracks.graph.has_edge(edge[0], edge[1]):
        show_warning("Edge is rejected because it exists already.")
        return False, action

    # reject if edge is horizontal
    elif self.tracks.get_time(edge[0]) == self.tracks.get_time(edge[1]):
        show_warning("Edge is rejected because it is horizontal.")
        return False, action

    # reject if target node already has an incoming edge
    elif self.tracks.graph.in_degree(edge[1]) > 0:
        show_warning("Edge is rejected because merges are currently not allowed.")
        return False, action

    elif self.tracks.graph.out_degree(edge[0]) > 1:
        show_warning(
            "Edge is rejected because triple divisions are currently not allowed."
        )
        return False, action

    elif time2 - time1 > 1:
        track_id2 = self.tracks.graph.nodes[edge[1]][NodeAttr.TRACK_ID.value]
        # check whether there are already any nodes with the same track id between source and target (shortest path between equal track_ids rule)
        for t in range(time1 + 1, time2):
            nodes = [
                n
                for n, attr in self.tracks.graph.nodes(data=True)
                if attr.get(self.tracks.time_attr) == t
                and attr.get(NodeAttr.TRACK_ID.value) == track_id2
            ]
            if len(nodes) > 0:
                show_warning("Please connect to the closest node")
                return False, action

    # all checks passed!
    return True, action

redo 

redo() -> bool

Obtain the action to redo from the history Returns: bool: True if the action was re-done, False if there were no more actions

Source code in src/funtracks/data_model/tracks_controller.py 
542
543
544
545
546
547
548
549
550
551
def redo(self) -> bool:
    """Obtain the action to redo from the history
    Returns:
        bool: True if the action was re-done, False if there were no more actions
    """
    if self.action_history.redo():
        self.tracks.refresh.emit()
        return True
    else:
        return False

undo 

undo() -> bool

Obtain the action to undo from the history, and invert. Returns: bool: True if the action was undone, False if there were no more actions

Source code in src/funtracks/data_model/tracks_controller.py 
531
532
533
534
535
536
537
538
539
540
def undo(self) -> bool:
    """Obtain the action to undo from the history, and invert.
    Returns:
        bool: True if the action was undone, False if there were no more actions
    """
    if self.action_history.undo():
        self.tracks.refresh.emit()
        return True
    else:
        return False

update_node_attrs 

update_node_attrs(nodes: Iterable[Node], attributes: Attrs)

Update the user provided node attributes (not the managed attributes). Also adds the action to the history and emits the refresh signal.

Parameters:

  • nodes

    (Iterable[Node]) –

    The nodes to update the attributes for

  • attributes

    (Attrs) –

    A mapping from user-provided attributes to values for each node.

Source code in src/funtracks/data_model/tracks_controller.py 
310
311
312
313
314
315
316
317
318
319
320
321
def update_node_attrs(self, nodes: Iterable[Node], attributes: Attrs):
    """Update the user provided node attributes (not the managed attributes).
    Also adds the action to the history and emits the refresh signal.

    Args:
        nodes (Iterable[Node]): The nodes to update the attributes for
        attributes (Attrs): A mapping from user-provided attributes to values for
            each node.
    """
    action = self._update_node_attrs(nodes, attributes)
    self.action_history.add_new_action(action)
    self.tracks.refresh.emit()

update_segmentations 

update_segmentations(to_remove: list[tuple[Node, SegMask]], to_update_smaller: list[tuple[Node, SegMask]], to_update_bigger: list[tuple[Node, SegMask]], to_add: list[tuple[Node, int, SegMask]], current_timepoint: int) -> None

Handle a change in the segmentation mask, checking for node addition, deletion, and attribute updates. Args: updated_pixels (list[(tuple(np.ndarray, np.ndarray, np.ndarray), np.ndarray, int)]): list holding the operations that updated the segmentation (directly from the napari labels paint event). Each element in the list consists of a tuple of np.ndarrays representing indices for each dimension, an array of the previous values, and an array or integer representing the new value(s) current_timepoint (int): the current time point in the viewer, used to set the selected node.

Source code in src/funtracks/data_model/tracks_controller.py 
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
def update_segmentations(
    self,
    to_remove: list[tuple[Node, SegMask]],  # (node_ids, pixels)
    to_update_smaller: list[tuple[Node, SegMask]],  # (node_id, pixels)
    to_update_bigger: list[tuple[Node, SegMask]],  # (node_id, pixels)
    to_add: list[tuple[Node, int, SegMask]],  # (node_id, track_id, pixels)
    current_timepoint: int,
) -> None:
    """Handle a change in the segmentation mask, checking for node addition, deletion, and attribute updates.
    Args:
        updated_pixels (list[(tuple(np.ndarray, np.ndarray, np.ndarray), np.ndarray, int)]):
            list holding the operations that updated the segmentation (directly from
            the napari labels paint event).
            Each element in the list consists of a tuple of np.ndarrays representing
            indices for each dimension, an array of the previous values, and an array
            or integer representing the new value(s)
        current_timepoint (int): the current time point in the viewer, used to set the selected node.
    """
    actions: list[TracksAction] = []
    node_to_select = None

    if len(to_remove) > 0:
        nodes = [node_id for node_id, _ in to_remove]
        pixels = [pixels for _, pixels in to_remove]
        actions.append(self._delete_nodes(nodes, pixels=pixels))
    if len(to_update_smaller) > 0:
        nodes = [node_id for node_id, _ in to_update_smaller]
        pixels = [pixels for _, pixels in to_update_smaller]
        actions.append(self._update_node_segs(nodes, pixels, added=False))
    if len(to_update_bigger) > 0:
        nodes = [node_id for node_id, _ in to_update_bigger]
        pixels = [pixels for _, pixels in to_update_bigger]
        actions.append(self._update_node_segs(nodes, pixels, added=True))
    if len(to_add) > 0:
        nodes = [node for node, _, _ in to_add]
        pixels = [pix for _, _, pix in to_add]
        track_ids = [
            val if val is not None else self.tracks.get_next_track_id()
            for _, val, _ in to_add
        ]
        times = [pix[0][0] for pix in pixels]
        attributes = {
            NodeAttr.TRACK_ID.value: track_ids,
            NodeAttr.TIME.value: times,
            "node_id": nodes,
        }

        result = self._add_nodes(attributes=attributes, pixels=pixels)
        if result is None:
            return
        else:
            action, nodes = result

        actions.append(action)

        # if this is the time point where the user added a node, select the new node
        if current_timepoint in times:
            index = times.index(current_timepoint)
            node_to_select = nodes[index]

    action_group = ActionGroup(self.tracks, actions)
    self.action_history.add_new_action(action_group)
    self.tracks.refresh.emit(node_to_select)