"""Continuous States Descriptor.
Provides piecewise-linear and piecewise-quadratic trajectory tracking, and
declarative threshold monitors natively integrated with signals and
the standard Modelly event queue.
"""
from __future__ import annotations
from collections.abc import Callable
from typing import TYPE_CHECKING, Any
import modelly.time.signals.core as core
from modelly.behaviour.states.threshold import Threshold
from modelly.time.signals import Observable, ObservableSignals
from modelly.time.signals.core import (
BaseObservable,
ComputedState,
_hashable_signal,
)
if TYPE_CHECKING:
from modelly.time.signals import HasEmitters
def _bind_thresholds(instance: HasEmitters) -> None:
"""Bind every declarative Threshold on the instance's class hierarchy."""
cls = instance.__class__
for klass in cls.__mro__:
if "_continuous_thresholds" in klass.__dict__:
for t_name in klass.__dict__["_continuous_thresholds"]:
thresh = getattr(cls, t_name, None)
if thresh is not None and hasattr(thresh, "bind"):
thresh.bind(instance)
[docs]
class ContinuousState(BaseObservable):
"""A descriptor that tracks and extrapolates continuous states over time.
Uses piecewise-linear interpolation by default: value(t) = base_value +
current_rate * (t - last_time). When the rate changes, the current value
is snapshotted as the new baseline so the trajectory remains mathematically continuous.
Chained states (e.g. position' = speed, where speed is itself a ContinuousState)
are extrapolated with the next order term instead:
value(t) = base_value + current_rate*dt + 0.5*second_order_rate*dt**2.
"""
signal_types = ObservableSignals
def __init__(
self,
fallback_value: float = 0.0,
rate: float | Callable[[Any], float] = 0.0,
) -> None:
"""Initialize a ContinuousState.
Args:
fallback_value: Initial value for the state (default 0.0).
rate: Either a constant float rate-of-change, or a callable f(agent) -> float
that is evaluated reactively via the diffing engine.
"""
super().__init__(fallback_value=fallback_value)
self.fallback_value = fallback_value
self._rate_input = rate
def __set_name__(self, owner: type[HasEmitters], name: str) -> None:
"""Bind the descriptor to the class and create shadow observables."""
super().__set_name__(owner, name)
self.private_state_name = f"_{name}_continuous_state"
self.rate_name = f"{name}_rate"
# Guard against collisions in both the dynamic dict and memory-optimized slots
collision = self.rate_name in owner.__dict__
if hasattr(owner, "__slots__"):
collision = collision or (self.rate_name in owner.__slots__)
if collision:
raise AttributeError(
f"Namespace collision: '{self.rate_name}' is a reserved shadow attribute "
f"generated by the ContinuousState '{name}'. Please rename your attribute."
)
# Use Observable (not BaseObservable) so that values are actually stored
# on the instance, and old/new in CHANGED notifications are correct.
rate_descriptor = Observable()
rate_descriptor.__set_name__(owner, self.rate_name)
setattr(owner, self.rate_name, rate_descriptor)
def _get_state(self, instance: HasEmitters) -> dict[str, Any]:
"""Retrieve or initialize the internal state tracking dictionary."""
if not hasattr(instance, self.private_state_name):
self._init_state(instance)
return getattr(instance, self.private_state_name)
def _init_state(self, instance: HasEmitters) -> None:
"""Create the tracking dictionary, binding Thresholds and the rate first."""
# Automatically bind any declarative Thresholds the first time
# a continuous state initializes, entirely bypassing the need for a scheduler.
if not getattr(instance, "_thresholds_bound", False):
instance._thresholds_bound = True
_bind_thresholds(instance)
time = instance.model.peek("time", 0.0)
state: dict[str, Any] = {
"base_value": self.fallback_value,
"last_time": time,
"rate_computed": None,
"current_rate": 0.0,
# Rate-of-the-rate, captured when current_rate's source is
# itself a ContinuousState. 0.0 when the rate is constant.
"second_order_rate": 0.0,
# Strong reference so Modelly's weakref GC doesn't silently drop the handler.
"_rate_handler": None,
}
setattr(instance, self.private_state_name, state)
if callable(self._rate_input):
self._wire_rate(instance, state)
else:
state["current_rate"] = float(self._rate_input)
def _wire_rate(self, instance: HasEmitters, state: dict[str, Any]) -> None:
"""Track a callable rate reactively and snapshot the trajectory on change."""
def compute_rate(agent):
return self._rate_input(agent)
rate_computed = ComputedState(instance, self.rate_name, compute_rate)
state["rate_computed"] = rate_computed
# Defer the first rate evaluation: the rate lambda may access other
# Observable attributes whose private backing stores don't exist yet because
# we are still inside Agent.__init__ -> register_agent -> bind().
def on_rate_change(message: Any) -> None:
self._snapshot_trajectory(instance)
state["_rate_handler"] = on_rate_change
instance.observe(
self.rate_name, ObservableSignals.CHANGED, state["_rate_handler"]
)
def _find_chained_rate(
self, comp_state: ComputedState, instance: HasEmitters
) -> float:
"""Return the rate-of-the-rate if this state's rate depends on another ContinuousState.
Scans the dependencies the rate callable's ComputedState recorded for one
that is a ContinuousState descriptor on the instance's class.
"""
for parent, attrs in comp_state.parents.items():
for attr_name in attrs:
descriptor = getattr(type(parent), attr_name, None)
if isinstance(descriptor, ContinuousState):
return descriptor.get_rate(parent)
return 0.0
def _extrapolate(self, state: dict[str, Any], dt: float) -> float:
"""Project the value forward by dt using the cached base and rates."""
return (
state["base_value"]
+ state["current_rate"] * dt
+ 0.5 * state["second_order_rate"] * dt * dt
)
def _snapshot_trajectory(self, instance: HasEmitters) -> None:
"""Commit the current extrapolated value as the new baseline when the rate changes."""
# Call order matters:
# 1. Extrapolate using the *old* cached rate(s) — prevents retroactive time-travel.
# 2. Lock in the new baseline and timestamp.
# 3. Evaluate and cache the *new* rates.
# 4. Notify Threshold observers that the trajectory has shifted.
state = self._get_state(instance)
current_time = instance.model.peek("time", 0.0)
dt = current_time - state["last_time"]
current_value = self._extrapolate(state, dt)
state["base_value"] = current_value
state["last_time"] = current_time
self._refresh_rate_if_dirty(state, instance)
# Do NOT go through the batching aggregator path here, as old==new would be swallowed.
signal = core.Message(
name=self.public_name,
owner=instance,
signal_type=ObservableSignals.CHANGED,
additional_kwargs={"old": current_value, "new": current_value},
)
instance._modelly_notify(signal)
def _is_uninitialized_observable_race(
self, instance: HasEmitters, exc: AttributeError
) -> bool:
"""Return True if the AttributeError looks like the known init-ordering race."""
missing_name = getattr(exc, "name", None)
if missing_name is None or not missing_name.startswith("_"):
return False
public_name = missing_name[1:]
observables = getattr(type(instance), "observables", None)
if not observables or public_name not in observables:
return False
descriptor = getattr(type(instance), public_name, None)
return (
isinstance(descriptor, BaseObservable)
and descriptor.private_name == missing_name
)
def _evaluate_rate(self, instance: HasEmitters) -> float:
"""Evaluate the rate callable using diffing engine to track dependencies."""
comp_state: ComputedState = self._get_state(instance)["rate_computed"]
try:
comp_state.evaluate()
except AttributeError as exc:
# The rate callable depends on an Observable that has not been assigned yet.
# This happens when bind() runs eager recalculation during Agent.__init__.
# Treat this as rate unknown, fall back to the "at rest" assumption (0.0).
if not self._is_uninitialized_observable_race(instance, exc):
raise
return 0.0
return float(comp_state.value)
def _refresh_rate_if_dirty(
self, state: dict[str, Any], instance: HasEmitters
) -> None:
"""Re-evaluate the rate and second-order rate together so they never drift."""
comp_state = state.get("rate_computed")
if comp_state is not None and comp_state.is_dirty:
state["current_rate"] = self._evaluate_rate(instance)
state["second_order_rate"] = self._find_chained_rate(comp_state, instance)
[docs]
def get_rate(self, instance: HasEmitters) -> float:
"""Return the current rate for the instance, evaluating lazily if needed.
Args:
instance (HasEmitters): The instance being evaluated.
Returns:
float: The currently active rate.
"""
state = self._get_state(instance)
self._refresh_rate_if_dirty(state, instance)
return state["current_rate"]
def __get__(self, instance: HasEmitters, owner: type | None = None) -> Any:
"""Extrapolate and return the continuous state."""
if instance is None:
return self
state = self._get_state(instance)
self._refresh_rate_if_dirty(state, instance)
current_time = instance.model.peek("time", 0.0)
dt = current_time - state["last_time"]
current_value = self._extrapolate(state, dt)
if core.CURRENT_COMPUTED is not None:
core.CURRENT_COMPUTED._record_access(
instance, self.public_name, current_value
)
core.PROCESSING_SIGNALS.add(_hashable_signal(instance, self.public_name))
return current_value
def __set__(self, instance: HasEmitters, value: float) -> None:
"""Assign a new mathematical baseline to the state."""
if (
core.CURRENT_COMPUTED is not None
and _hashable_signal(instance, self.public_name) in core.PROCESSING_SIGNALS
):
raise ValueError("Cyclical dependency detected in ContinuousState.")
state = self._get_state(instance)
current_time = instance.model.peek("time", 0.0)
has_subscribers = instance._has_subscribers(
self.public_name, ObservableSignals.CHANGED
)
old_value = self.__get__(instance) if has_subscribers else None
state["base_value"] = float(value)
state["last_time"] = current_time
if has_subscribers:
instance.notify(
self.public_name,
ObservableSignals.CHANGED,
old=old_value,
new=float(value),
)
core.PROCESSING_SIGNALS.discard(
_hashable_signal(instance, self.public_name)
)
__all__ = ["ContinuousState", "Threshold"]
