feat: optional graphml output

Author: 404Simon

vesskel/_io.py

@@ -11,6 +11,7 @@
 from PIL import Image
 
 from vesskel.config import OutputConfig
+from vesskel.graphml import write_graphml
 from vesskel.pipeline import AnalysisResult
 
 
@@ -114,6 +115,19 @@ def save_analysis_outputs(
     if config.write_radius and result.radius_matrix is not None:
         save_radius(out / f"{base_name}_radius", result.radius_matrix)
 
+    if (
+        config.write_graphml
+        and result.graph is not None
+        and result.branch_data is not None
+    ):
+        write_graphml(
+            result.graph,
+            result.branch_data,
+            out / f"{base_name}_graph.graphml",
+            summary_features=result.summary_features,
+            radius_matrix=result.radius_matrix,
+        )
+
     if config.write_branch_csv and result.branch_records:
         write_csv(out / f"{base_name}_branches.csv", result.branch_records)
 

vesskel/_napari.py

@@ -134,6 +134,7 @@ def _output_params(
             write_branch_csv: bool = False,
             write_node_csv: bool = False,
             write_radius: bool = False,
+            write_graphml: bool = False,
         ) -> None:
             return None
 
@@ -320,12 +321,16 @@ def _on_nodes_toggle_write_node_csv(*args) -> None:
 
         self.extract_nodes_widget.changed.connect(_on_nodes_toggle_write_node_csv)
 
+        self.write_graphml_widget = output_gui.write_graphml
+        self.write_graphml_widget.label = "Write graph (.graphml)"
+
         output_group.append(self.write_skeleton_npy_widget)
         output_group.append(self.write_skeleton_png_widget)
         output_group.append(self.write_summary_csv_widget)
         output_group.append(self.write_branch_csv_widget)
         output_group.append(self.write_node_csv_widget)
         output_group.append(self.write_radius_widget)
+        output_group.append(self.write_graphml_widget)
 
         # ============================================================
         # Output Directory
@@ -400,6 +405,7 @@ def _get_current_pipeline_config(self) -> PipelineConfig:
                 write_branch_csv=self.write_branch_csv_widget.value,
                 write_node_csv=self.write_node_csv_widget.value,
                 write_radius=self.write_radius_widget.value,
+                write_graphml=self.write_graphml_widget.value,
             ),
         )
 
@@ -426,6 +432,7 @@ def _set_pipeline_config(self, config: PipelineConfig) -> None:
         self.write_branch_csv_widget.value = o.write_branch_csv
         self.write_node_csv_widget.value = o.write_node_csv
         self.write_radius_widget.value = o.write_radius
+        self.write_graphml_widget.value = o.write_graphml
 
     # ------------------------------------------------------------------
     # Actions

vesskel/config.py

@@ -90,6 +90,7 @@ class OutputConfig:
     write_branch_csv: bool = False
     write_node_csv: bool = False
     write_radius: bool = False
+    write_graphml: bool = False
 
     def to_dict(self) -> dict[str, Any]:
         return {
@@ -99,6 +100,7 @@ def to_dict(self) -> dict[str, Any]:
             "write_branch_csv": self.write_branch_csv,
             "write_node_csv": self.write_node_csv,
             "write_radius": self.write_radius,
+            "write_graphml": self.write_graphml,
         }
 
     @classmethod

vesskel/graphml.py

@@ -0,0 +1,127 @@
+"""GraphML export for vessel skeleton graphs."""
+
+from __future__ import annotations
+
+from pathlib import Path
+from typing import TYPE_CHECKING
+
+import networkx as nx
+import numpy as np
+from skan.csr import skeleton_to_nx
+
+if TYPE_CHECKING:
+    from skan import Skeleton
+
+# Branch-table columns describing topology or node positions; excluded from
+# edge attributes (node positions live on the node elements instead).
+_EXCLUDED_COLUMNS = {
+    "node-id-src",
+    "node-id-dst",
+    "skeleton-id",
+}
+
+
+def _is_missing(value: object) -> bool:
+    """True for NaN/Inf floats, which GraphML cannot represent."""
+    if isinstance(value, (float, np.floating)):
+        return bool(np.isnan(value) or np.isinf(value))
+    return False
+
+
+def _edge_attributes(row: dict[str, object]) -> dict[str, object]:
+    return {
+        key: value
+        for key, value in row.items()
+        if not key.startswith("coord-")
+        and not key.startswith("image-coord-")
+        and key not in _EXCLUDED_COLUMNS
+        and not _is_missing(value)
+    }
+
+
+def build_networkx_graph(
+    graph: Skeleton,
+    branch_data,
+    summary_features: dict[str, float] | None = None,
+    radius_matrix: np.ndarray | None = None,
+) -> nx.MultiGraph:
+    """Convert a skan skeleton graph into a networkx MultiGraph.
+
+    Nodes carry coordinates, degree, node-type flags, and ``radius`` (when
+    *radius_matrix* is provided). Edges carry the per-branch features from
+    *branch_data*. NaN/Inf attributes are omitted. *summary_features* is
+    attached as graph-level attributes.
+
+    Parameters
+    ----------
+    graph : Skeleton
+        Pre-built skan Skeleton graph (e.g. from ``build_vessel_graph``).
+    branch_data : DataFrame
+        Pre-computed branch summary (e.g. from ``skan.summarize``).
+    summary_features : dict[str, float], optional
+        Per-image summary features attached as graph-level attributes.
+    radius_matrix : ndarray, optional
+        EDT radius array from ``compute_radii``; radius at each node is
+        attached as a ``radius`` node attribute when provided.
+    """
+    G = skeleton_to_nx(graph, branch_data)
+    G.graph.clear()
+
+    for summary in (summary_features or {}).items():
+        if not _is_missing(summary[1]):
+            G.graph[summary[0]] = summary[1]
+
+    for node_id in G.nodes():
+        coords = graph.coordinates[node_id]
+        degree = int(graph.degrees[node_id])
+        attrs = {f"coord_{d}": int(c) for d, c in enumerate(coords)}
+        attrs["degree"] = degree
+        attrs["is_endpoint"] = degree == 1
+        attrs["is_junction"] = degree >= 3
+        attrs["is_pass_through"] = degree == 2
+        if radius_matrix is not None:
+            attrs["radius"] = float(radius_matrix[tuple(coords)])
+        G.nodes[node_id].update(attrs)
+
+    G.clear_edges()
+    for row in branch_data.to_dict(orient="records"):
+        G.add_edge(
+            int(row["node-id-src"]),
+            int(row["node-id-dst"]),
+            **_edge_attributes(row),
+        )
+
+    return G
+
+
+def write_graphml(
+    graph: Skeleton,
+    branch_data,
+    path: str | Path,
+    *,
+    summary_features: dict[str, float] | None = None,
+    radius_matrix: np.ndarray | None = None,
+) -> None:
+    """Write a skeleton graph to a GraphML file.
+
+    Parameters
+    ----------
+    graph : Skeleton
+        Pre-built skan Skeleton graph (e.g. from ``build_vessel_graph``).
+    branch_data : DataFrame
+        Pre-computed branch summary (e.g. from ``skan.summarize``).
+    path : str or Path
+        Destination file path.
+    summary_features : dict[str, float], optional
+        Per-image summary features attached as graph-level attributes.
+    radius_matrix : ndarray, optional
+        EDT radius array from ``compute_radii``; radius at each node is
+        attached as a ``radius`` node attribute when provided.
+    """
+    G = build_networkx_graph(
+        graph,
+        branch_data,
+        summary_features=summary_features,
+        radius_matrix=radius_matrix,
+    )
+    nx.write_graphml(G, str(path))

vesskel/pipeline.py

@@ -25,6 +25,8 @@
 
 if TYPE_CHECKING:
     from napari.types import LayerDataTuple  # noqa: F401
+    from pandas import DataFrame
+    from skan import Skeleton
 
 
 @dataclass
@@ -38,6 +40,8 @@ class AnalysisResult:
     node_records: list[dict[str, object]]
     radius_matrix: np.ndarray | None = None
     preprocessed_binary: np.ndarray | None = None
+    graph: Skeleton | None = None
+    branch_data: DataFrame | None = None
 
 
 def preprocess_binary(
@@ -212,4 +216,6 @@ def analyze_binary_image(
         node_records=node_records,
         radius_matrix=radius_matrix,
         preprocessed_binary=preprocessed_binary,
+        graph=graph,
+        branch_data=branch_data,
     )