epicure.geff_import

  1import os
  2
  3import geff
  4import networkx as nx
  5import numpy as np
  6from zarr.storage import StoreLike
  7
  8import epicure.Utils as ut
  9
 10
 11def _check_preconditions(graph: nx.Graph, metadata: geff.GeffMetadata | None) -> None:
 12    """
 13    Check that the graph meets the preconditions for import: directed graph and no merging cells.
 14
 15    Args:
 16        graph (nx.Graph): The GEFF graph to check.
 17        metadata (geff.GeffMetadata | None): The GEFF metadata.
 18    """
 19
 20    if metadata is None:
 21        if not isinstance(graph, nx.DiGraph):
 22            ut.show_error(f"The GEFF graph must be directed. Found type: {type(graph)}.")
 23    else:
 24        if not metadata.directed:
 25            ut.show_error(
 26                "The GEFF graph must be directed. Metadata indicates an undirected graph."
 27            )
 28
 29    assert isinstance(graph, nx.DiGraph), (
 30        f"The GEFF graph must be directed. Found type: {type(graph)}."
 31    )
 32    fusions = [n for n in graph.nodes() if graph.in_degree(n) > 1]
 33    if len(fusions) > 0:
 34        ut.show_warning(
 35            f"Merging cells detected (nodes: {fusions}). EpiCure behavior may be affected."
 36        )
 37
 38
 39def _identify_labels_path(metadata: geff.GeffMetadata) -> tuple[str | None, str | None]:
 40    """
 41    Identify the label key and path to the labels image from GEFF metadata.
 42
 43    Args:
 44        metadata (geff.GeffMetadata): The GEFF metadata.
 45
 46    Returns:
 47        tuple[str | None, str | None]: A tuple containing the label key
 48            and the path to the labels image if present, None otherwise.
 49    """
 50    if metadata.related_objects is None:
 51        return None, None
 52
 53    related_objects = [obj for obj in metadata.related_objects if obj.type == "labels"]
 54    if len(related_objects) == 0:
 55        return None, None
 56    elif len(related_objects) > 1:
 57        ut.show_warning(
 58            "Multiple related objects of type 'labels' found in GEFF metadata. "
 59            "Cannot determine which one to use. No labels image will be imported."
 60        )
 61        return None, None
 62    else:
 63        obj = related_objects[0]
 64        ut.show_debug(
 65            f"Labels image path identified from GEFF metadata: '{obj.path}'.",
 66        )
 67        if obj.label_prop is not None:
 68            ut.show_debug(
 69                f"Label property for labels image identified from GEFF metadata: '{obj.label_prop}'.",
 70            )
 71            return obj.label_prop, obj.path
 72        else:
 73            ut.show_warning(
 74                "Label property for labels image not specified in GEFF metadata. "
 75                "Falling back to 'label'.",
 76            )
 77            return "label", obj.path
 78
 79
 80def _identify_prop(
 81    geff_md: geff.GeffMetadata | None,
 82    geff_graph: nx.DiGraph,
 83    prop_name: str,
 84) -> str | None:
 85    """
 86    Identify the actual name of a property if it exists, given a property name to look for.
 87
 88    Args:
 89        geff_md (geff.GeffMetadata | None): The GEFF metadata.
 90        geff_graph (nx.DiGraph): The GEFF graph.
 91        prop_name (str): The name of the property to check.
 92
 93    Returns:
 94        str | None: The actual name of the property if present in all graph nodes, None otherwise.
 95    """
 96    if geff_md is not None and geff_md.node_props_metadata:
 97        for prop_id in geff_md.node_props_metadata.keys():
 98            if prop_id.lower() == prop_name.lower():
 99                if all(prop_id in geff_graph.nodes[node] for node in geff_graph.nodes):
100                    ut.show_debug(
101                        f"Property '{prop_id}' found in GEFF metadata and present in all graph nodes.",
102                    )
103                    return prop_id
104                else:
105                    ut.show_debug(
106                        f"Property '{prop_id}' found in GEFF metadata but not present in all graph nodes.",
107                    )
108
109    return None
110
111
112def _identify_time_axis(
113    geff_md: geff.GeffMetadata | None,
114    geff_graph: nx.DiGraph,
115) -> str:
116    """
117    Identify the time axis from GEFF metadata.
118
119    The function will try to infer it first from the display hints, then from the axes.
120    In case the function fallbacks to the GEFF axes, it will take time axes in order
121    and check if the corresponding key exists in the graph. The first one to be found
122    will be returned as the time axis.
123
124    Args:
125        geff_md (geff.GeffMetadata | None): The GEFF metadata.
126        geff_graph (nx.DiGraph): The GEFF graph.
127
128    Returns:
129        str: The identified time axis.
130    """
131    # Check for time in display hints.
132    time_key = None
133    hints = geff_md.display_hints if geff_md is not None else None
134    if hints is not None:
135        time_key = getattr(hints, "display_time", None)
136        if time_key is not None:
137            if all(time_key in geff_graph.nodes[node] for node in geff_graph.nodes):
138                ut.show_debug(
139                    f"Valid time axis inferred from GEFF display hints: '{time_key}'.",
140                )
141            else:
142                ut.show_debug(
143                    f"Time axis '{time_key}' inferred from GEFF display hints is not present "
144                    "in all the graph nodes. Falling back to GEFF axes to identify it.",
145                )
146                time_key = None
147
148    # Fallback to GEFF axes.
149    axes = geff_md.axes if geff_md is not None else None
150    if time_key is None and axes is not None:
151        time_axes = [axis for axis in axes if axis.type == "time"]
152        for axis in time_axes:
153            if axis.name is not None and all(
154                axis.name in geff_graph.nodes[node] for node in geff_graph.nodes
155            ):
156                time_key = axis.name
157                ut.show_debug(
158                    f"Valid time axis inferred from GEFF axes: '{time_key}'.",
159                )
160                break
161
162    if time_key is None:
163        ut.show_error(
164            "No valid time axis found. Please ensure that the GEFF file contains "
165            "a time axis or a time display hint."
166        )
167
168    return time_key
169
170
171def _identify_space_axes(
172    geff_md: geff.GeffMetadata | None,
173    geff_graph: nx.DiGraph,
174) -> tuple[str, str]:
175    """
176    Identify the space axes (x, y) from GEFF metadata.
177
178    The function will try to infer them from the GEFF metadata,
179    first from the display hints, then from the axes.
180
181    Args:
182        geff_md (geff.GeffMetadata | None): The GEFF metadata.
183        geff_graph (nx.DiGraph): The GEFF graph.
184
185    Returns:
186        tuple[str, str]: The identified space axes.
187    """
188    space_keys = [None, None]
189
190    # Check for space in display hints.
191    hints = geff_md.display_hints if geff_md is not None else None
192    hint_fields = ["display_horizontal", "display_vertical"]
193    for i, hint_field in enumerate(hint_fields):
194        if hints is not None:
195            space_key = getattr(hints, hint_field, None)
196            if space_key is not None:
197                if all(space_key in geff_graph.nodes[node] for node in geff_graph.nodes):
198                    ut.show_debug(
199                        f"Valid space axis inferred from GEFF display hints: '{space_key}'.",
200                    )
201                else:
202                    ut.show_debug(
203                        f"Space axis '{space_key}' inferred from GEFF display hints is not present "
204                        "in all the graph nodes. Falling back to GEFF axes to identify it.",
205                    )
206                    space_key = None
207            space_keys[i] = space_key
208
209    # Fallback to GEFF axes: space axes are consumed in order.
210    axes = geff_md.axes if geff_md is not None else None
211    if axes is not None:
212        space_axes = iter(axis for axis in axes if axis.type == "space")
213        for i, key in enumerate(space_keys):
214            if key is None:
215                for axis in space_axes:
216                    if axis.name is not None and all(
217                        axis.name in geff_graph.nodes[node] for node in geff_graph.nodes
218                    ):
219                        space_keys[i] = axis.name
220                        ut.show_debug(
221                            f"Valid space axis inferred from GEFF axes: '{space_keys[i]}'.",
222                        )
223                        break
224
225    if space_keys[0] is None or space_keys[1] is None:
226        ut.show_error(
227            "No valid space axes found. Please ensure that the GEFF file contains "
228            "space axes or space display hints."
229        )
230
231    return space_keys[0], space_keys[1]
232
233
234def _generate_label(geff_graph: nx.DiGraph) -> None:
235    """
236    Add a 'label' node attribute to each node in the graph.
237    Each linear path (unbranched segment) receives a unique label starting from 0.
238
239    Args:
240        geff_graph (nx.DiGraph): The graph to label. Modified in-place.
241
242    Modifies:
243        geff_graph (nx.DiGraph): Adds a 'label' attribute to each node.
244    """
245    labeled_nodes = set()
246    label_counter = 0
247
248    for start_node in geff_graph.nodes():
249        if start_node in labeled_nodes:
250            continue
251
252        # Start a new linear path.
253        current = start_node
254        path_nodes = []
255
256        # Follow the chain as long as we have a single successor
257        # with a single predecessor (linear continuation).
258        while current is not None and current not in labeled_nodes:
259            path_nodes.append(current)
260            labeled_nodes.add(current)
261
262            successors = list(geff_graph.successors(current))
263
264            if len(successors) == 1:
265                next_node = successors[0]
266                predecessors = list(geff_graph.predecessors(next_node))
267                # Continue only if next node has exactly one predecessor.
268                if len(predecessors) == 1:
269                    current = next_node
270                else:
271                    current = None  # branching point ahead
272            else:
273                current = None  # end of linear path
274
275        # Assign label to all nodes in this path.
276        for node in path_nodes:
277            geff_graph.nodes[node]["label"] = label_counter
278
279        label_counter += 1
280
281
282def _build_positions_array(
283    geff_graph: nx.DiGraph,
284    label_key: str,
285    time_key: str,
286    x_key: str,
287    y_key: str,
288) -> np.ndarray:
289    """
290    Build the positions array from the GEFF graph.
291
292    Args:
293        geff_graph (nx.DiGraph): The GEFF graph containing the nodes with their attributes.
294        label_key (str): The key for the label attribute in the graph nodes.
295        time_key (str): The key for the time/frame attribute in the graph nodes.
296        x_key (str): The key for the x coordinate attribute in the graph nodes.
297        y_key (str): The key for the y coordinate attribute in the graph nodes.
298
299    Returns:
300        np.ndarray: The filled positions array with columns [label, time, y, x].
301    """
302    positions = np.empty((len(geff_graph), 4), dtype=np.int32)
303    for i, node in enumerate(geff_graph.nodes()):
304        node_data = geff_graph.nodes[node]
305        positions[i, 0] = node_data[label_key]
306        positions[i, 1] = node_data[time_key]
307        positions[i, 2] = node_data[y_key]
308        positions[i, 3] = node_data[x_key]
309    # TODO Check it s ok not to sort position by time
310    return positions
311
312
313def _build_tracks_dict(geff_graph: nx.DiGraph, label_key: str) -> dict[int, list[int]]:
314    """
315    Build the tracks dictionary from the GEFF graph ({daughter_label: [mother_labels]}).
316
317    Args:
318        geff_graph (nx.DiGraph): The GEFF graph containing the nodes and edges.
319        label_key (str): The key for the label attribute in the graph nodes.
320
321    Returns:
322        dict[int, list[int]]: A dictionary mapping each daughter label to a list of its mother labels.
323    """
324    tracks: dict[int, list[int]] = {}
325    divisions = [n for n in geff_graph.nodes() if geff_graph.out_degree(n) > 1]
326    for div in divisions:
327        for daughter in geff_graph.successors(div):
328            mother_label = geff_graph.nodes[div][label_key]
329            daughter_label = geff_graph.nodes[daughter][label_key]
330            if daughter_label not in tracks:
331                tracks[daughter_label] = []
332            tracks[daughter_label].append(mother_label)
333    return tracks
334
335
336def _get_metadata(
337    metadata: geff.GeffMetadata, time_key: str, x_key: str, y_key: str
338) -> dict[str, str]:
339    """
340    Extract metadata from GEFF metadata object.
341
342    In EpiCure, time and space metadata are stored as UnitT, ScaleT, UnitXY, ScaleXY,
343    with UnitXY and UnitT being expressed in real world units. However, the data
344    in the GEFF graph AND in EpiCure are expressed in pixel and frame units.
345
346    Args:
347        metadata (geff.GeffMetadata): The GEFF metadata.
348        time_key (str): The key for the time/frame attribute.
349        x_key (str): The key for the x coordinate attribute.
350        y_key (str): The key for the y coordinate attribute.
351
352    Returns:
353        dict[str, str]: A dictionary of metadata key-value pairs.
354    """
355    md = {}
356    x_axis = None
357    y_axis = None
358    if metadata.axes is not None:
359        for axis in metadata.axes:
360            if axis.name == time_key:
361                md["UnitT"] = axis.scaled_unit
362                md["ScaleT"] = axis.scale
363            elif axis.name == x_key:
364                x_axis = axis
365            elif axis.name == y_key:
366                y_axis = axis
367
368    if x_axis is not None and y_axis is not None:
369        if x_axis.scaled_unit == y_axis.scaled_unit:
370            md["UnitXY"] = x_axis.scaled_unit
371        else:
372            ut.show_warning(
373                f"Different units for x and y axes: '{x_axis.scaled_unit}' and '{y_axis.scaled_unit}'. "
374                "UnitXY metadata will not be set."
375            )
376        if x_axis.scale == y_axis.scale:
377            md["ScaleXY"] = x_axis.scale
378        else:
379            ut.show_warning(
380                f"Different scales for x and y axes: '{x_axis.scale}' and '{y_axis.scale}'. "
381                "ScaleXY metadata will not be set."
382            )
383
384    return md
385
386
387def import_geff(
388    geff_path: StoreLike,
389) -> tuple[np.ndarray, dict[int, list[int]], dict[str, str], str | None]:
390    """
391    Import a GEFF file.
392
393    Args:
394        geff_path (StoreLike): The path to the GEFF file to import.
395
396    Returns:
397        tuple[np.ndarray, dict[int, list[int]], dict[str, str], str | None]: A tuple containing:
398            - A positions array with columns [label, time, y, x].
399            - A tracks dictionary mapping each daughter label to a list of its mother labels.
400            - A dictionary of metadata key-value pairs.
401            - The path to the labels image array if present, None otherwise.
402    """
403    geff_graph, geff_md = geff.read(geff_path, structure_validation=True)
404
405    _check_preconditions(geff_graph, geff_md)
406
407    if geff_md is not None:
408        label_key, labels_path = _identify_labels_path(geff_md)
409    else:
410        label_key, labels_path = None, None
411    # Even if we have a label key from the related objects, we need to check
412    # that it's actually present in the graph nodes.
413    label_key = _identify_prop(geff_md, geff_graph, label_key)
414    if label_key is None:
415        label_key = "label"
416        _generate_label(geff_graph)
417        ut.show_debug("Node labels generated and assigned to 'label'.")
418    else:
419        ut.show_debug(f"Identified label key: '{label_key}'.")
420
421    time_key = _identify_prop(geff_md, geff_graph, "frame")
422    if time_key is None:
423        ut.show_debug(
424            "No frame-like property identified in GEFF metadata. "
425            "Attempting to identify time axis from display hints or axes.",
426        )
427        time_key = _identify_time_axis(geff_md, geff_graph)
428
429    x_key = _identify_prop(geff_md, geff_graph, "x")
430    y_key = _identify_prop(geff_md, geff_graph, "y")
431    if x_key is None or y_key is None:
432        ut.show_debug(
433            "No x/y properties identified in GEFF metadata. "
434            "Attempting to identify space axes from display hints or axes.",
435        )
436    x_key, y_key = _identify_space_axes(geff_md, geff_graph)
437    ut.show_debug(f"Identified axes: '{time_key}', x: '{x_key}', y: '{y_key}'.")
438
439    positions = _build_positions_array(geff_graph, label_key, time_key, x_key, y_key)
440    tracks = _build_tracks_dict(geff_graph, label_key)
441
442    if geff_md is not None:
443        metadata = _get_metadata(geff_md, time_key, x_key, y_key)
444    else:
445        metadata = {}
446
447    ## labels path is a relative path (relative to GEFF). Convert it to absolute path
448    abs_path = os.path.join(geff_path, labels_path)
449    abs_path = os.path.abspath(abs_path)
450    return positions, tracks, metadata, abs_path