motile_plugin.data_model

Submodules

• motile_plugin.data_model.action_history
• motile_plugin.data_model.actions
• motile_plugin.data_model.node_type
• motile_plugin.data_model.solution_tracks
• motile_plugin.data_model.tracks
• motile_plugin.data_model.tracks_controller

Classes

Tracks	A set of tracks consisting of a graph and an optional segmentation.
SolutionTracks	Difference from Tracks: every node must have a track_id
NodeType	Types of nodes in the track graph. Currently used for standardizing
TracksController	A set of high level functions to change the data model.

Package Contents

class motile_plugin.data_model.Tracks(graph: networkx.DiGraph, segmentation: numpy.ndarray | None = None, time_attr: str = NodeAttr.TIME.value, pos_attr: str | tuple[str] | list[str] = NodeAttr.POS.value, scale: list[float] | None = None, ndim: int | None = None)

A set of tracks consisting of a graph and an optional segmentation. The graph nodes represent detections and must have a time attribute and position attribute. Edges in the graph represent links across time.

Attributes:

graph (nx.DiGraph): A graph with nodes representing detections and

and edges representing links across time. Assumed to be “valid” tracks (e.g., this is not supposed to be a candidate graph), but the structure is not verified.

segmentation (Optional(np.ndarray)): An optional segmentation that

accompanies the tracking graph. If a segmentation is provided, it is assumed that the graph has an attribute (default “seg_id”) holding the segmentation id. Defaults to None.

time_attr (str): The attribute in the graph that specifies the time

frame each node is in.

pos_attr (str | tuple[str] | list[str]): The attribute in the graph

that specifies the position of each node. Can be a single attribute that holds a list, or a list of attribute keys.

For bulk operations on attributes, a KeyError will be raised if a node or edge in the input set is not in the graph. All operations before the error node will be performed, and those after will not.

refresh

GRAPH_FILE = 'graph.json'

SEG_FILE = 'seg.npy'

ATTRS_FILE = 'attrs.json'

graph

segmentation

time_attr

pos_attr

scale

ndim

get_positions(nodes: collections.abc.Iterable[Node], incl_time: bool = False) → numpy.ndarray

Get the positions of nodes in the graph. Optionally include the time frame as the first dimension. Raises an error if any of the nodes are not in the graph.

Args:

node (Iterable[Node]): The node ids in the graph to get the positions of incl_time (bool, optional): If true, include the time as the

first element of each position array. Defaults to False.

Returns:

np.ndarray: A N x ndim numpy array holding the positions, where N is the

number of nodes passed in

get_position(node: Node, incl_time=False) → list

set_positions(nodes: collections.abc.Iterable[Node], positions: numpy.ndarray | collections.abc.Iterable[Edge], incl_time: bool = False)

Set the location of a node in the graph. Optionally include the time frame as the first dimension. Raises an error if the node is not in the graph.

Args:

node (Any): The node id in the graph to set the location of. location (np.ndarray): The location to set. If incl_time is true, time

is the first element.

incl_time (bool, optional): If true, include the time as the

first element of the location array. Defaults to False.

set_position(node: Node, position: list, incl_time=False)

get_times(nodes: collections.abc.Iterable[Node]) → collections.abc.Sequence[int]

get_time(node: Node) → int

Get the time frame of a given node. Raises an error if the node is not in the graph.

Args:

node (Any): The node id to get the time frame for

Returns:

int: The time frame that the node is in

set_times(nodes: collections.abc.Iterable[Node], times: collections.abc.Iterable[int])

set_time(node: Any, time: int)

Set the time frame of a given node. Raises an error if the node is not in the graph.

Args:

node (Any): The node id to set the time frame for time (int): The time to set

get_seg_ids(nodes: collections.abc.Iterable[Node], required=False) → collections.abc.Sequence[int | None]

get_seg_id(node: Any) → int | None

Get the segmentation id of a given node. Raises a KeyError if the node is not in the graph. Returns None if the node does not have an associated segmentation.

Args:

node (Any): The node id to get the seg id of

Returns:

int | None: The seg id of the node, or None if the node does not have a segmentation

set_seg_ids(nodes: collections.abc.Iterable[Node], seg_ids: collections.abc.Iterable[int])

Get the segmentation id of a given node. Raises a KeyError if the node is not in the graph.

Args:

node (Any): The node id to set the seg id of seg_id (int): The segmentation id to set for the node

set_seg_id(node: Node, seg_id: int)

add_nodes(nodes: collections.abc.Iterable[Node], times: collections.abc.Iterable[int], positions: numpy.ndarray | None = None, seg_ids: collections.abc.Iterable[int] | None = None, attrs: Attrs | None = None)

add_node(node: Node, time: int, position: collections.abc.Sequence | None = None, seg_id: int | None = None, attrs: Attrs | None = None)

Add a node to the graph. Will update the internal mappings and generate the segmentation-controlled attributes if there is a segmentation present. The segmentation should have been previously updated, otherwise the attributes will not update properly.

Args:

node (Node): The node id to add time (int): the time frame of the node to add position (Seqeunce | None): The spatial position of the node (excluding time).

Can be None if it should be automatically detected from the segmentation. Either seg_id or position must be provided. Defaults to None.

seg_id (int | None): The segmentation id of the node, used to match the node

to the segmentation at the given time, or None if the node does not have a matching segmentation. Either seg_id or position must be provided. Defaults to None.

remove_nodes(nodes: collections.abc.Iterable[Node])

remove_node(node: Node)

Remove the node from the graph and update the internal mappings. Does not update the segmentation if present.

Args:

node (Node): The node to remove from the graph

add_edges(edges: collections.abc.Iterable[Edge])

add_edge(edge: Edge)

remove_edges(edges: collections.abc.Iterable[Edge])

remove_edge(edge: Edge)

get_node(seg_id: int, time: int) → Node | None

Get the node with the given segmentation ID in the given time point. Useful for going from segmentation labels to graph nodes.

Args:

seg_id (int): The segmentation id of the node time (int): the time point of the node

Returns:

Node | None: the node id with the given seg_id in the given time, or None

if no such node exists.

get_areas(nodes: collections.abc.Iterable[Node]) → collections.abc.Sequence[int | None]

Get the area/volume of a given node. Raises a KeyError if the node is not in the graph. Returns None if the given node does not have an Area attribute.

Args:

node (Node): The node id to get the area/volume for

Returns:

int: The area/volume of the node

get_area(node: Node) → int | None

Get the area/volume of a given node. Raises a KeyError if the node is not in the graph. Returns None if the given node does not have an Area attribute.

Args:

node (Node): The node id to get the area/volume for

Returns:

int: The area/volume of the node

get_ious(edges: collections.abc.Iterable[Edge])

get_iou(edge: Edge)

get_pixels(nodes: list[Node]) → list[tuple[numpy.ndarray, Ellipsis]] | None

Get the pixels corresponding to each node in the nodes list.

Args:

nodes (list[Node]): A list of node to get the values for.

Returns:

list[tuple[np.ndarray, …]] | None: A list of tuples, where each tuple represents the pixels for one of the input nodes, or None if the segmentation is None. The tuple will have length equal to the number of segmentation dimensions, and can be used to index the segmentation.

set_pixels(pixels: collections.abc.Iterable[tuple[numpy.ndarray, Ellipsis]], values: collections.abc.Iterable[int | None])

Set the given pixels in the segmentation to the given value.

Args:

pixels (Iterable[tuple[np.ndarray]]): The pixels that should be set,

formatted like the output of np.nonzero (each element of the tuple represents one dimension, containing an array of indices in that dimension). Can be used to directly index the segmentation.

value (Iterable[int | None]): The value to set each pixel to

update_segmentations(nodes: collections.abc.Iterable[Node], pixels: collections.abc.Iterable[SegMask], added: bool = True) → None

Updates the segmentation of the given nodes. Also updates the auto-computed attributes of the nodes and incident edges.

_set_node_attributes(nodes: collections.abc.Iterable[Node], attributes: Attrs)

Update the attributes for given nodes

_set_edge_attributes(edges: collections.abc.Iterable[Edge], attributes: Attrs) → None

Set the edge attributes for the given edges. Attributes should already exist (although adding will work in current implementation, they cannot currently be removed)

Args:

edges (list[Edge]): A list of edges to set the attributes for attributes (Attributes): A dictionary of attribute name -> numpy array,

where the length of the arrays matches the number of edges. Attributes should already exist: this function will only update the values.

save(directory: pathlib.Path)

Save the tracks to the given directory. Currently, saves the graph as a json file in networkx node link data format, saves the segmentation as a numpy npz file, and saves the time and position attributes and scale information in an attributes json file.

Args:

directory (Path): The directory to save the tracks in.

_save_graph(directory: pathlib.Path)

Save the graph to file. Currently uses networkx node link data format (and saves it as json).

Args:

directory (Path): The directory in which to save the graph file.

_save_seg(directory: pathlib.Path)

Save a segmentation as a numpy array using np.save. In the future, could be changed to use zarr or other file types.

Args:

directory (Path): The directory in which to save the segmentation

_save_attrs(directory: pathlib.Path)

Save the time_attr, pos_attr, and scale in a json file in the given directory.

Args:

directory (Path): The directory in which to save the attributes

classmethod load(directory: pathlib.Path, seg_required=False) → Tracks

Load a Tracks object from the given directory. Looks for files in the format generated by Tracks.save.

Args:

directory (Path): The directory containing tracks to load seg_required (bool, optional): If true, raises a FileNotFoundError if the

segmentation file is not present in the directory. Defaults to False.

Returns:

Tracks: A tracks object loaded from the given directory

static _load_graph(graph_file: pathlib.Path) → networkx.DiGraph

Load the graph from the given json file. Expects networkx node_link_graph formatted json.

Args:

graph_file (Path): The json file to load into a networkx graph

Raises:

FileNotFoundError: If the file does not exist

Returns:

nx.DiGraph: A networkx graph loaded from the file.

static _load_seg(seg_file: pathlib.Path, seg_required: bool = False) → numpy.ndarray | None

Load a segmentation from a file. If the file doesn’t exist, either return None or raise a FileNotFoundError depending on the seg_required flag.

Args:

seg_file (Path): The npz file to load. seg_required (bool, optional): If true, raise a FileNotFoundError if the

segmentation is not present. Defaults to False.

Returns:

np.ndarray | None: The segmentation array, or None if it wasn’t present and

seg_required was False.

static _load_attrs(attrs_file: pathlib.Path) → dict

classmethod delete(directory: pathlib.Path)

_compute_ndim(seg: numpy.ndarray | None, scale: list[float] | None, provided_ndim: int | None)

_create_seg_time_to_node() → dict[int, dict[int, Node]]

Create a dictionary mapping seg_id -> dict(time_point -> node_id)

_set_node_attr(node: Node, attr: motile_toolbox.candidate_graph.NodeAttr, value: Any)

_set_nodes_attr(nodes: collections.abc.Iterable[Node], attr: str, values: collections.abc.Iterable[Any])

_get_node_attr(node: Node, attr: str, required: bool = False)

_get_nodes_attr(nodes: collections.abc.Iterable[Node], attr: str, required: bool = False)

_set_edge_attr(edge: Edge, attr: str, value: Any)

_set_edges_attr(edges: collections.abc.Iterable[Edge], attr: str, values: collections.abc.Iterable[Any])

_get_edge_attr(edge: Edge, attr: str, required: bool = False)

_get_edges_attr(edges: collections.abc.Iterable[Edge], attr: str, required: bool = False)

_remove_from_seg_time_to_node(nodes: collections.abc.Iterable[Node])

_add_to_seg_time_to_node(nodes: collections.abc.Iterable[Node])

_compute_node_attrs(seg_ids: collections.abc.Iterable[int | None], times: collections.abc.Iterable[int]) → Attrs

Get the segmentation controlled node attributes (area and position) from the segmentation with label seg_id in the given time point.

Args:

seg_id (int): The label id to query the current segmentation for time (int): The time frame of the current segmentation to query

Returns:

dict[str, int]: A dictionary containing the attributes that could be

determined from the segmentation. It will be empty if self.segmentation is None. If self.segmentation exists but seg_id is not present in time, area will be 0 and position will be None. If self.segmentation exists and seg_id is present in time, area and position will be included.

_compute_edge_attrs(edges: collections.abc.Iterable[Edge]) → Attrs

Get the segmentation controlled edge attributes (IOU) from the segmentations associated with the endpoints of the edge. The endpoints should already exist and have associated segmentations.

Args:

edge (Edge): The edge to compute the segmentation-based attributes from

Returns:

dict[str, int]: A dictionary containing the attributes that could be

determined from the segmentation. It will be empty if self.segmentation is None or if self.segmentation exists but the endpoint segmentations are not found.

class motile_plugin.data_model.SolutionTracks(graph: networkx.DiGraph, segmentation: numpy.ndarray | None = None, time_attr: str = NodeAttr.TIME.value, pos_attr: str | tuple[str] | list[str] = NodeAttr.POS.value, scale: list[float] | None = None, ndim: int | None = None)

Bases: motile_plugin.data_model.tracks.Tracks

Difference from Tracks: every node must have a track_id

max_track_id: int

classmethod from_tracks(tracks: motile_plugin.data_model.tracks.Tracks)

property node_id_to_track_id: dict[motile_plugin.data_model.tracks.Node, int]

get_next_track_id() → int

Return the next available track_id and update self.max_track_id

get_track_id(node) → int

set_track_id(node: motile_plugin.data_model.tracks.Node, value: int)

_initialize_track_ids()

_assign_tracklet_ids()

Add a track_id attribute to a graph by removing division edges, assigning one id to each connected component. Also sets the max_track_id and initializes a dictionary from track_id to nodes

export_tracks(outfile: pathlib.Path | str)

Export the tracks from this run to a csv with the following columns: t,[z],y,x,id,parent_id,track_id Cells without a parent_id will have an empty string for the parent_id. Whether or not to include z is inferred from self.ndim

class motile_plugin.data_model.NodeType

Bases: enum.Enum

Types of nodes in the track graph. Currently used for standardizing visualization. All nodes are exactly one type.

SPLIT = 'SPLIT'

END = 'END'

CONTINUE = 'CONTINUE'

class motile_plugin.data_model.TracksController(tracks: motile_plugin.data_model.solution_tracks.SolutionTracks)

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

tracks

action_history

node_id_counter = 1

add_nodes(attributes: motile_plugin.data_model.tracks.Attrs, pixels: list[motile_plugin.data_model.tracks.SegMask] | None = None) → None

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

Args:

nodes (np.ndarray[int]):an array of node ids attributes (dict[str, np.ndarray]): dictionary containing at least time and position attributes

_get_pred_and_succ(track_id: int, time: int) → tuple[motile_plugin.data_model.tracks.Node | None, motile_plugin.data_model.tracks.Node | None]

Get the last node with the given track id before time, and the first node with the track id after time, if any. Does not assume that a node with the given track_id and time is already in tracks, but it can be.

Args:

track_id (int): The track id to search for time (int): The time point to find the immediate predecessor and successor

for

Returns:

tuple[Node | None, Node | None]: The last node before time with the given track id, and the first node after time with the given track id, or Nones if there are no such nodes.

_add_nodes(attributes: motile_plugin.data_model.tracks.Attrs, pixels: list[motile_plugin.data_model.tracks.SegMask] | None = None) → tuple[motile_plugin.data_model.actions.TracksAction, list[motile_plugin.data_model.tracks.Node]]

Add nodes to the graph. Includes all attributes and the segmentation. Will return the actions needed to add the nodes, and the node ids generated for the new nodes. If there is a segmentation, the attributes must include: - time - seg_id - track_id If there is not a segmentation, the attributes must include: - time - pos - track_id

Logic of the function: - remove edges (when we add a node in a track between two nodes

connected by a skip edge)

• add the nodes

• add edges (to connect each node to its immediate

  predecessor and successor with the same track_id, if any)

Args:

nodes (list[Node]): a list of node ids attributes (Attributes): dictionary containing at least time and track id,

and either seg_id (if pixels are provided) or position (if not)

pixels (list[SegMask] | None): A list of pixels associated with the node,

or None if there is no segmentation. These pixels will be updated in the tracks.segmentation, set to the provided seg_id

delete_nodes(nodes: collections.abc.Iterable[Any]) → None

Calls the _delete_nodes function and then emits the refresh signal

Args:

nodes (np.ndarray): array of node_ids to be deleted

_delete_nodes(nodes: numpy.ndarray[Any], pixels: list[motile_plugin.data_model.tracks.SegMask] | None = None) → motile_plugin.data_model.actions.TracksAction

Delete the nodes provided by the array from the graph but maintain successor track_ids. Reconnect to the nearest predecessor and/or nearest successor on the same track, if any.

Function logic: - delete all edges incident to the nodes - delete the nodes - add edges to preds and succs of nodes if they have the same track id - update track ids if we removed a division by deleting the dge

Args:

nodes (np.ndarray): array of node_ids to be deleted

update_node_segs(nodes: collections.abc.Iterable[motile_plugin.data_model.tracks.Node], attributes: dict[str, numpy.ndarray]) → None

Calls the _update_node_segs function to update the node attributtes in given array. Then calls the refresh signal.

Args: nodes (np.ndarray[int]):an array of node ids attributes (dict[str, np.ndarray]): dictionary containing the attributes to be updated

update_node_attrs(nodes: collections.abc.Iterable[motile_plugin.data_model.tracks.Node], attributes: motile_plugin.data_model.tracks.Attrs)

_update_node_attrs(nodes: collections.abc.Iterable[motile_plugin.data_model.tracks.Node], attributes: motile_plugin.data_model.tracks.Attrs)

_update_node_segs(nodes: numpy.ndarray[Any], pixels: list[motile_plugin.data_model.tracks.SegMask], added=False) → motile_plugin.data_model.actions.TracksAction

Update the segmentation and segmentation-managed attributes for a set of nodes.

Args: nodes (np.ndarray[int]):an array of node ids attributes (dict[str, np.ndarray]): dictionary containing the attributes to be updated

add_edges(edges: numpy.ndarray[int]) → None

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

Args:

edges (np.array[int]): An Nx2 array of N edges, each with source and target

node ids

_add_edges(edges: numpy.ndarray[int]) → motile_plugin.data_model.actions.TracksAction

Add edges and attributes to the graph. Also update the track ids and corresponding segmentations of the target node tracks and potentially sibling tracks.

Args:

edges (np.array[int]): An Nx2 array of N edges, each with source and target

node ids

attributes (dict[str, np.ndarray]): dictionary mapping attribute names to

an array of values, where the index in the array matches the edge index

Returns:

True if the edges were successfully added, False if any edge was invalid.

is_valid(edge) → tuple[bool, motile_plugin.data_model.actions.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

delete_edges(edges: numpy.ndarray)

Delete edges from the graph.

Args:

edges (np.ndarray): _description_

_delete_edges(edges: numpy.ndarray) → motile_plugin.data_model.actions.ActionGroup

update_segmentations(to_remove: list[motile_plugin.data_model.tracks.Node], to_update_smaller: list[tuple], to_update_bigger: list[tuple], to_add: list[tuple], 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.

undo() → None

Obtain the action to undo from the history, and invert

redo() → None

Obtain the action to redo from the history

_get_new_node_ids(n: int) → list[motile_plugin.data_model.tracks.Node]

Get a list of new node ids for creating new nodes. They will be unique from all existing nodes, but have no other guarantees.

Args:

n (int): The number of new node ids to return

Returns:

list[Node]: A list of new node ids.