feat(registry): add looped edges for graph view for analytical computations

Author: LeonidElkin

src/pysatl_core/distributions/registry/__init__.py

@@ -25,6 +25,8 @@
 )
 from .graph_primitives import (
     DEFAULT_COMPUTATION_KEY,
+    AnalyticalLoopEdgeMeta,
+    ComputationEdgeMeta,
     EdgeMeta,
     GraphInvariantError,
 )
@@ -33,6 +35,8 @@
     # Graph primitives and constants
     "DEFAULT_COMPUTATION_KEY",
     "EdgeMeta",
+    "ComputationEdgeMeta",
+    "AnalyticalLoopEdgeMeta",
     "GraphInvariantError",
     # Constraint types
     "Constraint",

src/pysatl_core/distributions/registry/graph.py

@@ -29,14 +29,16 @@
 from pysatl_core.distributions.registry.constraint import GraphPrimitiveConstraint
 from pysatl_core.distributions.registry.graph_primitives import (
     DEFAULT_COMPUTATION_KEY,
+    AnalyticalLoopEdgeMeta,
+    ComputationEdgeMeta,
     EdgeMeta,
     GraphInvariantError,
 )
 
 if TYPE_CHECKING:
     from pysatl_core.distributions.computation import ComputationMethod
     from pysatl_core.distributions.distribution import Distribution
-    from pysatl_core.types import GenericCharacteristicName
+    from pysatl_core.types import GenericCharacteristicName, LabelName
 
 
 # --------------------------------------------------------------------------- #
@@ -89,7 +91,7 @@ def __init__(self) -> None:
         # Adjacency: src → dst → label → [EdgeMeta]
         self._adj: dict[
             GenericCharacteristicName,
-            dict[GenericCharacteristicName, dict[str, list[EdgeMeta]]],
+            dict[GenericCharacteristicName, dict[LabelName, list[EdgeMeta]]],
         ] = {}
         self._all_nodes: set[GenericCharacteristicName] = set()
 
@@ -98,7 +100,7 @@ def __init__(self) -> None:
         self._def_rules: dict[GenericCharacteristicName, GraphPrimitiveConstraint] = {}
 
         # Label preference for path finding
-        self.label_preference: tuple[str, ...] = (DEFAULT_COMPUTATION_KEY,)
+        self.label_preference: tuple[LabelName, ...] = (DEFAULT_COMPUTATION_KEY,)
 
         self._initialized = True
 
@@ -128,6 +130,11 @@ def _ensure_node(self, node: GenericCharacteristicName) -> bool:
         """Check if a node has been declared via add_characteristic()."""
         return node in self._all_nodes
 
+    @property
+    def declared_characteristics(self) -> set[GenericCharacteristicName]:
+        """Return declared registry characteristics."""
+        return set(self._all_nodes)
+
     def _add_presence_rule(
         self, name: GenericCharacteristicName, constraint: GraphPrimitiveConstraint | None
     ) -> None:
@@ -168,7 +175,7 @@ def add_computation(
         self,
         method: ComputationMethod[Any, Any],
         *,
-        label: str = DEFAULT_COMPUTATION_KEY,
+        label: LabelName = DEFAULT_COMPUTATION_KEY,
         constraint: GraphPrimitiveConstraint | None = None,
     ) -> None:
         """
@@ -178,7 +185,7 @@ def add_computation(
         ----------
         method : ComputationMethod
             Computation object with exactly one source and one target.
-        label : str, default=DEFAULT_COMPUTATION_KEY
+        label : LabelName, default=DEFAULT_COMPUTATION_KEY
             Variant label for the edge.
         constraint : GraphPrimitiveConstraint, optional
             Edge applicability constraint. If None, a pass-through constraint is used.
@@ -208,7 +215,7 @@ def add_computation(
         #  constraints
         self._adj[src][dst].setdefault(label, [])
         self._adj[src][dst][label].append(
-            EdgeMeta(
+            ComputationEdgeMeta(
                 method=method,
                 constraint=constraint or GraphPrimitiveConstraint(),
             )
@@ -292,6 +299,33 @@ def _compute_definitive_nodes(self, distr: Distribution) -> set[GenericCharacter
                 definitive.add(name)
         return definitive
 
+    @staticmethod
+    def _attach_analytical_loops(
+        adj: dict[
+            GenericCharacteristicName,
+            dict[GenericCharacteristicName, dict[LabelName, EdgeMeta]],
+        ],
+        distr: Distribution,
+        present_nodes: set[GenericCharacteristicName],
+    ) -> None:
+        """
+        Attach analytical self-loops for distribution-provided computations.
+
+        Notes
+        -----
+        Analytical loops are only added for characteristics present in this view.
+        Each labeled analytical computation becomes one loop edge ``char -> char``.
+        """
+        for characteristic_name, labeled_methods in distr.analytical_computations.items():
+            if characteristic_name not in present_nodes:
+                continue
+
+            loop_variants = adj.setdefault(characteristic_name, {}).setdefault(
+                characteristic_name, {}
+            )
+            for label_name, analytical_method in labeled_methods.items():
+                loop_variants[label_name] = AnalyticalLoopEdgeMeta(method=analytical_method)
+
     def view(self, distr: Distribution) -> RegistryView:
         """
         Create a filtered view of the graph for the given distribution.
@@ -310,16 +344,17 @@ def view(self, distr: Distribution) -> RegistryView:
         -----
         1. Filters edges by their constraints
         2. Removes edges touching absent nodes
-        3. Computes definitive nodes from the remaining present nodes
-        4. Validates graph invariants
+        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
         adj: dict[
-            GenericCharacteristicName, dict[GenericCharacteristicName, dict[str, EdgeMeta]]
+            GenericCharacteristicName, dict[GenericCharacteristicName, dict[LabelName, EdgeMeta]]
         ] = {}
         for src, d in self._adj.items():
             for dst, variants in d.items():
-                kept: dict[str, EdgeMeta] = {}
+                kept: dict[LabelName, EdgeMeta] = {}
                 for label, metas in variants.items():
                     for edge in metas:
                         if edge.constraint.allows(distr):
@@ -343,7 +378,10 @@ def view(self, distr: Distribution) -> RegistryView:
         for node in present_nodes:
             adj.setdefault(node, {})
 
-        # 3) Compute definitive nodes (must be present)
+        # 3) Attach analytical loops
+        self._attach_analytical_loops(adj, distr, present_nodes)
+
+        # 4) Compute definitive nodes (must be present)
         definitive_nodes = self._compute_definitive_nodes(distr) & present_nodes
 
         return RegistryView(adj, definitive_nodes, present_nodes)
@@ -387,14 +425,15 @@ def __init__(
         self,
         adj: Mapping[
             GenericCharacteristicName,
-            Mapping[GenericCharacteristicName, Mapping[str, EdgeMeta]],
+            Mapping[GenericCharacteristicName, Mapping[LabelName, EdgeMeta]],
         ],
         definitive_nodes: set[GenericCharacteristicName],
         present_nodes: set[GenericCharacteristicName],
     ) -> None:
         # Deep copy adjacency to ensure immutability
         self._adj: dict[
-            GenericCharacteristicName, dict[GenericCharacteristicName, dict[str, EdgeMeta]]
+            GenericCharacteristicName,
+            dict[GenericCharacteristicName, dict[LabelName, EdgeMeta]],
         ] = {}
         for src, d in adj.items():
             self._adj[src] = {dst: dict(variants) for dst, variants in d.items()}
@@ -419,7 +458,7 @@ def indefinitive_characteristics(self) -> set[GenericCharacteristicName]:
 
     def successors(
         self, v: GenericCharacteristicName
-    ) -> Mapping[GenericCharacteristicName, Mapping[str, EdgeMeta]]:
+    ) -> Mapping[GenericCharacteristicName, Mapping[LabelName, EdgeMeta]]:
         """
         Get outgoing edges from a characteristic.
 
@@ -430,7 +469,7 @@ def successors(
 
         Returns
         -------
-        Mapping[str, Mapping[str, EdgeMeta]]
+        Mapping[str, Mapping[LabelName, EdgeMeta]]
             Destination → label → edge metadata.
         """
         return self._adj.get(v, {})
@@ -473,7 +512,7 @@ def predecessors(self, v: GenericCharacteristicName) -> set[GenericCharacteristi
 
     def variants(
         self, src: GenericCharacteristicName, dst: GenericCharacteristicName
-    ) -> Mapping[str, EdgeMeta]:
+    ) -> Mapping[LabelName, EdgeMeta]:
         """
         Get all labeled edges between two characteristics.
 
@@ -484,17 +523,35 @@ def variants(
 
         Returns
         -------
-        Mapping[str, EdgeMeta]
+        Mapping[LabelName, EdgeMeta]
             Label → edge metadata mapping.
         """
         return self._adj.get(src, {}).get(dst, {})
 
+    def analytical_variants(self, state: GenericCharacteristicName) -> Mapping[LabelName, EdgeMeta]:
+        """
+        Get analytical self-loop variants for a characteristic.
+
+        Parameters
+        ----------
+        state : str
+            Characteristic name.
+
+        Returns
+        -------
+        Mapping[LabelName, EdgeMeta]
+            Label → analytical loop metadata for ``state -> state``.
+        """
+        return {
+            label: edge for label, edge in self.variants(state, state).items() if edge.is_analytical
+        }
+
     def find_path(
         self,
         src: GenericCharacteristicName,
         dst: GenericCharacteristicName,
         *,
-        prefer_label: str | None = None,
+        prefer_label: LabelName | None = None,
     ) -> list[Any] | None:
         """
         Find a computation path from src to dst using BFS.
@@ -503,12 +560,12 @@ def find_path(
         ----------
         src, dst : str
             Source and destination characteristics.
-        prefer_label : str, optional
+        prefer_label : LabelName, optional
             Preferred edge label to use when multiple options exist.
 
         Returns
         -------
-        list of ComputationMethod or None
+        list of Any or None
             List of computation methods forming the path, or None if no path exists.
 
         Notes
@@ -668,17 +725,17 @@ def _reachable_from(
 
     @staticmethod
     def _pick_method(
-        variants: Mapping[str, EdgeMeta],
-        prefer_label: str | None,
+        variants: Mapping[LabelName, EdgeMeta],
+        prefer_label: LabelName | None,
     ) -> Any:
         """
         Select a method from label variants.
 
         Parameters
         ----------
-        variants : Mapping[str, EdgeMeta]
+        variants : Mapping[LabelName, EdgeMeta]
             Available edge variants.
-        prefer_label : str, optional
+        prefer_label : LabelName, optional
             Preferred label.
 
         Returns

src/pysatl_core/distributions/registry/graph_primitives.py

@@ -8,36 +8,73 @@
 __copyright__ = "Copyright (c) 2025 PySATL project"
 __license__ = "SPDX-License-Identifier: MIT"
 
+from abc import ABC, abstractmethod
 from dataclasses import dataclass, field
-from typing import TYPE_CHECKING
+from typing import Any
 
+from pysatl_core.distributions.computation import (
+    AnalyticalComputation,
+    ComputationMethod,
+)
 from pysatl_core.distributions.registry.constraint import GraphPrimitiveConstraint
+from pysatl_core.types import LabelName
 
-if TYPE_CHECKING:
-    from typing import Any
+type EdgeMethod = ComputationMethod[Any, Any] | AnalyticalComputation[Any, Any]
 
-    from pysatl_core.distributions.computation import ComputationMethod
-
-DEFAULT_COMPUTATION_KEY: str = "PySATL_default_computation"
+DEFAULT_COMPUTATION_KEY: LabelName = "PySATL_default_computation"
 """Default label for computation edges when no specific label is provided."""
 
 
 @dataclass(frozen=True, slots=True)
-class EdgeMeta:
+class EdgeMeta(ABC):
     """
     Metadata for a computation edge in the characteristic graph.
 
     Parameters
     ----------
-    method : ComputationMethod
+    method : EdgeMethod
         The computation method that defines the edge.
     constraint : GraphPrimitiveConstraint
         Constraint determining when this edge is applicable to a distribution.
         Defaults to a pass-through constraint that always allows.
+    is_analytical : bool
+        Whether this edge represents an analytical computation.
     """
 
-    method: ComputationMethod[Any, Any]
+    method: EdgeMethod
     constraint: GraphPrimitiveConstraint = field(default_factory=GraphPrimitiveConstraint)
+    is_analytical: bool = field(default=False)
+
+    @abstractmethod
+    def edge_kind(self) -> str:
+        """Return edge kind identifier."""
+        ...
+
+
+@dataclass(frozen=True, slots=True)
+class ComputationEdgeMeta(EdgeMeta):
+    """
+    Edge metadata for conversion computations from the registry graph.
+    """
+
+    method: ComputationMethod[Any, Any]
+    is_analytical: bool = field(default=False)
+
+    def edge_kind(self) -> str:
+        return "computation"
+
+
+@dataclass(frozen=True, slots=True)
+class AnalyticalLoopEdgeMeta(EdgeMeta):
+    """
+    Edge metadata for self-loop analytical computations from a distribution.
+    """
+
+    method: AnalyticalComputation[Any, Any]
+    is_analytical: bool = field(default=True)
+
+    def edge_kind(self) -> str:
+        return "analytical_loop"
 
 
 class GraphInvariantError(RuntimeError):