epicure.geff_export

View Source

  1import os
  2from typing import Dict, List
  3
  4import geff
  5import geff_spec
  6import networkx as nx
  7import numpy as np
  8import pandas as pd
  9from scipy.cluster.hierarchy import DisjointSet
 10
 11import epicure.Utils as ut
 12
 13
 14def create_label_to_track_mapping(
 15    divisions: Dict[int, List[int]], unique_labels: List[int]
 16) -> Dict[int, int]:
 17    """
 18    Create a mapping from labels to track IDs using scipy's DisjointSet for efficient track grouping.
 19
 20    Args:
 21        divisions: dict of {daughter_label: [mother_labels]} from epic.tracking.graph
 22        unique_labels: list of unique labels present in the tracking data
 23
 24    Returns:
 25        dict: {label: track_id} - mapping from each label to its track ID
 26    """
 27    if not divisions:
 28        # No divisions - each unique label is its own track.
 29        return {label: label for label in unique_labels}
 30
 31    ds = DisjointSet(unique_labels)
 32
 33    # Union connected labels based on mother-daughter relationships.
 34    for daughter, mothers in divisions.items():
 35        if daughter not in unique_labels:  # weirdly, this can happen
 36            continue
 37        for mother in mothers:
 38            if mother in unique_labels:
 39                ds.merge(daughter, mother)
 40
 41    # A connected component is a track. We use the root as track ID.
 42    # Create a mapping from label to track_id (root).
 43    label_to_track_id = {}
 44    for label in unique_labels:
 45        root = ds[label]
 46        label_to_track_id[label] = root
 47
 48    return label_to_track_id
 49
 50
 51def build_nodes_df(
 52    track_data: np.ndarray, divisions: Dict[int, List[int]]
 53) -> pd.DataFrame:
 54    """
 55    Build a DataFrame representing the nodes for the GEFF graph.
 56
 57    Args:
 58        track_data: numpy array with columns [label, frame, y, x]
 59        divisions: dict of {daughter_label: [mother_labels]} from epic.tracking.graph
 60
 61    Returns:
 62        pd.DataFrame with columns [node_id, label, frame, y, x, track_id]
 63    """
 64    df = pd.DataFrame(track_data, columns=["label", "frame", "y", "x"])
 65    df["node_id"] = df.index
 66
 67    # Generate and assign track IDs.
 68    labels = list(df["label"].unique())
 69    label_to_track_id = create_label_to_track_mapping(divisions, labels)
 70    df["track_id"] = df["label"].map(label_to_track_id)
 71
 72    return df
 73
 74
 75def build_edges_df(divisions: Dict[int, List[int]], df_nodes: pd.DataFrame):
 76    """
 77    Build a DataFrame representing the edges for the GEFF graph.
 78
 79    Args:
 80        divisions: dict of {daughter_label: [mother_labels]} from epic.tracking.graph
 81        df_nodes: DataFrame with node information
 82
 83    Returns:
 84        pd.DataFrame with columns [in_id, out_id]
 85    """
 86    if divisions is not None:
 87        for daughter, mothers in divisions.items():
 88            if len(mothers) > 1:
 89                ut.show_error(
 90                    f"Merge event detected. Label {daughter} "
 91                    f"has the following mother labels: {mothers}."
 92                )
 93    # TODO: does GEFF support merge events?
 94
 95    # Division edges: for each daughter-mother pair, create an edge.
 96    edges_data = [
 97        {"daughter": daughter, "mother": mother}
 98        for daughter, mothers in divisions.items()
 99        for mother in mothers
100    ]
101    df_edges = pd.DataFrame(edges_data)
102    # Labels stay the same until there is a division. But node IDs are unique.
103    # It means that in df_nodes, labels appears multiple times. Because of this
104    # we cannot easily map between df_nodes and df_edges. So we create intermediary
105    # columns to ease the mapping.
106    df_nodes["first_frame"] = df_nodes.groupby("label")["frame"].transform("min")
107    df_nodes["last_frame"] = df_nodes.groupby("label")["frame"].transform("max")
108    # A daughter is at the first frame of its label, a mother at the last frame of its label.
109    df_nodes["daughter"] = df_nodes["first_frame"] == df_nodes["frame"]
110    df_nodes["mother"] = df_nodes["last_frame"] == df_nodes["frame"]
111    df_nodes.drop(columns=["first_frame", "last_frame"], inplace=True)
112    # Now we can map between df_nodes and df_edges.
113    # The in_id is the node ID of the matching label that is a mother,
114    # and the out_id is the node ID of the matching label that is a daughter.
115    if "mother" in df_edges:
116        df_edges["in_id"] = df_edges["mother"].map(
117            df_nodes[df_nodes["mother"]].set_index("label")["node_id"]
118        )
119        df_edges["out_id"] = df_edges["daughter"].map(
120            df_nodes[df_nodes["daughter"]].set_index("label")["node_id"]
121        )
122        df_nodes.drop(columns=["daughter", "mother"], inplace=True)
123
124    # Non-division edges: for each label, connect consecutive nodes within that label.
125    non_division_edges = []
126    for label in df_nodes["label"].unique():
127        label_spots = df_nodes[df_nodes["label"] == label].sort_values("frame")
128        if len(label_spots) > 1:
129            for i in range(len(label_spots) - 1):
130                current_spot = label_spots.iloc[i]
131                next_spot = label_spots.iloc[i + 1]
132                non_division_edges.append(
133                    {"in_id": current_spot["node_id"], "out_id": next_spot["node_id"]}
134                )
135
136    # Combine division and non-division edges.
137    df_non_division_edges = pd.DataFrame(non_division_edges)
138    if not df_edges.empty and not df_non_division_edges.empty:
139        # Make sure both dataframes have the same columns.
140        df_edges = df_edges[["in_id", "out_id"]]
141        df_edges = pd.concat([df_edges, df_non_division_edges], ignore_index=True)
142    elif not df_non_division_edges.empty:
143        df_edges = df_non_division_edges
144
145    # Final cleanup and type conversion.
146    if not df_edges.empty:
147        # We can have NaN if a label has no mother (appears at first frame)
148        # or no daughter (disappears at last frame).
149        df_edges.dropna(inplace=True)
150        # Convert to int in case of NaN.
151        df_edges["in_id"] = df_edges["in_id"].astype(int)
152        df_edges["out_id"] = df_edges["out_id"].astype(int)
153
154    return df_edges
155
156
157def build_nx_digraph(epic) -> nx.DiGraph:
158    """
159    Build a NetworkX directed graph from EpiCure data.
160
161    Args:
162        epic: EpiCure instance with tracking data and graph
163
164    Returns:
165        nx.DiGraph: directed graph representing the cell lineages
166    """
167
168    df_nodes = build_nodes_df(epic.tracking.track_data, epic.tracking.graph)
169    df_edges = build_edges_df(epic.tracking.graph, df_nodes)
170
171    graph = nx.DiGraph()
172    for _, row in df_nodes.iterrows():
173        graph.add_node(row["node_id"], **row.to_dict())
174    for _, edge in df_edges.iterrows():
175        graph.add_edge(edge["in_id"], edge["out_id"])
176
177    node_attrs = {row["node_id"]: row.to_dict() for _, row in df_nodes.iterrows()}
178    nx.set_node_attributes(graph, node_attrs)
179
180    return graph
181
182
183def build_props_metadata() -> Dict[str, geff_spec.PropMetadata]:
184    """
185    Build GEFF properties metadata.
186
187    Returns:
188        dict: mapping from property names to their metadata
189    """
190    md_x = geff_spec.PropMetadata(
191        identifier="x",
192        dtype="int",
193        varlength=False,
194        unit="pixel",
195        name="x",
196        description="X coordinate of center of the cell",
197    )
198    md_y = geff_spec.PropMetadata(
199        identifier="y",
200        dtype="int",
201        varlength=False,
202        unit="pixel",
203        name="y",
204        description="Y coordinate of the center of the cell",
205    )
206    md_t = geff_spec.PropMetadata(
207        identifier="frame",
208        dtype="int32",
209        varlength=False,
210        unit="frame",
211        name="frame",
212        description="Time",
213    )
214    md_label = geff_spec.PropMetadata(
215        identifier="label",
216        dtype="int64",
217        varlength=False,
218        name="label",
219        description="Label of the cell",
220    )
221    md_nid = geff_spec.PropMetadata(
222        identifier="node_id",
223        dtype="int64",
224        varlength=False,
225        name="node_id",
226        description="Unique identifier of the node",
227    )
228
229    return {"x": md_x, "y": md_y, "frame": md_t, "label": md_label, "node_id": md_nid}
230
231
232def build_geff_metadata(epic):
233    """
234    Build GEFF metadata.
235
236    Args:
237        epic: EpiCure instance with metadata information
238    """
239    axes = [
240        geff_spec.Axis(
241            name="x",
242            type="space",
243            unit="pixel",
244            scale=epic.epi_metadata.get("ScaleXY", 1),
245            scaled_unit=epic.epi_metadata.get("UnitXY"),
246        ),
247        geff_spec.Axis(
248            name="y",
249            type="space",
250            unit="pixel",
251            scale=epic.epi_metadata.get("ScaleXY", 1),
252            scaled_unit=epic.epi_metadata.get("UnitXY"),
253        ),
254        geff_spec.Axis(
255            name="frame",
256            type="time",
257            unit="frame",
258            scale=epic.epi_metadata.get("ScaleT", 1),
259            scaled_unit=epic.epi_metadata.get("UnitT"),
260        ),
261    ]
262    display_hints = geff_spec.DisplayHint(
263        display_horizontal="x",
264        display_vertical="y",
265        display_time="frame",
266    )
267
268    return geff.GeffMetadata(
269        directed=True,
270        axes=axes,
271        display_hints=display_hints,
272        node_props_metadata=build_props_metadata(),
273        edge_props_metadata={},
274        track_node_props={"lineage": "track_id", "tracklet": "label"},
275        related_objects=[
276            geff_spec.RelatedObject(
277                type="labels",
278                path=os.path.join("..", epic.imgname + "_labels.tif"),
279                label_prop="label",
280            ),
281        ],
282    )
283
284
285def save_geff(epic, outname):
286    """
287    Save EpiCure tracking data as a GEFF file.
288
289    Args:
290        epic: EpiCure instance with tracking data and graph
291        outname: path to save the GEFF file
292    """
293    geff_graph = build_nx_digraph(epic)
294    geff_md = build_geff_metadata(epic)
295
296    geff.write(
297        geff_graph,
298        outname,
299        metadata=geff_md,
300        zarr_format=2,  # could be 3 but 2 by default in GEFF
301        structure_validation=True,
302        overwrite=True,
303    )

def create_label_to_track_mapping( divisions: Dict[int, List[int]], unique_labels: List[int]) -> Dict[int, int]: View Source

15def create_label_to_track_mapping(
16    divisions: Dict[int, List[int]], unique_labels: List[int]
17) -> Dict[int, int]:
18    """
19    Create a mapping from labels to track IDs using scipy's DisjointSet for efficient track grouping.
20
21    Args:
22        divisions: dict of {daughter_label: [mother_labels]} from epic.tracking.graph
23        unique_labels: list of unique labels present in the tracking data
24
25    Returns:
26        dict: {label: track_id} - mapping from each label to its track ID
27    """
28    if not divisions:
29        # No divisions - each unique label is its own track.
30        return {label: label for label in unique_labels}
31
32    ds = DisjointSet(unique_labels)
33
34    # Union connected labels based on mother-daughter relationships.
35    for daughter, mothers in divisions.items():
36        if daughter not in unique_labels:  # weirdly, this can happen
37            continue
38        for mother in mothers:
39            if mother in unique_labels:
40                ds.merge(daughter, mother)
41
42    # A connected component is a track. We use the root as track ID.
43    # Create a mapping from label to track_id (root).
44    label_to_track_id = {}
45    for label in unique_labels:
46        root = ds[label]
47        label_to_track_id[label] = root
48
49    return label_to_track_id

Create a mapping from labels to track IDs using scipy's DisjointSet for efficient track grouping.

Args: divisions: dict of {daughter_label: [mother_labels]} from epic.tracking.graph unique_labels: list of unique labels present in the tracking data

Returns: dict: {label: track_id} - mapping from each label to its track ID

def build_nodes_df( track_data: numpy.ndarray, divisions: Dict[int, List[int]]) -> pandas.core.frame.DataFrame: View Source

52def build_nodes_df(
53    track_data: np.ndarray, divisions: Dict[int, List[int]]
54) -> pd.DataFrame:
55    """
56    Build a DataFrame representing the nodes for the GEFF graph.
57
58    Args:
59        track_data: numpy array with columns [label, frame, y, x]
60        divisions: dict of {daughter_label: [mother_labels]} from epic.tracking.graph
61
62    Returns:
63        pd.DataFrame with columns [node_id, label, frame, y, x, track_id]
64    """
65    df = pd.DataFrame(track_data, columns=["label", "frame", "y", "x"])
66    df["node_id"] = df.index
67
68    # Generate and assign track IDs.
69    labels = list(df["label"].unique())
70    label_to_track_id = create_label_to_track_mapping(divisions, labels)
71    df["track_id"] = df["label"].map(label_to_track_id)
72
73    return df

Build a DataFrame representing the nodes for the GEFF graph.

Args: track_data: numpy array with columns [label, frame, y, x] divisions: dict of {daughter_label: [mother_labels]} from epic.tracking.graph

Returns: pd.DataFrame with columns [node_id, label, frame, y, x, track_id]

def build_edges_df( divisions: Dict[int, List[int]], df_nodes: pandas.core.frame.DataFrame): View Source

 76def build_edges_df(divisions: Dict[int, List[int]], df_nodes: pd.DataFrame):
 77    """
 78    Build a DataFrame representing the edges for the GEFF graph.
 79
 80    Args:
 81        divisions: dict of {daughter_label: [mother_labels]} from epic.tracking.graph
 82        df_nodes: DataFrame with node information
 83
 84    Returns:
 85        pd.DataFrame with columns [in_id, out_id]
 86    """
 87    if divisions is not None:
 88        for daughter, mothers in divisions.items():
 89            if len(mothers) > 1:
 90                ut.show_error(
 91                    f"Merge event detected. Label {daughter} "
 92                    f"has the following mother labels: {mothers}."
 93                )
 94    # TODO: does GEFF support merge events?
 95
 96    # Division edges: for each daughter-mother pair, create an edge.
 97    edges_data = [
 98        {"daughter": daughter, "mother": mother}
 99        for daughter, mothers in divisions.items()
100        for mother in mothers
101    ]
102    df_edges = pd.DataFrame(edges_data)
103    # Labels stay the same until there is a division. But node IDs are unique.
104    # It means that in df_nodes, labels appears multiple times. Because of this
105    # we cannot easily map between df_nodes and df_edges. So we create intermediary
106    # columns to ease the mapping.
107    df_nodes["first_frame"] = df_nodes.groupby("label")["frame"].transform("min")
108    df_nodes["last_frame"] = df_nodes.groupby("label")["frame"].transform("max")
109    # A daughter is at the first frame of its label, a mother at the last frame of its label.
110    df_nodes["daughter"] = df_nodes["first_frame"] == df_nodes["frame"]
111    df_nodes["mother"] = df_nodes["last_frame"] == df_nodes["frame"]
112    df_nodes.drop(columns=["first_frame", "last_frame"], inplace=True)
113    # Now we can map between df_nodes and df_edges.
114    # The in_id is the node ID of the matching label that is a mother,
115    # and the out_id is the node ID of the matching label that is a daughter.
116    if "mother" in df_edges:
117        df_edges["in_id"] = df_edges["mother"].map(
118            df_nodes[df_nodes["mother"]].set_index("label")["node_id"]
119        )
120        df_edges["out_id"] = df_edges["daughter"].map(
121            df_nodes[df_nodes["daughter"]].set_index("label")["node_id"]
122        )
123        df_nodes.drop(columns=["daughter", "mother"], inplace=True)
124
125    # Non-division edges: for each label, connect consecutive nodes within that label.
126    non_division_edges = []
127    for label in df_nodes["label"].unique():
128        label_spots = df_nodes[df_nodes["label"] == label].sort_values("frame")
129        if len(label_spots) > 1:
130            for i in range(len(label_spots) - 1):
131                current_spot = label_spots.iloc[i]
132                next_spot = label_spots.iloc[i + 1]
133                non_division_edges.append(
134                    {"in_id": current_spot["node_id"], "out_id": next_spot["node_id"]}
135                )
136
137    # Combine division and non-division edges.
138    df_non_division_edges = pd.DataFrame(non_division_edges)
139    if not df_edges.empty and not df_non_division_edges.empty:
140        # Make sure both dataframes have the same columns.
141        df_edges = df_edges[["in_id", "out_id"]]
142        df_edges = pd.concat([df_edges, df_non_division_edges], ignore_index=True)
143    elif not df_non_division_edges.empty:
144        df_edges = df_non_division_edges
145
146    # Final cleanup and type conversion.
147    if not df_edges.empty:
148        # We can have NaN if a label has no mother (appears at first frame)
149        # or no daughter (disappears at last frame).
150        df_edges.dropna(inplace=True)
151        # Convert to int in case of NaN.
152        df_edges["in_id"] = df_edges["in_id"].astype(int)
153        df_edges["out_id"] = df_edges["out_id"].astype(int)
154
155    return df_edges

Build a DataFrame representing the edges for the GEFF graph.

Args: divisions: dict of {daughter_label: [mother_labels]} from epic.tracking.graph df_nodes: DataFrame with node information

Returns: pd.DataFrame with columns [in_id, out_id]

def build_nx_digraph(epic) -> networkx.classes.digraph.DiGraph: View Source

158def build_nx_digraph(epic) -> nx.DiGraph:
159    """
160    Build a NetworkX directed graph from EpiCure data.
161
162    Args:
163        epic: EpiCure instance with tracking data and graph
164
165    Returns:
166        nx.DiGraph: directed graph representing the cell lineages
167    """
168
169    df_nodes = build_nodes_df(epic.tracking.track_data, epic.tracking.graph)
170    df_edges = build_edges_df(epic.tracking.graph, df_nodes)
171
172    graph = nx.DiGraph()
173    for _, row in df_nodes.iterrows():
174        graph.add_node(row["node_id"], **row.to_dict())
175    for _, edge in df_edges.iterrows():
176        graph.add_edge(edge["in_id"], edge["out_id"])
177
178    node_attrs = {row["node_id"]: row.to_dict() for _, row in df_nodes.iterrows()}
179    nx.set_node_attributes(graph, node_attrs)
180
181    return graph

Build a NetworkX directed graph from EpiCure data.

Args: epic: EpiCure instance with tracking data and graph

Returns: nx.DiGraph: directed graph representing the cell lineages

def build_props_metadata() -> Dict[str, geff_spec._prop_metadata.PropMetadata]: View Source

184def build_props_metadata() -> Dict[str, geff_spec.PropMetadata]:
185    """
186    Build GEFF properties metadata.
187
188    Returns:
189        dict: mapping from property names to their metadata
190    """
191    md_x = geff_spec.PropMetadata(
192        identifier="x",
193        dtype="int",
194        varlength=False,
195        unit="pixel",
196        name="x",
197        description="X coordinate of center of the cell",
198    )
199    md_y = geff_spec.PropMetadata(
200        identifier="y",
201        dtype="int",
202        varlength=False,
203        unit="pixel",
204        name="y",
205        description="Y coordinate of the center of the cell",
206    )
207    md_t = geff_spec.PropMetadata(
208        identifier="frame",
209        dtype="int32",
210        varlength=False,
211        unit="frame",
212        name="frame",
213        description="Time",
214    )
215    md_label = geff_spec.PropMetadata(
216        identifier="label",
217        dtype="int64",
218        varlength=False,
219        name="label",
220        description="Label of the cell",
221    )
222    md_nid = geff_spec.PropMetadata(
223        identifier="node_id",
224        dtype="int64",
225        varlength=False,
226        name="node_id",
227        description="Unique identifier of the node",
228    )
229
230    return {"x": md_x, "y": md_y, "frame": md_t, "label": md_label, "node_id": md_nid}

Build GEFF properties metadata.

Returns: dict: mapping from property names to their metadata

def build_geff_metadata(epic): View Source

233def build_geff_metadata(epic):
234    """
235    Build GEFF metadata.
236
237    Args:
238        epic: EpiCure instance with metadata information
239    """
240    axes = [
241        geff_spec.Axis(
242            name="x",
243            type="space",
244            unit="pixel",
245            scale=epic.epi_metadata.get("ScaleXY", 1),
246            scaled_unit=epic.epi_metadata.get("UnitXY"),
247        ),
248        geff_spec.Axis(
249            name="y",
250            type="space",
251            unit="pixel",
252            scale=epic.epi_metadata.get("ScaleXY", 1),
253            scaled_unit=epic.epi_metadata.get("UnitXY"),
254        ),
255        geff_spec.Axis(
256            name="frame",
257            type="time",
258            unit="frame",
259            scale=epic.epi_metadata.get("ScaleT", 1),
260            scaled_unit=epic.epi_metadata.get("UnitT"),
261        ),
262    ]
263    display_hints = geff_spec.DisplayHint(
264        display_horizontal="x",
265        display_vertical="y",
266        display_time="frame",
267    )
268
269    return geff.GeffMetadata(
270        directed=True,
271        axes=axes,
272        display_hints=display_hints,
273        node_props_metadata=build_props_metadata(),
274        edge_props_metadata={},
275        track_node_props={"lineage": "track_id", "tracklet": "label"},
276        related_objects=[
277            geff_spec.RelatedObject(
278                type="labels",
279                path=os.path.join("..", epic.imgname + "_labels.tif"),
280                label_prop="label",
281            ),
282        ],
283    )

Build GEFF metadata.

Args: epic: EpiCure instance with metadata information

def save_geff(epic, outname): View Source

286def save_geff(epic, outname):
287    """
288    Save EpiCure tracking data as a GEFF file.
289
290    Args:
291        epic: EpiCure instance with tracking data and graph
292        outname: path to save the GEFF file
293    """
294    geff_graph = build_nx_digraph(epic)
295    geff_md = build_geff_metadata(epic)
296
297    geff.write(
298        geff_graph,
299        outname,
300        metadata=geff_md,
301        zarr_format=2,  # could be 3 but 2 by default in GEFF
302        structure_validation=True,
303        overwrite=True,
304    )

Save EpiCure tracking data as a GEFF file.

Args: epic: EpiCure instance with tracking data and graph outname: path to save the GEFF file