antecedent.extensibility

Typing Protocols for slow-path Python callbacks.

View Source

  1"""Typing Protocols for slow-path Python callbacks.
  2
  3These are documentation / type-checking aids. Native bridges accept any
  4callable matching the shapes below; they reacquire the GIL and force serial
  5execution (non-native performance).
  6"""
  7
  8from __future__ import annotations
  9
 10from typing import Protocol, Sequence, runtime_checkable
 11
 12import numpy as np
 13from numpy.typing import NDArray
 14
 15
 16@runtime_checkable
 17class CiBatchTest(Protocol):
 18    """Batch conditional-independence test.
 19
 20    Parameters
 21    ----------
 22    columns:
 23        List of 1-d float64 columns.
 24    queries:
 25        List of ``(x, y, z_idxs)`` where ``z_idxs`` is a list of conditioning
 26        column indexes.
 27    """
 28
 29    def __call__(
 30        self,
 31        columns: Sequence[NDArray[np.float64]],
 32        queries: Sequence[tuple[int, int, list[int]]],
 33    ) -> Sequence[tuple[float, float]]:
 34        """Return ``(statistic, p_value)`` per query."""
 35
 36
 37@runtime_checkable
 38class MechanismWrapper(Protocol):
 39    """Per-node mechanism override for GCM sampling / abduction.
 40
 41    Required: ``sample_noise``, ``evaluate``.
 42    Optional: ``infer_noise``, ``log_prob`` — when omitted, Rust uses an
 43    additive-noise default (``noise = y - f(pa, 0)``, Gaussian ``N(0,1)`` log-prob).
 44    """
 45
 46    def sample_noise(self, n: int) -> NDArray[np.float64]:
 47        """Draw structural noise of length ``n``."""
 48
 49    def evaluate(
 50        self,
 51        parents: Sequence[NDArray[np.float64]],
 52        noise: NDArray[np.float64],
 53    ) -> NDArray[np.float64]:
 54        """Map parents + noise → child values (length ``noise``)."""
 55
 56    def infer_noise(
 57        self,
 58        value: NDArray[np.float64],
 59        parents: Sequence[NDArray[np.float64]],
 60    ) -> NDArray[np.float64]:
 61        """Optional: abduce noise from factual ``value`` and parents."""
 62
 63    def log_prob(
 64        self,
 65        values: NDArray[np.float64],
 66        parents: Sequence[NDArray[np.float64]],
 67    ) -> NDArray[np.float64]:
 68        """Optional: log-density of ``values`` given parents."""
 69
 70
 71@runtime_checkable
 72class UtilityFn(Protocol):
 73    """Batch utility for decision evaluation."""
 74
 75    def __call__(
 76        self,
 77        actions: NDArray[np.float64],
 78        outcomes: NDArray[np.float64],
 79    ) -> NDArray[np.float64]:
 80        """Return flat utilities of length ``len(actions) * len(outcomes)``."""
 81
 82
 83@runtime_checkable
 84class EffectValidator(Protocol):
 85    """Custom effect refuter returning a report dict."""
 86
 87    def __call__(
 88        self,
 89        *,
 90        ate: float,
 91        se_analytic: float,
 92        method: str,
 93        adjustment_set: list[str],
 94    ) -> dict:
 95        """Must include ``passed: bool``; optional ``refuted_ate``, ``comparison``."""
 96
 97
 98__all__ = [
 99    "CiBatchTest",
100    "EffectValidator",
101    "MechanismWrapper",
102    "UtilityFn",
103]

@runtime_checkable

class CiBatchTest(typing.Protocol): View Source

17@runtime_checkable
18class CiBatchTest(Protocol):
19    """Batch conditional-independence test.
20
21    Parameters
22    ----------
23    columns:
24        List of 1-d float64 columns.
25    queries:
26        List of ``(x, y, z_idxs)`` where ``z_idxs`` is a list of conditioning
27        column indexes.
28    """
29
30    def __call__(
31        self,
32        columns: Sequence[NDArray[np.float64]],
33        queries: Sequence[tuple[int, int, list[int]]],
34    ) -> Sequence[tuple[float, float]]:
35        """Return ``(statistic, p_value)`` per query."""

Batch conditional-independence test.

Parameters

columns: List of 1-d float64 columns. queries: List of (x, y, z_idxs) where z_idxs is a list of conditioning column indexes.

CiBatchTest(*args, **kwargs) View Source

1771def _no_init_or_replace_init(self, *args, **kwargs):
1772    cls = type(self)
1773
1774    if cls._is_protocol:
1775        raise TypeError('Protocols cannot be instantiated')
1776
1777    # Already using a custom `__init__`. No need to calculate correct
1778    # `__init__` to call. This can lead to RecursionError. See bpo-45121.
1779    if cls.__init__ is not _no_init_or_replace_init:
1780        return
1781
1782    # Initially, `__init__` of a protocol subclass is set to `_no_init_or_replace_init`.
1783    # The first instantiation of the subclass will call `_no_init_or_replace_init` which
1784    # searches for a proper new `__init__` in the MRO. The new `__init__`
1785    # replaces the subclass' old `__init__` (ie `_no_init_or_replace_init`). Subsequent
1786    # instantiation of the protocol subclass will thus use the new
1787    # `__init__` and no longer call `_no_init_or_replace_init`.
1788    for base in cls.__mro__:
1789        init = base.__dict__.get('__init__', _no_init_or_replace_init)
1790        if init is not _no_init_or_replace_init:
1791            cls.__init__ = init
1792            break
1793    else:
1794        # should not happen
1795        cls.__init__ = object.__init__
1796
1797    cls.__init__(self, *args, **kwargs)

@runtime_checkable

class EffectValidator(typing.Protocol): View Source

84@runtime_checkable
85class EffectValidator(Protocol):
86    """Custom effect refuter returning a report dict."""
87
88    def __call__(
89        self,
90        *,
91        ate: float,
92        se_analytic: float,
93        method: str,
94        adjustment_set: list[str],
95    ) -> dict:
96        """Must include ``passed: bool``; optional ``refuted_ate``, ``comparison``."""

Custom effect refuter returning a report dict.

EffectValidator(*args, **kwargs) View Source

1771def _no_init_or_replace_init(self, *args, **kwargs):
1772    cls = type(self)
1773
1774    if cls._is_protocol:
1775        raise TypeError('Protocols cannot be instantiated')
1776
1777    # Already using a custom `__init__`. No need to calculate correct
1778    # `__init__` to call. This can lead to RecursionError. See bpo-45121.
1779    if cls.__init__ is not _no_init_or_replace_init:
1780        return
1781
1782    # Initially, `__init__` of a protocol subclass is set to `_no_init_or_replace_init`.
1783    # The first instantiation of the subclass will call `_no_init_or_replace_init` which
1784    # searches for a proper new `__init__` in the MRO. The new `__init__`
1785    # replaces the subclass' old `__init__` (ie `_no_init_or_replace_init`). Subsequent
1786    # instantiation of the protocol subclass will thus use the new
1787    # `__init__` and no longer call `_no_init_or_replace_init`.
1788    for base in cls.__mro__:
1789        init = base.__dict__.get('__init__', _no_init_or_replace_init)
1790        if init is not _no_init_or_replace_init:
1791            cls.__init__ = init
1792            break
1793    else:
1794        # should not happen
1795        cls.__init__ = object.__init__
1796
1797    cls.__init__(self, *args, **kwargs)

@runtime_checkable

class MechanismWrapper(typing.Protocol): View Source

38@runtime_checkable
39class MechanismWrapper(Protocol):
40    """Per-node mechanism override for GCM sampling / abduction.
41
42    Required: ``sample_noise``, ``evaluate``.
43    Optional: ``infer_noise``, ``log_prob`` — when omitted, Rust uses an
44    additive-noise default (``noise = y - f(pa, 0)``, Gaussian ``N(0,1)`` log-prob).
45    """
46
47    def sample_noise(self, n: int) -> NDArray[np.float64]:
48        """Draw structural noise of length ``n``."""
49
50    def evaluate(
51        self,
52        parents: Sequence[NDArray[np.float64]],
53        noise: NDArray[np.float64],
54    ) -> NDArray[np.float64]:
55        """Map parents + noise → child values (length ``noise``)."""
56
57    def infer_noise(
58        self,
59        value: NDArray[np.float64],
60        parents: Sequence[NDArray[np.float64]],
61    ) -> NDArray[np.float64]:
62        """Optional: abduce noise from factual ``value`` and parents."""
63
64    def log_prob(
65        self,
66        values: NDArray[np.float64],
67        parents: Sequence[NDArray[np.float64]],
68    ) -> NDArray[np.float64]:
69        """Optional: log-density of ``values`` given parents."""

Per-node mechanism override for GCM sampling / abduction.

Required: sample_noise, evaluate. Optional: infer_noise, log_prob — when omitted, Rust uses an additive-noise default (noise = y - f(pa, 0), Gaussian N(0,1) log-prob).

MechanismWrapper(*args, **kwargs) View Source

1771def _no_init_or_replace_init(self, *args, **kwargs):
1772    cls = type(self)
1773
1774    if cls._is_protocol:
1775        raise TypeError('Protocols cannot be instantiated')
1776
1777    # Already using a custom `__init__`. No need to calculate correct
1778    # `__init__` to call. This can lead to RecursionError. See bpo-45121.
1779    if cls.__init__ is not _no_init_or_replace_init:
1780        return
1781
1782    # Initially, `__init__` of a protocol subclass is set to `_no_init_or_replace_init`.
1783    # The first instantiation of the subclass will call `_no_init_or_replace_init` which
1784    # searches for a proper new `__init__` in the MRO. The new `__init__`
1785    # replaces the subclass' old `__init__` (ie `_no_init_or_replace_init`). Subsequent
1786    # instantiation of the protocol subclass will thus use the new
1787    # `__init__` and no longer call `_no_init_or_replace_init`.
1788    for base in cls.__mro__:
1789        init = base.__dict__.get('__init__', _no_init_or_replace_init)
1790        if init is not _no_init_or_replace_init:
1791            cls.__init__ = init
1792            break
1793    else:
1794        # should not happen
1795        cls.__init__ = object.__init__
1796
1797    cls.__init__(self, *args, **kwargs)

def sample_noise(self, n: int) -> 'NDArray[np.float64]': View Source

47    def sample_noise(self, n: int) -> NDArray[np.float64]:
48        """Draw structural noise of length ``n``."""

Draw structural noise of length n.

def evaluate( self, parents: 'Sequence[NDArray[np.float64]]', noise: 'NDArray[np.float64]') -> 'NDArray[np.float64]': View Source

50    def evaluate(
51        self,
52        parents: Sequence[NDArray[np.float64]],
53        noise: NDArray[np.float64],
54    ) -> NDArray[np.float64]:
55        """Map parents + noise → child values (length ``noise``)."""

Map parents + noise → child values (length noise).

def infer_noise( self, value: 'NDArray[np.float64]', parents: 'Sequence[NDArray[np.float64]]') -> 'NDArray[np.float64]': View Source

57    def infer_noise(
58        self,
59        value: NDArray[np.float64],
60        parents: Sequence[NDArray[np.float64]],
61    ) -> NDArray[np.float64]:
62        """Optional: abduce noise from factual ``value`` and parents."""

Optional: abduce noise from factual value and parents.

def log_prob( self, values: 'NDArray[np.float64]', parents: 'Sequence[NDArray[np.float64]]') -> 'NDArray[np.float64]': View Source

64    def log_prob(
65        self,
66        values: NDArray[np.float64],
67        parents: Sequence[NDArray[np.float64]],
68    ) -> NDArray[np.float64]:
69        """Optional: log-density of ``values`` given parents."""

Optional: log-density of values given parents.

@runtime_checkable

class UtilityFn(typing.Protocol): View Source

72@runtime_checkable
73class UtilityFn(Protocol):
74    """Batch utility for decision evaluation."""
75
76    def __call__(
77        self,
78        actions: NDArray[np.float64],
79        outcomes: NDArray[np.float64],
80    ) -> NDArray[np.float64]:
81        """Return flat utilities of length ``len(actions) * len(outcomes)``."""

Batch utility for decision evaluation.

UtilityFn(*args, **kwargs) View Source

1771def _no_init_or_replace_init(self, *args, **kwargs):
1772    cls = type(self)
1773
1774    if cls._is_protocol:
1775        raise TypeError('Protocols cannot be instantiated')
1776
1777    # Already using a custom `__init__`. No need to calculate correct
1778    # `__init__` to call. This can lead to RecursionError. See bpo-45121.
1779    if cls.__init__ is not _no_init_or_replace_init:
1780        return
1781
1782    # Initially, `__init__` of a protocol subclass is set to `_no_init_or_replace_init`.
1783    # The first instantiation of the subclass will call `_no_init_or_replace_init` which
1784    # searches for a proper new `__init__` in the MRO. The new `__init__`
1785    # replaces the subclass' old `__init__` (ie `_no_init_or_replace_init`). Subsequent
1786    # instantiation of the protocol subclass will thus use the new
1787    # `__init__` and no longer call `_no_init_or_replace_init`.
1788    for base in cls.__mro__:
1789        init = base.__dict__.get('__init__', _no_init_or_replace_init)
1790        if init is not _no_init_or_replace_init:
1791            cls.__init__ = init
1792            break
1793    else:
1794        # should not happen
1795        cls.__init__ = object.__init__
1796
1797    cls.__init__(self, *args, **kwargs)