feat(registry): added support for hyper edges in the graph

Author: LeonidElkin

src/pysatl_core/distributions/registry/graph.py

@@ -10,7 +10,7 @@
 
 Core concepts:
  - **Nodes**: Characteristics (PDF, CDF, etc.) with presence and definitiveness rules
- - **Edges**: Unary computation methods between characteristics
+ - **Edges**: Computation methods from one-or-many sources to one target
  - **Constraints**: Rules that determine when nodes/edges are applicable
  - **View**: A filtered subgraph for a specific distribution
  - **Definitive characteristics**: Starting points for computations
@@ -64,14 +64,14 @@ class CharacteristicRegistry:
     add_characteristic(name, is_definitive, presence_constraint=None, definitive_constraint=None)
         Declare a characteristic with presence and optional definitiveness rules.
     add_computation(method, label=DEFAULT_COMPUTATION_KEY, constraint=None)
-        Add a unary computation edge between declared nodes.
+        Add a computation edge between declared nodes.
     view(distr)
         Create a filtered view for the given distribution.
 
     Notes
     -----
     - Nodes must be declared before adding computations
-    - Only unary computations (1 source → 1 target) are supported
+    - Only many-to-one computations (n sources → 1 target) are supported
     - No invariant validation happens during mutation; validation occurs when
       creating a view with view()
     """
@@ -88,10 +88,12 @@ def __init__(self) -> None:
         if getattr(self, "_initialized", False):
             return
 
-        # Adjacency: src → dst → label → [EdgeMeta]
+        # Adjacency projection: src → dst → label → [ComputationEdgeMeta]
+        # For hyperedges (many sources -> one target), the same edge metadata object
+        # is projected under each source to preserve graph reachability semantics.
         self._adj: dict[
             GenericCharacteristicName,
-            dict[GenericCharacteristicName, dict[LabelName, list[EdgeMeta]]],
+            dict[GenericCharacteristicName, dict[LabelName, list[ComputationEdgeMeta]]],
         ] = {}
         self._all_nodes: set[GenericCharacteristicName] = set()
 
@@ -179,12 +181,12 @@ def add_computation(
         constraint: GraphPrimitiveConstraint | None = None,
     ) -> None:
         """
-        Add a labeled unary computation edge.
+        Add a labeled computation edge.
 
         Parameters
         ----------
         method : ComputationMethod
-            Computation object with exactly one source and one target.
+            Computation object with one-or-many sources and one target.
         label : LabelName, default=DEFAULT_COMPUTATION_KEY
             Variant label for the edge.
         constraint : GraphPrimitiveConstraint, optional
@@ -193,33 +195,36 @@ def add_computation(
         Raises
         ------
         ValueError
-            If method is not unary, or source/target nodes are not declared.
+            If method has no sources, or source/target nodes are not declared.
 
         Notes
         -----
         - Multiple edges with different labels can exist between the same nodes
         - The first matching edge for each label is kept when creating views
+        - Hyperedges are represented as projected edges from each source to target,
+          while preserving one shared underlying computation method.
         """
-        if len(method.sources) != 1:
-            raise ValueError("Only unary computations are supported (1 source → 1 target).")
+        if not method.sources:
+            raise ValueError("Computation must define at least one source characteristic.")
 
-        src = method.sources[0]
+        unique_sources = tuple(dict.fromkeys(method.sources))
         dst = method.target
 
-        if not self._ensure_node(src) or not self._ensure_node(dst):
+        if not self._ensure_node(dst) or any(not self._ensure_node(src) for src in unique_sources):
             raise ValueError("Source characteristic or destination characteristic is invalid.")
 
-        self._adj[src].setdefault(dst, {})
+        edge_meta = ComputationEdgeMeta(
+            method=method,
+            constraint=constraint or GraphPrimitiveConstraint(),
+        )
+
         # TODO: We need to be careful here if some constraint more general and with the same label
         #  than other it can consume it. Actually, the same label methods should not intersect their
         #  constraints
-        self._adj[src][dst].setdefault(label, [])
-        self._adj[src][dst][label].append(
-            ComputationEdgeMeta(
-                method=method,
-                constraint=constraint or GraphPrimitiveConstraint(),
-            )
-        )
+        for src in unique_sources:
+            self._adj[src].setdefault(dst, {})
+            self._adj[src][dst].setdefault(label, [])
+            self._adj[src][dst][label].append(edge_meta)
 
     def add_characteristic(
         self,
@@ -342,41 +347,36 @@ def view(self, distr: Distribution) -> RegistryView:
 
         Notes
         -----
-        1. Filters edges by their constraints
-        2. Removes edges touching absent nodes
+        1. Computes present nodes for the distribution
+        2. Filters edges by node presence and edge constraints
         3. Adds analytical self-loops from distribution analytical computations
         4. Computes definitive nodes from the remaining present nodes
         5. Validates graph invariants
         """
-        # 1) Filter edges by applicability
+        # 1) Compute present nodes once and pre-create adjacency.
+        present_nodes = self._compute_present_nodes(distr)
         adj: dict[
             GenericCharacteristicName, dict[GenericCharacteristicName, dict[LabelName, EdgeMeta]]
-        ] = {}
-        for src, d in self._adj.items():
-            for dst, variants in d.items():
+        ] = {node: {} for node in present_nodes}
+
+        # 2) Filter edges by node presence and applicability.
+        for src in present_nodes:
+            for dst, variants in self._adj.get(src, {}).items():
+                if dst not in present_nodes:
+                    continue
                 kept: dict[LabelName, EdgeMeta] = {}
                 for label, metas in variants.items():
                     for edge in metas:
-                        if edge.constraint.allows(distr):
+                        if edge.constraint.allows(distr) and all(
+                            source in present_nodes for source in edge.method.sources
+                        ):
                             kept[label] = edge
                             # TODO: It is possible that there are two edges under the same label
                             #  that fit the same distribution, this should not be the case.
                             #  Taking the first one for now
                             break
                 if kept:
-                    adj.setdefault(src, {}).setdefault(dst, {}).update(kept)
-
-        # 2) Filter by node presence
-        present_nodes = self._compute_present_nodes(distr)
-        if present_nodes:
-            adj = {
-                src: {dst: dict(variants) for dst, variants in d.items() if dst in present_nodes}
-                for src, d in adj.items()
-                if src in present_nodes
-            }
-        # Ensure isolated present nodes are preserved
-        for node in present_nodes:
-            adj.setdefault(node, {})
+                    adj[src][dst] = kept
 
         # 3) Attach analytical loops
         self._attach_analytical_loops(adj, distr, present_nodes)
@@ -438,6 +438,14 @@ def __init__(
         for src, d in adj.items():
             self._adj[src] = {dst: dict(variants) for dst, variants in d.items()}
 
+        self._rev_adj: dict[GenericCharacteristicName, set[GenericCharacteristicName]] = {
+            node: set() for node in self._adj
+        }
+        for src, d in self._adj.items():
+            for dst, variants in d.items():
+                if variants:
+                    self._rev_adj.setdefault(dst, set()).add(src)
+
         self.definitive_characteristics: set[GenericCharacteristicName] = set(definitive_nodes)
         self.all_characteristics: set[GenericCharacteristicName] = set(present_nodes)
 
@@ -504,11 +512,7 @@ def predecessors(self, v: GenericCharacteristicName) -> set[GenericCharacteristi
         set of str
             Characteristics that can reach v directly.
         """
-        res: set[GenericCharacteristicName] = set()
-        for src, d in self._adj.items():
-            if v in d and d[v]:
-                res.add(src)
-        return res
+        return set(self._rev_adj.get(v, set()))
 
     def variants(
         self, src: GenericCharacteristicName, dst: GenericCharacteristicName
@@ -643,16 +647,8 @@ def _definitive_strongly_connected(self) -> bool:
         if fwd != (defs - {start}):
             return False
 
-        # Check reverse reachability
-        seen: set[GenericCharacteristicName] = {start}
-        stack = [start]
-        while stack:
-            v = stack.pop()
-            for w in self.predecessors(v):
-                if w in defs and w not in seen:
-                    seen.add(w)
-                    stack.append(w)
-        return seen == defs
+        rev = self._reachable_from_many({start}, allowed=defs, reverse=True)
+        return rev == (defs - {start})
 
     def _all_indefinitives_reachable_from_definitives(self) -> bool:
         """
@@ -667,9 +663,7 @@ def _all_indefinitives_reachable_from_definitives(self) -> bool:
         if not indefs:
             return True
 
-        total: set[GenericCharacteristicName] = set()
-        for d in self.definitive_characteristics:
-            total |= self._reachable_from(d)
+        total = self._reachable_from_many(self.definitive_characteristics)
         return indefs.issubset(total)
 
     def _exists_path_from_indefinitive_to_definitive(self) -> bool:
@@ -682,46 +676,78 @@ def _exists_path_from_indefinitive_to_definitive(self) -> bool:
             True if such a path exists (which would violate invariants).
         """
         defs = self.definitive_characteristics
-        return any(self._reachable_from(i) & defs for i in self.indefinitive_characteristics)
+        if not defs:
+            return False
 
-    def _reachable_from(
+        can_reach_definitive = self._reachable_from_many(defs, reverse=True)
+        return bool(can_reach_definitive & self.indefinitive_characteristics)
+
+    def _reachable_from_many(
         self,
-        src: GenericCharacteristicName,
+        sources: set[GenericCharacteristicName],
         *,
         allowed: set[GenericCharacteristicName] | None = None,
+        reverse: bool = False,
     ) -> set[GenericCharacteristicName]:
         """
-        Compute forward reachable nodes from src.
+        Compute reachable nodes from multiple sources.
 
         Parameters
         ----------
-        src : str
-            Starting node.
+        sources : set of str
+            Starting nodes.
         allowed : set of str, optional
-            Restrict to this set of nodes.
+            Restrict traversal to this set of nodes.
+        reverse : bool, default=False
+            If True, traverse reverse edges.
 
         Returns
         -------
         set of str
-            Nodes reachable from src (excluding src itself).
+            Nodes reachable from sources (excluding sources themselves).
         """
-        if allowed is not None and src not in allowed:
+        starts = {s for s in sources if allowed is None or s in allowed}
+        if not starts:
             return set()
 
         visited: set[GenericCharacteristicName] = set()
-        stack = [src]
+        stack = list(starts)
         while stack:
             v = stack.pop()
             if v in visited:
                 continue
             visited.add(v)
-            for w in self.successors_nodes(v):
+            neighbors = self._rev_adj.get(v, set()) if reverse else self.successors_nodes(v)
+            for w in neighbors:
                 if allowed is not None and w not in allowed:
                     continue
                 if w not in visited:
                     stack.append(w)
-        visited.discard(src)
-        return visited
+
+        return visited - starts
+
+    def _reachable_from(
+        self,
+        src: GenericCharacteristicName,
+        *,
+        allowed: set[GenericCharacteristicName] | None = None,
+    ) -> set[GenericCharacteristicName]:
+        """
+        Compute forward reachable nodes from src.
+
+        Parameters
+        ----------
+        src : str
+            Starting node.
+        allowed : set of str, optional
+            Restrict to this set of nodes.
+
+        Returns
+        -------
+        set of str
+            Nodes reachable from src (excluding src itself).
+        """
+        return self._reachable_from_many({src}, allowed=allowed)
 
     @staticmethod
     def _pick_method(
@@ -747,5 +773,5 @@ def _pick_method(
             return variants[prefer_label].method
         if DEFAULT_COMPUTATION_KEY in variants:
             return variants[DEFAULT_COMPUTATION_KEY].method
-        label = sorted(variants.keys())[0]
+        label = min(variants)
         return variants[label].method