Final improvements

take it or leave it ^^

Author: jsnel

pyglotaran_extras/inspect/kinetic_scheme/_k_matrix_parser.py

@@ -177,9 +177,13 @@ def extract_dataset_transitions(
 
     # Filter to only decay-type megacomplexes (silently skip non-decay).
     # Use the raw megacomplex object to avoid filling each item twice.
-    decay_megacomplexes = [
-        mc for mc in megacomplexes if hasattr(model.megacomplex[mc], "get_k_matrix")
-    ]
+    decay_megacomplexes: list[str] = []
+    for mc in megacomplexes:
+        if mc not in model.megacomplex:
+            msg = f"Megacomplex '{mc}' referenced by dataset '{dataset_name}' not found in model."
+            raise ValueError(msg)
+        if hasattr(model.megacomplex[mc], "get_k_matrix"):
+            decay_megacomplexes.append(mc)
 
     return extract_transitions(
         decay_megacomplexes,

pyglotaran_extras/inspect/kinetic_scheme/_layout.py

@@ -2,6 +2,7 @@
 
 from __future__ import annotations
 
+import hashlib
 from collections import deque
 from enum import Enum
 from typing import TYPE_CHECKING
@@ -401,11 +402,16 @@ def _build_in_degrees(
         In-degree for each compartment node.
     """
     in_degree: dict[str, int] = dict.fromkeys(compartment_labels, 0)
+    seen_edges: set[tuple[str, str]] = set()
     for edge in graph.edges:
+        edge_pair = (edge.source, edge.target)
+        if edge_pair in seen_edges:
+            continue
+        seen_edges.add(edge_pair)
         if (
             edge.source in compartment_labels
             and edge.target in compartment_labels
-            and (edge.source, edge.target) not in back_edges
+            and edge_pair not in back_edges
         ):
             in_degree[edge.target] += 1
     return in_degree
@@ -464,20 +470,27 @@ def _propagate_layers(
         Edges to skip.
     """
     processed: set[str] = set()
+    seen_edges: set[tuple[str, str]] = set()
     while queue:
         node = queue.popleft()
         if node in processed:
             continue
         processed.add(node)
 
         for neighbor in sorted(graph.successors(node)):
-            if neighbor not in compartment_labels or (node, neighbor) in back_edges:
+            edge_pair = (node, neighbor)
+            if (
+                neighbor not in compartment_labels
+                or edge_pair in back_edges
+                or edge_pair in seen_edges
+            ):
                 continue
+            seen_edges.add(edge_pair)
             new_layer = layers[node] + 1
             if neighbor not in layers or new_layer > layers[neighbor]:
                 layers[neighbor] = new_layer
-            in_degree[neighbor] -= 1
-            if in_degree[neighbor] <= 0:
+            in_degree[neighbor] = max(0, in_degree[neighbor] - 1)
+            if in_degree[neighbor] == 0:
                 queue.append(neighbor)
 
 
@@ -552,7 +565,22 @@ def sort_key(label: str) -> tuple[int, int, str]:
         else:
             barycenters[label] = 0.0
 
-    return sorted(nodes_in_layer, key=lambda n: (barycenters[n], n))
+    def barycenter_sort_key(label: str) -> tuple[float, str]:
+        """Return sort key based on barycenter and label.
+
+        Parameters
+        ----------
+        label : str
+            Node label.
+
+        Returns
+        -------
+        tuple[float, str]
+            Tuple of barycenter value and label for sorting.
+        """
+        return barycenters[label], label
+
+    return sorted(nodes_in_layer, key=barycenter_sort_key)
 
 
 def _node_sort_index(label: str) -> float:
@@ -566,9 +594,10 @@ def _node_sort_index(label: str) -> float:
     Returns
     -------
     float
-        The sort index based on label hash.
+        The sort index based on deterministic label hash.
     """
-    return float(hash(label) % 1000) / 1000.0
+    digest = hashlib.md5(label.encode(), usedforsecurity=False).digest()
+    return int.from_bytes(digest[:4], "big") / 4294967296.0
 
 
 def _find_back_edges(graph: KineticGraph, compartment_labels: set[str]) -> set[tuple[str, str]]:

tests/inspect/kinetic_scheme/test_k_matrix_parser.py

@@ -2,6 +2,8 @@
 
 from __future__ import annotations
 
+from copy import deepcopy
+
 import pytest
 from glotaran.testing.simulated_data.parallel_spectral_decay import SCHEME as SCHEME_PAR
 from glotaran.testing.simulated_data.sequential_spectral_decay import SCHEME as SCHEME_SEQ
@@ -175,3 +177,14 @@ def test_exclude_megacomplexes(self) -> None:
             exclude_megacomplexes={"megacomplex_sequential_decay"},
         )
         assert len(transitions) == 0
+
+    def test_missing_dataset_megacomplex_raises(self) -> None:
+        """Dataset referencing undefined megacomplex raises ValueError."""
+        model = deepcopy(SCHEME_SEQ.model)
+        model.dataset["dataset_1"].megacomplex = ["missing_mc"]
+
+        with pytest.raises(
+            ValueError,
+            match="referenced by dataset 'dataset_1' not found in model",
+        ):
+            extract_dataset_transitions("dataset_1", model, SCHEME_SEQ.parameters)

tests/inspect/kinetic_scheme/test_layout.py

@@ -2,12 +2,15 @@
 
 from __future__ import annotations
 
+import hashlib
+
 import pytest
 
 from pyglotaran_extras.inspect.kinetic_scheme._k_matrix_parser import Transition
 from pyglotaran_extras.inspect.kinetic_scheme._kinetic_graph import KineticGraph
 from pyglotaran_extras.inspect.kinetic_scheme._layout import LayoutAlgorithm
 from pyglotaran_extras.inspect.kinetic_scheme._layout import _find_connected_components
+from pyglotaran_extras.inspect.kinetic_scheme._layout import _node_sort_index
 from pyglotaran_extras.inspect.kinetic_scheme._layout import compute_layout
 
 
@@ -115,6 +118,15 @@ def test_deterministic_output(self) -> None:
         for label in pos1:
             assert pos1[label] == pos2[label]
 
+    def test_node_sort_index_is_deterministic(self) -> None:
+        """Node sort index should use deterministic hashing."""
+        label = "species_2"
+        digest = hashlib.md5(label.encode(), usedforsecurity=False).digest()
+        expected = int.from_bytes(digest[:4], "big") / 4294967296.0
+        actual = _node_sort_index(label)
+        assert actual == expected
+        assert 0.0 <= actual < 1.0
+
     def test_parallel_nodes_side_by_side(self) -> None:
         """Parallel decay nodes (all isolated) should be on the same row."""
         graph = _make_parallel_graph()