test: add graphml output tests

Author: 404Simon

tests/_helpers.py

@@ -42,3 +42,40 @@ def read_feature_csv(path: Path) -> dict[str, float]:
 
 def hash_array(arr: np.ndarray) -> str:
     return hashlib.sha256(arr.tobytes()).hexdigest()
+
+
+def cross_image(size: int = 32) -> np.ndarray:
+    """Binary cross: one junction and four endpoints."""
+    img = np.zeros((size, size), dtype=np.uint8)
+    img[size // 2, size // 4 : 3 * size // 4] = 1
+    img[size // 4 : 3 * size // 4, size // 2] = 1
+    return img
+
+
+def loop_image(size: int = 20) -> np.ndarray:
+    """Binary rectangle ring: two junctions connected by parallel branches."""
+    img = np.zeros((size, size), dtype=np.uint8)
+    margin = size // 4
+    inner = size - margin
+    img[margin, margin:inner] = 1
+    img[inner - 1, margin:inner] = 1
+    img[margin:inner, margin] = 1
+    img[margin:inner, inner - 1] = 1
+    return img
+
+
+def cross_volume(size: int = 16) -> np.ndarray:
+    """Binary volume with two perpendicular lines crossing at the center."""
+    vol = np.zeros((size, size, size), dtype=np.uint8)
+    vol[size // 2, size // 2, :] = 1
+    vol[size // 2, :, size // 2] = 1
+    return vol
+
+
+def line_volume(shape: tuple[int, int, int], axis: int = 0) -> np.ndarray:
+    """Binary volume with a single straight line through the center along *axis*."""
+    vol = np.zeros(shape, dtype=np.uint8)
+    idx = [s // 2 for s in shape]
+    idx[axis] = slice(None)
+    vol[tuple(idx)] = 1
+    return vol

tests/conftest.py

@@ -1,7 +1,37 @@
+import numpy as np
+import pytest
+from skan import Skeleton, summarize
+
+from ._helpers import cross_image, cross_volume, loop_image
+
+
 def pytest_addoption(parser):
     parser.addoption(
         "--update-baseline",
         action="store_true",
         default=False,
         help="Regenerate and save baseline skeletons",
     )
+
+
+@pytest.fixture
+def cross_skel() -> np.ndarray:
+    return cross_image()
+
+
+@pytest.fixture
+def cross_graph(cross_skel):
+    graph = Skeleton(cross_skel)
+    return graph, summarize(graph, separator="-")
+
+
+@pytest.fixture
+def loop_graph():
+    graph = Skeleton(loop_image())
+    return graph, summarize(graph, separator="-")
+
+
+@pytest.fixture
+def cross_volume_graph():
+    graph = Skeleton(cross_volume())
+    return graph, summarize(graph, separator="-")

tests/test_config.py

@@ -66,6 +66,7 @@ def test_defaults(self):
         assert c.write_summary_csv is True
         assert c.write_branch_csv is False
         assert c.write_radius is False
+        assert c.write_graphml is False
 
     def test_round_trip_dict(self):
         original = OutputConfig(
@@ -74,6 +75,7 @@ def test_round_trip_dict(self):
             write_summary_csv=True,
             write_branch_csv=True,
             write_radius=True,
+            write_graphml=True,
         )
         restored = OutputConfig.from_dict(original.to_dict())
         assert restored == original

tests/test_graphml.py

@@ -0,0 +1,141 @@
+"""Tests for vesskel.graphml GraphML export."""
+
+import networkx as nx
+import numpy as np
+from skan import Skeleton
+
+from vesskel.graphml import build_networkx_graph, write_graphml
+
+
+def _junction_endpoint_count(graph: Skeleton) -> int:
+    """Number of skan nodes that are junctions or endpoints (degree != 2)."""
+    return int((graph.degrees != 2).sum())
+
+
+class TestBuildNetworkxGraph:
+    def test_nodes_are_junctions_or_endpoints(self, cross_graph):
+        graph, branch_data = cross_graph
+        G = build_networkx_graph(graph, branch_data)
+
+        assert isinstance(G, nx.MultiGraph)
+        assert G.number_of_nodes() == _junction_endpoint_count(graph)
+        assert G.number_of_edges() == len(branch_data)
+        assert all(int(graph.degrees[n]) != 2 for n in G.nodes())
+
+    def test_node_attributes(self, cross_graph):
+        graph, branch_data = cross_graph
+        G = build_networkx_graph(graph, branch_data)
+
+        node_id, data = next(iter(G.nodes(data=True)))
+        assert data["coord_0"] == int(graph.coordinates[node_id, 0])
+        assert data["coord_1"] == int(graph.coordinates[node_id, 1])
+        assert data["degree"] == int(graph.degrees[node_id])
+
+        endpoints = [data for _, data in G.nodes(data=True) if data["is_endpoint"]]
+        assert len(endpoints) >= 2
+
+    def test_node_radius_absent_without_radius_matrix(self, cross_graph):
+        graph, branch_data = cross_graph
+        G = build_networkx_graph(graph, branch_data)
+        assert all("radius" not in data for _, data in G.nodes(data=True))
+
+    def test_node_radius_sampled_when_provided(self, cross_graph, cross_skel):
+        graph, branch_data = cross_graph
+        radius_matrix = np.full(cross_skel.shape, 2.5, dtype=np.float64)
+        G = build_networkx_graph(graph, branch_data, radius_matrix=radius_matrix)
+        for node_id, data in G.nodes(data=True):
+            assert data["radius"] == 2.5
+            assert data["radius"] == float(
+                radius_matrix[tuple(graph.coordinates[node_id])]
+            )
+
+    def test_edge_attributes_and_exclusions(self, cross_graph):
+        graph, branch_data = cross_graph
+        G = build_networkx_graph(graph, branch_data)
+
+        _, _, data = next(iter(G.edges(data=True)))
+        assert "branch-distance" in data
+        assert "euclidean-distance" in data
+        assert "node-id-src" not in data
+        assert "node-id-dst" not in data
+        assert not any(k.startswith("coord-") for k in data)
+        assert not any(k.startswith("image-coord-") for k in data)
+
+    def test_parallel_branches_preserved(self, loop_graph):
+        graph, branch_data = loop_graph
+        pairs = list(zip(branch_data["node-id-src"], branch_data["node-id-dst"]))
+        assert len(pairs) > len(set(pairs))  # fixture has parallel branches
+
+        G = build_networkx_graph(graph, branch_data)
+        assert G.number_of_edges() == len(pairs)
+
+    def test_nan_attributes_dropped(self, cross_graph):
+        graph, branch_data = cross_graph
+        branch_data = branch_data.copy()
+        branch_data["tortuosity"] = np.nan
+        branch_data["straightness"] = np.inf
+        G = build_networkx_graph(graph, branch_data)
+
+        for _, _, data in G.edges(data=True):
+            assert "tortuosity" not in data
+            assert "straightness" not in data
+
+    def test_summary_features_attached_and_nan_dropped(self, cross_graph):
+        graph, branch_data = cross_graph
+        G = build_networkx_graph(
+            graph,
+            branch_data,
+            summary_features={"num_nodes": 5.0, "mean_tortuosity": np.nan},
+        )
+        assert G.graph["num_nodes"] == 5.0
+        assert "mean_tortuosity" not in G.graph
+
+    def test_3d_skeleton(self, cross_volume_graph):
+        graph, branch_data = cross_volume_graph
+        G = build_networkx_graph(graph, branch_data)
+
+        assert G.number_of_nodes() == _junction_endpoint_count(graph)
+        _, data = next(iter(G.nodes(data=True)))
+        assert "coord_2" in data
+
+
+class TestWriteGraphml:
+    def test_write_and_read_round_trip(self, tmp_path, cross_graph, cross_skel):
+        graph, branch_data = cross_graph
+        radius_matrix = np.full(cross_skel.shape, 3.0, dtype=np.float64)
+        path = tmp_path / "img_graph.graphml"
+        write_graphml(
+            graph,
+            branch_data,
+            path,
+            summary_features={"total_length": 42.0},
+            radius_matrix=radius_matrix,
+        )
+
+        G = nx.read_graphml(str(path), node_type=int)
+        assert G.number_of_nodes() == _junction_endpoint_count(graph)
+        assert G.number_of_edges() == len(branch_data)
+        assert G.graph["total_length"] == 42.0
+
+        node_id, data = next(iter(G.nodes(data=True)))
+        assert data["coord_0"] == int(graph.coordinates[node_id, 0])
+        assert data["radius"] == 3.0
+        assert "degree" in data
+        assert "branch-distance" in next(iter(G.edges(data=True)))[2]
+
+    def test_parallel_edges_round_trip_as_multigraph(self, tmp_path, loop_graph):
+        graph, branch_data = loop_graph
+        path = tmp_path / "loop_graph.graphml"
+        write_graphml(graph, branch_data, path)
+
+        G = nx.read_graphml(str(path), node_type=int)
+        assert isinstance(G, nx.MultiGraph)
+        assert G.number_of_edges() == len(branch_data)
+
+    def test_empty_branch_data(self, tmp_path, cross_graph):
+        graph, branch_data = cross_graph
+        path = tmp_path / "empty_graph.graphml"
+        write_graphml(graph, branch_data.iloc[0:0], path)
+        G = nx.read_graphml(str(path), node_type=int)
+        assert G.number_of_nodes() == 0
+        assert G.number_of_edges() == 0

tests/test_io.py

@@ -4,6 +4,7 @@
 
 import numpy as np
 import pytest
+from skan import Skeleton, summarize
 
 from vesskel._io import save_analysis_outputs
 from vesskel.config import OutputConfig
@@ -220,3 +221,35 @@ def test_base_name_with_spaces(self, tmp_path):
         d = tmp_path / "my img"
         assert d.is_dir()
         assert (d / "my img_skeleton.npy").exists()
+
+    # -- graphml -----------------------------------------------------------
+
+    @pytest.fixture
+    def graph_result(self, cross_skel) -> AnalysisResult:
+        graph = Skeleton(cross_skel)
+        branch_data = summarize(graph, separator="-")
+        return AnalysisResult(
+            skeleton=cross_skel,
+            layers=[],
+            summary_features={"num_nodes": float(len(branch_data))},
+            branch_records=[],
+            node_records=[],
+            graph=graph,
+            branch_data=branch_data,
+            radius_matrix=np.ones_like(cross_skel, dtype=np.float64) * 2.0,
+        )
+
+    def test_saves_graphml(self, tmp_path, graph_result):
+        cfg = OutputConfig(write_graphml=True)
+        save_analysis_outputs(tmp_path, "img", graph_result, cfg)
+        assert (tmp_path / "img" / "img_graph.graphml").exists()
+
+    def test_skips_graphml_when_disabled(self, tmp_path, graph_result):
+        cfg = OutputConfig(write_graphml=False)
+        save_analysis_outputs(tmp_path, "img", graph_result, cfg)
+        assert not (tmp_path / "img" / "img_graph.graphml").exists()
+
+    def test_skips_graphml_when_graph_missing(self, tmp_path):
+        cfg = OutputConfig(write_graphml=True)
+        save_analysis_outputs(tmp_path, "img", self._result(), cfg)
+        assert not (tmp_path / "img" / "img_graph.graphml").exists()