"""Create, estimate, and sample from a null distribution.
Defines the NullDistribution, NullSampler, NullAccumulatorFactory, NullAccumulator,
NullEstimator, and the NullDataEncoder classes for use with mixle.
The NullDistribution object and its related classes are space filling objects meant for consistency in type hints.
Notes:
The density evaluates to 1.0 for any value (Any data type).
The sampler generates None for any size input.
Sequence encodings return None for any input.
"""
from typing import Any, Optional
import numpy as np
from numpy.random import RandomState
from mixle.enumeration.algorithms import QuantizedCrossIndex, QuantizedEnumerationIndex
from mixle.stats.compute.pdist import (
DataSequenceEncoder,
DistributionEnumerator,
DistributionSampler,
ParameterEstimator,
SequenceEncodableProbabilityDistribution,
SequenceEncodableStatisticAccumulator,
StatisticAccumulatorFactory,
)
[docs]
class NullDistribution(SequenceEncodableProbabilityDistribution):
"""Place-holder distribution assigning density 1.0 (log-density 0.0) to any observation (Any data type)."""
[docs]
@classmethod
def compute_capabilities(cls):
from mixle.stats.compute.capabilities import DistributionCapabilities
return DistributionCapabilities(engine_ready=("numpy", "torch"), kernel_status="generic")
[docs]
@classmethod
def compute_declaration(cls):
from mixle.stats.compute.declarations import DistributionDeclaration
return DistributionDeclaration(
name="null",
distribution_type=cls,
parameters=(),
statistics=(),
support="any",
differentiable=False,
)
def __init__(self, name: str | None = None) -> None:
"""NullDistribution object.
Args:
name (Optional[str]): Optional name for object instance.
Attributes:
name (Optional[str]): Optional name for object instance.
"""
self.name = name
def __str__(self) -> str:
"""Returns string representation of NullDistribution object."""
return "NullDistribution(name=%s)" % repr(self.name)
[docs]
def density(self, x: Any | None) -> float:
"""Density of NullDistribution. Always 1.0.
Args:
x (Optional[Any]): Observation of any type (ignored).
Returns:
1.0 for any input.
"""
return 1.0
[docs]
def log_density(self, x: Any | None) -> float:
"""Log-density of NullDistribution. Always 0.0.
Args:
x (Optional[Any]): Observation of any type (ignored).
Returns:
0.0 for any input.
"""
return 0.0
[docs]
def seq_log_density(self, x: Any | None) -> np.ndarray:
"""Vectorized log-density evaluated at sequence encoded input x. Always 0.0.
Args:
x (Optional[Any]): Sequence encoded data; NullDataEncoder returns the sequence length.
Returns:
A zero vector with one entry per encoded observation.
"""
if isinstance(x, (int, np.integer)):
return np.zeros(int(x), dtype=float)
return np.zeros(0, dtype=float)
[docs]
def backend_seq_log_density(self, x: Any | None, engine: Any) -> Any:
"""Engine-neutral vectorized log-density: zero for every encoded row."""
if isinstance(x, (int, np.integer)):
return engine.zeros(int(x))
return engine.zeros(0)
[docs]
@classmethod
def backend_stacked_params(cls, dists: tuple["NullDistribution", ...], engine: Any) -> dict[str, Any]:
"""Return stacked parameters for homogeneous null mixtures."""
return {"num_components": len(dists)}
[docs]
@classmethod
def backend_stacked_log_density(cls, x: Any | None, params: dict[str, Any], engine: Any) -> Any:
"""Return an ``(n, k)`` zero matrix for null-component log densities."""
n = int(x) if isinstance(x, (int, np.integer)) else 0
return engine.zeros((n, int(params["num_components"])))
[docs]
@classmethod
def backend_stacked_sufficient_statistics(
cls, x: Any | None, weights: Any, params: dict[str, Any], engine: Any
) -> tuple[None, ...]:
"""Return empty legacy statistics for each null component."""
return tuple(None for _ in range(int(params["num_components"])))
[docs]
def sampler(self, seed: int | None = None) -> "NullSampler":
"""Create a NullSampler object.
Args:
seed (Optional[int]): Seed for random number generator (unused).
Returns:
NullSampler object.
"""
return NullSampler(dist=self, seed=seed)
[docs]
def estimator(self, pseudo_count: float | None = None) -> "NullEstimator":
"""Create a NullEstimator object.
Args:
pseudo_count (Optional[float]): Kept for interface consistency (has no effect on estimation).
Returns:
NullEstimator object.
"""
if pseudo_count is None:
return NullEstimator(name=self.name)
else:
return NullEstimator(pseudo_count=pseudo_count, name=self.name)
[docs]
def dist_to_encoder(self) -> "NullDataEncoder":
"""Returns a NullDataEncoder object for encoding sequences of data."""
return NullDataEncoder()
[docs]
def enumerator(self) -> "NullEnumerator":
"""Returns a NullEnumerator object enumerating the support of the NullDistribution."""
return NullEnumerator(self)
[docs]
def quantized_index(self, max_bits: float, bin_width_bits: float = 1.0) -> QuantizedEnumerationIndex:
"""Build the single-item bounded bit-quantized index for NullDistribution."""
return QuantizedEnumerationIndex.from_items(
[(None, 0.0)], max_bits=max_bits, bin_width_bits=bin_width_bits, sorted_items=True, truncated=False
)
[docs]
def quantized_multi_cross_index(self, others, max_bits, bin_width_bits: float = 1.0) -> QuantizedCrossIndex:
"""Build an exact aligned cross-bin view for null distributions."""
dists = [self] + list(others)
if any(not isinstance(dist, NullDistribution) for dist in dists):
return super().quantized_multi_cross_index(others, max_bits=max_bits, bin_width_bits=bin_width_bits)
return QuantizedCrossIndex.from_items(
[(None, tuple([0.0] * len(dists)))], max_bits=max_bits, bin_width_bits=bin_width_bits, truncated=False
)
[docs]
def quantized_cross_index(self, other, max_bits, bin_width_bits: float = 1.0) -> QuantizedCrossIndex:
"""Build an exact aligned cross-bin view for two null distributions."""
return self.quantized_multi_cross_index([other], max_bits=max_bits, bin_width_bits=bin_width_bits)
[docs]
class NullEnumerator(DistributionEnumerator):
"""Yields the single value None with probability one, matching NullSampler.sample()."""
def __init__(self, dist: "NullDistribution") -> None:
"""NullEnumerator object.
Args:
dist (NullDistribution): NullDistribution instance to enumerate.
"""
super().__init__(dist)
self._done = False
def __next__(self) -> tuple[None, float]:
"""Returns the single (None, 0.0) pair, then raises StopIteration."""
if self._done:
raise StopIteration
self._done = True
return (None, 0.0)
[docs]
class NullSampler(DistributionSampler):
"""Sampler for the NullDistribution. Always returns None."""
def __init__(self, dist: "NullDistribution", seed: int | None = None) -> None:
"""NullSampler object.
Args:
dist (NullDistribution): NullDistribution instance to sample from.
seed (Optional[int]): Seed for random number generator (unused).
"""
self.rng = RandomState(seed)
self.dist = dist
[docs]
def sample(self, size: int | None = None) -> None:
"""Returns None for any requested size.
Args:
size (Optional[int]): Number of samples requested (ignored).
Returns:
None.
"""
return None
[docs]
class NullAccumulator(SequenceEncodableStatisticAccumulator):
"""Accumulator for NullDistribution. Accumulates no sufficient statistics."""
def __init__(self, keys: str | None = None) -> None:
"""NullAccumulator object.
Args:
keys (Optional[str]): Optional key for merging sufficient statistics.
Attributes:
key (Optional[str]): Optional key for merging sufficient statistics.
"""
self.keys = keys
[docs]
def update(self, x: Any | None, weight: float, estimate: Optional["NullDistribution"]) -> None:
"""No-op update. Nothing is accumulated for the NullDistribution.
Args:
x (Optional[Any]): Observation of any type (ignored).
weight (float): Weight for observation (ignored).
estimate (Optional[NullDistribution]): Previous estimate (ignored).
"""
pass
[docs]
def seq_update(self, x: Any | None, weights: np.ndarray, estimate: Optional["NullDistribution"]) -> None:
"""No-op vectorized update. Nothing is accumulated for the NullDistribution.
Args:
x (Optional[Any]): Sequence encoded data (ignored).
weights (np.ndarray): Weights for observations (ignored).
estimate (Optional[NullDistribution]): Previous estimate (ignored).
"""
pass
[docs]
def seq_update_engine(self, x, weights, estimate, engine) -> None:
# NullDistribution has no parameters: accumulation is a no-op on every engine.
pass
[docs]
def initialize(self, x: Any | None, weight: float, rng: Optional["np.random.RandomState"]) -> None:
"""No-op initialization for a single observation.
Args:
x (Optional[Any]): Observation of any type (ignored).
weight (float): Weight for observation (ignored).
rng (Optional[np.random.RandomState]): Random number generator (unused).
"""
self.update(x, weight, None)
[docs]
def seq_initialize(self, x: Any | None, weights: np.ndarray, rng: np.random.RandomState) -> None:
"""No-op vectorized initialization.
Args:
x (Optional[Any]): Sequence encoded data (ignored).
weights (np.ndarray): Weights for observations (ignored).
rng (np.random.RandomState): Random number generator (unused).
"""
self.seq_update(x, weights, None)
[docs]
def combine(self, suff_stat: Any | None) -> "NullAccumulator":
"""Combine sufficient statistics (no-op).
Args:
suff_stat (Optional[Any]): Sufficient statistics (ignored).
Returns:
Self, unchanged.
"""
return self
[docs]
def value(self) -> None:
"""Returns None (the NullAccumulator has no sufficient statistics)."""
return None
[docs]
def from_value(self, x: Any | None) -> "NullAccumulator":
"""Set accumulator from sufficient statistics (no-op).
Args:
x (Optional[Any]): Sufficient statistics (ignored).
Returns:
Self, unchanged.
"""
return self
[docs]
def key_merge(self, stats_dict: dict[str, Any]) -> None:
"""Register the key in stats_dict (the NullAccumulator stores no sufficient statistics).
Args:
stats_dict (Dict[str, Any]): Dict mapping keys to shared sufficient statistics.
Returns:
None.
"""
if self.keys is not None:
if self.keys in stats_dict:
pass
else:
stats_dict[self.keys] = None
[docs]
def key_replace(self, stats_dict: dict[str, Any]) -> None:
"""No-op kept for interface consistency (the NullAccumulator stores no sufficient statistics).
Args:
stats_dict (Dict[str, Any]): Dict mapping keys to shared sufficient statistics (ignored).
Returns:
None.
"""
pass
[docs]
def acc_to_encoder(self) -> "NullDataEncoder":
"""Returns a NullDataEncoder object for encoding sequences of data."""
return NullDataEncoder()
[docs]
class NullAccumulatorFactory(StatisticAccumulatorFactory):
"""Factory for creating NullAccumulator objects."""
def __init__(self, keys: str | None = None) -> None:
"""NullAccumulatorFactory object.
Args:
keys (Optional[str]): Optional key passed to created accumulators.
Attributes:
keys (Optional[str]): Optional key passed to created accumulators.
"""
self.keys = keys
[docs]
def make(self) -> "NullAccumulator":
"""Returns a new NullAccumulator object."""
return NullAccumulator(keys=self.keys)
[docs]
class NullEstimator(ParameterEstimator):
"""Estimator that always produces a NullDistribution regardless of the data."""
def __init__(
self,
pseudo_count: float | None = None,
suff_stat: Any | None = None,
name: str | None = None,
keys: str | None = None,
) -> None:
"""NullEstimator object.
Args:
pseudo_count (Optional[float]): Kept for interface consistency (has no effect on estimation).
suff_stat (Optional[Any]): Kept for interface consistency (has no effect on estimation).
name (Optional[str]): Optional name for object instance.
keys (Optional[str]): Optional key for merging sufficient statistics.
Attributes:
pseudo_count (Optional[float]): Kept for interface consistency.
suff_stat (Optional[Any]): Kept for interface consistency.
name (Optional[str]): Optional name for object instance.
keys (Optional[str]): Optional key for merging sufficient statistics.
"""
self.pseudo_count = pseudo_count
self.suff_stat = suff_stat
self.keys = keys
self.name = name
[docs]
def accumulator_factory(self) -> "NullAccumulatorFactory":
"""Returns a NullAccumulatorFactory for creating NullAccumulator objects."""
return NullAccumulatorFactory(self.keys)
[docs]
def estimate(self, nobs: float | None, suff_stat: Any | None = None) -> "NullDistribution":
"""Returns a NullDistribution; arguments are ignored.
Args:
nobs (Optional[float]): Number of observations (ignored).
suff_stat (Optional[Any]): Sufficient statistics (ignored).
Returns:
NullDistribution object.
"""
return NullDistribution(name=self.name)
[docs]
class NullDataEncoder(DataSequenceEncoder):
"""Data encoder for the NullDistribution. Encodes any sequence as its length."""
def __str__(self) -> str:
"""Returns string representation of NullDataEncoder object."""
return "NullDataEncoder"
def __eq__(self, other) -> bool:
"""Checks if other object is an instance of a NullDataEncoder.
Args:
other (object): Object to compare against.
Returns:
True if other is a NullDataEncoder instance, else False.
"""
return isinstance(other, NullDataEncoder)
[docs]
def seq_encode(self, x: Any) -> int:
"""Encode a sequence of observations as its length.
Args:
x (Any): Sequence of observations of any type (ignored).
Returns:
Number of observations in the sequence.
"""
return len(x)