feat(transformations): transformations base primitives

Author: LeonidElkin

src/pysatl_core/__init__.py

@@ -10,6 +10,9 @@
 
 from importlib.metadata import version
 
+# isort: off
+from .transformations import *
+from .transformations import __all__ as _transformations_all
 from .distributions import *
 from .distributions import __all__ as _distr_all
 from .families import *
@@ -18,6 +21,7 @@
 from .sampling import __all__ as _sampling_all
 from .types import *
 from .types import __all__ as _types_all
+# isort: on
 
 __version__ = version("pysatl-core")
 __all__ = [
@@ -26,9 +30,11 @@
     *_family_all,
     *_types_all,
     *_sampling_all,
+    *_transformations_all,
 ]
 
 del _distr_all
 del _family_all
 del _types_all
+del _transformations_all
 del _sampling_all

src/pysatl_core/distributions/distribution.py

@@ -20,6 +20,7 @@
     ComputationStrategy,
     SamplingStrategy,
 )
+from pysatl_core.transformations.operators_mixin import TransformationOperatorsMixin
 from pysatl_core.types import DEFAULT_ANALYTICAL_COMPUTATION_LABEL, NumericArray
 
 _KEEP: object = object()
@@ -38,7 +39,7 @@
     )
 
 
-class Distribution(ABC):
+class Distribution(TransformationOperatorsMixin, ABC):
     """
     Protocol defining the interface for probability distributions.
 
@@ -57,7 +58,10 @@ class Distribution(ABC):
             | Mapping[LabelName, AnalyticalComputation[Any, Any]]
         ),
     ]
-        Direct analytical computations provided by the distribution.
+        Distribution-provided characteristic methods.
+        For non-transformed distributions every method in this mapping is
+        fully analytical, so this mapping matches the set of loops with
+        ``is_analytical=True`` in the graph view.
     sampling_strategy : SamplingStrategy
         Strategy for generating random samples.
     computation_strategy : ComputationStrategy
@@ -92,7 +96,9 @@ def __init__(
                     | Mapping[LabelName, AnalyticalComputation[Any, Any]]
                 ),
             ]
-            Analytical computations provided by the distribution.
+            Distribution-provided characteristic methods.
+            For non-transformed distributions these methods are fully
+            analytical.
         support : Support or None, default=None
             Support of the distribution.
         sampling_strategy : SamplingStrategy or None, default=None
@@ -141,9 +147,45 @@ def distribution_type(self) -> DistributionType:
     def analytical_computations(
         self,
     ) -> Mapping[GenericCharacteristicName, Mapping[LabelName, AnalyticalComputation[Any, Any]]]:
-        """Return analytical computations provided directly by this distribution."""
+        """
+        Return distribution-provided characteristic methods.
+
+        For non-transformed distributions this mapping coincides with
+        graph loops marked as ``is_analytical=True``.
+        """
         return self._analytical_computations
 
+    def loop_is_analytical(
+        self,
+        characteristic_name: GenericCharacteristicName,
+        label_name: LabelName,
+    ) -> bool:
+        """
+        Tell whether a self-loop method is fully analytical in the graph.
+
+        Parameters
+        ----------
+        characteristic_name : GenericCharacteristicName
+            Characteristic name of the self-loop.
+        label_name : LabelName
+            Label of the analytical computation variant.
+
+        Returns
+        -------
+        bool
+            ``True`` when every required predecessor in the transformation
+            chain is analytical.
+
+        Notes
+        -----
+        Presence in ``analytical_computations`` means that a characteristic
+        has at least one analytical ancestor in its derivation chain.
+        For non-transformed distributions these notions coincide, therefore
+        this method always returns ``True``.
+        """
+        _ = characteristic_name, label_name
+        return True
+
     @property
     def sampling_strategy(self) -> SamplingStrategy:
         """Return the currently attached sampling strategy."""

src/pysatl_core/distributions/registry/__init__.py

@@ -29,6 +29,7 @@
     ComputationEdgeMeta,
     EdgeMeta,
     GraphInvariantError,
+    TransformationLoopEdgeMeta,
 )
 
 __all__ = [
@@ -37,6 +38,7 @@
     "EdgeMeta",
     "ComputationEdgeMeta",
     "AnalyticalLoopEdgeMeta",
+    "TransformationLoopEdgeMeta",
     "GraphInvariantError",
     # Constraint types
     "Constraint",

src/pysatl_core/distributions/registry/graph.py

@@ -33,6 +33,7 @@
     ComputationEdgeMeta,
     EdgeMeta,
     GraphInvariantError,
+    TransformationLoopEdgeMeta,
 )
 
 if TYPE_CHECKING:
@@ -304,6 +305,15 @@ def _compute_definitive_nodes(self, distr: Distribution) -> set[GenericCharacter
                 definitive.add(name)
         return definitive
 
+    @staticmethod
+    def _loop_is_analytical(
+        distr: Distribution,
+        characteristic_name: GenericCharacteristicName,
+        label_name: LabelName,
+    ) -> bool:
+        """Resolve loop analytical flag for a distribution-provided method."""
+        return distr.loop_is_analytical(characteristic_name, label_name)
+
     @staticmethod
     def _attach_analytical_loops(
         adj: dict[
@@ -314,12 +324,14 @@ def _attach_analytical_loops(
         present_nodes: set[GenericCharacteristicName],
     ) -> None:
         """
-        Attach analytical self-loops for distribution-provided computations.
+        Attach distribution-provided self-loops to the view graph.
 
         Notes
         -----
-        Analytical loops are only added for characteristics present in this view.
-        Each labeled analytical computation becomes one loop edge ``char -> char``.
+        Loops are only added for characteristics present in this view.
+        Each labeled computation in ``analytical_computations`` becomes one
+        loop edge ``char -> char``. The loop class is selected via
+        ``distr.loop_is_analytical(...)``.
         """
         for characteristic_name, labeled_methods in distr.analytical_computations.items():
             if characteristic_name not in present_nodes:
@@ -329,7 +341,15 @@ def _attach_analytical_loops(
                 characteristic_name, {}
             )
             for label_name, analytical_method in labeled_methods.items():
-                loop_variants[label_name] = AnalyticalLoopEdgeMeta(method=analytical_method)
+                loop_variants[label_name] = (
+                    AnalyticalLoopEdgeMeta(method=analytical_method)
+                    if CharacteristicRegistry._loop_is_analytical(
+                        distr,
+                        characteristic_name,
+                        label_name,
+                    )
+                    else TransformationLoopEdgeMeta(method=analytical_method)
+                )
 
     def view(self, distr: Distribution) -> RegistryView:
         """

src/pysatl_core/distributions/registry/graph_primitives.py

@@ -77,6 +77,23 @@ def edge_kind(self) -> str:
         return "analytical_loop"
 
 
+@dataclass(frozen=True, slots=True)
+class TransformationLoopEdgeMeta(EdgeMeta):
+    """
+    Edge metadata for transformation-provided self-loops.
+
+    Such loops are attached from ``analytical_computations`` as regular
+    stopping points for the strategy, but they are not considered fully
+    analytical by the graph semantics.
+    """
+
+    method: AnalyticalComputation[Any, Any]
+    is_analytical: bool = field(default=False)
+
+    def edge_kind(self) -> str:
+        return "transformation_loop"
+
+
 class GraphInvariantError(RuntimeError):
     """
     Raised when characteristic graph invariants are violated.

src/pysatl_core/distributions/strategies.py

@@ -11,7 +11,7 @@
 __copyright__ = "Copyright (c) 2025 PySATL project"
 __license__ = "SPDX-License-Identifier: MIT"
 
-from typing import TYPE_CHECKING, Protocol, cast
+from typing import TYPE_CHECKING, Any, Protocol, cast
 
 import numpy as np
 
@@ -20,14 +20,13 @@
 
 if TYPE_CHECKING:
     from collections.abc import Mapping
-    from typing import Any
 
     from pysatl_core.distributions.computation import (
         AnalyticalComputation,
         FittedComputationMethod,
-        Method,
     )
     from pysatl_core.distributions.distribution import Distribution
+    from pysatl_core.distributions.registry.graph import RegistryView
     from pysatl_core.types import GenericCharacteristicName, LabelName
 
 
@@ -126,16 +125,30 @@ def _pick_analytical_method(
                 f"Characteristic '{state}' provides no labeled analytical computations."
             ) from exc
 
+    @staticmethod
+    def _pick_loop_method(
+        state: GenericCharacteristicName,
+        view: RegistryView,
+    ) -> Method[Any, Any] | None:
+        """
+        Pick the first available self-loop method for a characteristic in a view.
+        """
+        loops = view.variants(state, state)
+        if not loops:
+            return None
+        return cast(Method[Any, Any], next(iter(loops.values())).method)
+
     def query_method(
         self, state: GenericCharacteristicName, distr: Distribution, **options: Any
     ) -> Method[Any, Any]:
         """
         Resolve a computation method for the target characteristic.
 
         Resolution order:
-        1. Analytical implementation from the distribution
-        2. Cached fitted method (if caching enabled)
-        3. Conversion path from an analytical characteristic via the graph
+        1. Cached fitted method (if caching enabled)
+        2. Analytical implementation for non-registry characteristics
+        3. First self-loop from the registry view
+        4. Conversion path from loop characteristics via the graph
 
         Parameters
         ----------
@@ -157,34 +170,46 @@ def query_method(
             If no analytical base exists, no conversion path is found,
             or a cycle is detected.
         """
-        # 1. Check for analytical implementation
-        if state in distr.analytical_computations:
-            return self._pick_analytical_method(state, distr.analytical_computations[state])
-
-        # 2. Check cache if enabled
+        # 1. Check cache if enabled
         if self._enable_caching:
             cached = self._cache.get(state)
             if cached is not None:
                 return cached
 
-        # 3. Require at least one analytical characteristic
+        # 2. Require at least one analytical characteristic
         if not distr.analytical_computations:
             raise RuntimeError(
                 "Distribution provides no analytical computations to ground conversions."
             )
 
-        # 4. Get filtered graph view for this distribution
-        reg = characteristic_registry().view(distr)
+        # 3. Non-registry characteristics are resolved directly.
+        # It covers the situation where user is providing their analytical computation which isn't
+        # in the graph
+        registry = characteristic_registry()
+        if state not in registry.declared_characteristics:
+            if state in distr.analytical_computations:
+                return self._pick_analytical_method(state, distr.analytical_computations[state])
+            raise RuntimeError(
+                f"Characteristic '{state}' is not declared in the registry and has no "
+                "analytical implementation in the distribution."
+            )
+
+        # 4. Get filtered graph view for this distribution.
+        view = registry.view(distr)
 
         self._push_guard(distr, state)
         try:
-            # 5. Try each analytical characteristic as a source
+            loop_method = self._pick_loop_method(state, view)
+            if loop_method is not None:
+                return loop_method
+
+            # 5. Try each loop characteristic as a source
             for src in distr.analytical_computations:
-                if src == state:
-                    return self._pick_analytical_method(src, distr.analytical_computations[src])
+                if not view.variants(src, src):
+                    continue
 
                 # Find conversion path in the graph
-                path = reg.find_path(src, state)
+                path = view.find_path(src, state)
                 if not path:
                     continue
 
@@ -201,7 +226,8 @@ def query_method(
                 return last_fitted
 
             raise RuntimeError(
-                f"No conversion path from any analytical characteristic to '{state}'."
+                "No conversion path from any characteristic in "
+                f"analytical_computations to '{state}'."
             )
         finally:
             self._pop_guard(distr, state)