"""
Standard variable implementation with clear separation of
deterministic values and stochastic noise.
This module provides the Variable class, which extends BaseVariable
with multi-source value resolution capabilities. The implementation
follows scientific convention by clearly separating deterministic
variable values from stochastic noise components. This allows users to maintain
causal integrity while enabling flexible stochastic modeling when needed.
Key concepts:
- Deterministic value: The resolved value from multiple sources without noise
- Noise model: A stochastic process that can perturb the deterministic value
- RNG: Explicit random number generator for reproducible stochasticity
- Separation: Noise is applied separately, not baked into the variable's intrinsic value
"""
from __future__ import annotations
from typing import Optional, Any, Dict, List, Tuple
import numpy as np
from .base import BaseVariable, Domain
from .types import Distribution
[docs]
class Variable(BaseVariable):
"""
Multi-source variable with deterministic resolution and optional stochastic noise.
Variable extends BaseVariable with the ability to receive values from multiple sources
(e.g., mechanisms, processes, measurements) and provides configurable strategies for
resolving conflicts. Following scientific convention, deterministic values and stochastic
noise are clearly separated:
1. Deterministic value: Computed from source values using weighted averaging
2. Noise model: Optional stochastic perturbation applied separately
3. RNG: Explicit random number generator for reproducible noise
This separation ensures causal integrity while enabling flexible stochastic modeling.
Users can choose when and how to apply noise based on their specific requirements.
Parameters
----------
name : str
Unique identifier for the variable (e.g., "temperature", "glucose_level").
domain : Domain
Mathematical domain defining valid values (e.g., Interval, Categorical).
value : Any, optional
Initial deterministic value. Must satisfy domain constraints if provided.
units : str, optional
Physical or logical units (e.g., "°C", "mg/dL", "count").
noise : Distribution, optional
Noise model implementing the Distribution protocol.
Signature: noise(rng: np.random.Generator, n: int) -> np.ndarray
rng : np.random.Generator, optional
Random number generator for reproducible stochasticity.
Required if noise is specified.
Attributes
----------
noise : Optional[Distribution]
Noise model for stochastic perturbations.
rng : Optional[np.random.Generator]
Random number generator for noise sampling.
_sources : List[str]
List of registered source identifiers.
_weights : Dict[str, float]
Source weights for weighted averaging.
_source_values : Dict[str, Any]
Current values provided by each source.
Raises
------
ValueError
If initial value violates domain constraints, or if noise is specified
without providing an RNG.
Examples
--------
>>> from causaloop import Variable, Interval
>>> import numpy as np
>>> # Create deterministic variable
>>> var = Variable("temperature", Interval(0, 100), 25.0, "°C")
>>> var.value # Deterministic value
25.0
>>> # Create variable with noise model
>>> def gaussian_noise(rng: np.random.Generator, n: int) -> np.ndarray:
... return rng.normal(loc=0.0, scale=0.1, size=n)
>>>
>>> rng = np.random.default_rng(seed=42) # For reproducibility
>>> noisy_var = Variable("signal", Interval(-1, 1), 0.0,
... noise=gaussian_noise, rng=rng)
>>>
>>> # Deterministic value (always the same)
>>> noisy_var.value
0.0
>>>
>>> # Apply noise (different each call due to RNG state)
>>> noisy_var.apply_noise()
(0.0, 0.03..)
>>>
>>> # Sample noise separately
>>> noise_samples = noisy_var.sample_noise(n=5)
>>> noise_samples.shape
(5,)
See Also
--------
BaseVariable : Parent class with domain validation.
CausalVariable : Extension with temporal causal tracking.
"""
def __init__(
self,
name: str,
domain: Domain,
value: Optional[Any] = None,
units: Optional[str] = None,
noise: Optional[Distribution] = None,
rng: Optional[np.random.Generator] = None,
):
"""
Initialize a Variable with clear separation of deterministic value and noise.
Parameters
----------
name : str
Unique identifier for the variable. Should be descriptive and follow
domain-specific naming conventions.
domain : Domain
Domain object defining the mathematical space of valid values.
Determines validation rules and influences type inference.
value : Any, optional
Initial deterministic value. Must satisfy domain constraints.
If None, variable starts in uninitialized state.
units : str, optional
Physical or logical units for interpreting the variable's value.
Used for dimensional analysis, visualization, and documentation.
noise : Distribution, optional
Noise model implementing the Distribution protocol. The model should
generate noise samples when called with (rng, n) arguments.
Common examples: Gaussian noise, uniform noise, Poisson noise.
rng : np.random.Generator, optional
Random number generator for reproducible stochastic sampling.
Must be provided if noise is specified. Use seeded RNGs for
reproducible research.
Raises
------
ValueError
If the provided value violates domain constraints, or if noise
is specified without an accompanying RNG.
Examples
--------
>>> from causaloop import Variable, Interval
>>> import numpy as np
>>> rng = np.random.default_rng(seed=42) # Reproducible
>>> def noise(rng: np.random.Generator, n: int) -> np.ndarray:
... return rng.normal(loc=0.0, scale=0.1, size=n)
>>> var = Variable("measurement", Interval(0, 10), noise=noise, rng=rng)
>>> var.sample_noise(2)
array([ 0.03..., -0.10...])
Notes
-----
The initialization enforces the separation principle: deterministic
values are stored and validated immediately, while noise models are
configured for optional later use. This prevents accidental mixing
of deterministic and stochastic components.
For reproducible research, always provide a seeded RNG when using
noise models:
"""
super().__init__(name, domain, value, units)
self.noise = noise
self.rng = rng
if noise is not None and rng is None:
raise ValueError(
"Noise model requires an explicit random number generator (RNG). "
"Provide an RNG instance when specifying noise."
)
self._sources: List[str] = []
self._weights: Dict[str, float] = {}
self._source_values: Dict[str, Any] = {}
@property
def has_noise(self) -> bool:
"""
Check if the variable has a configured noise model.
Returns
-------
bool
True if a noise model and RNG are configured, False otherwise.
Examples
--------
>>> from causaloop import Variable, Interval
>>> import numpy as np
>>> var = Variable("signal", Interval(-1, 1))
>>> var.has_noise
False
>>> noisy_var = Variable("noisy_signal", Interval(-1, 1),
... noise=lambda rng, n: rng.normal(0, 0.1, n),
... rng=np.random.default_rng())
>>> noisy_var.has_noise
True
Notes
-----
This property checks both the presence of a noise model and a valid
RNG. A variable with a noise model but no RNG is considered invalid
and will raise an error during initialization.
"""
return self.noise is not None and self.rng is not None
[docs]
def add_source(self, source: str, weight: float = 1.0) -> Variable:
"""
Register a potential source for this variable.
Parameters
----------
source : str
Unique identifier for the source (e.g., "metabolism", "sensor_A",
"process_1"). Should be descriptive and consistent.
weight : non-negative float, default=1.0
Relative weight of this source in weighted averaging.
Higher weights give the source more influence. Weights are
relative, not normalized.
Returns
-------
Variable
Self for method chaining.
Raises
------
ValueError
If weight is negative.
Examples
--------
>>> from causaloop import Variable, Interval
>>> var = Variable("output", Interval(0, 100))
>>> var.add_source("process_A", weight=2.0)
Variable('output', value=None, sources=1)
>>> var.add_source("process_B", weight=1.0)
Variable('output', value=None, sources=2)
>>> var.get_source_weight("process_A")
2.0
Notes
-----
Sources can be registered before they provide values, allowing
pre-configuration of the variable's dependency structure. This is
useful for declaring expected influences in complex systems.
Weights are used in the default "weighted" resolution strategy:
weighted_sum = Σ(weight_i * value_i)
total_weight = Σ(weight_i)
result = weighted_sum / total_weight
Weight values are relative: a source with weight 2.0 has twice the
influence of a source with weight 1.0.
"""
if weight < 0:
raise ValueError("Weights should be non-negative")
if source not in self._sources:
self._sources.append(source)
self._weights[source] = weight
return self
[docs]
def register_output(self, source: str, value: Any) -> None:
"""
Register a deterministic value from a specific source.
Parameters
----------
source : str
Source identifier that generated this value.
value : Any
The deterministic value to register. Must satisfy all domain
constraints.
Raises
------
ValueError
If the value violates domain constraints.
Examples
--------
>>> from causaloop import Variable, Interval
>>> var = Variable("glucose", Interval(50, 300))
>>> var.add_source("metabolism", weight=1.0)
Variable('glucose', value=None, sources=1)
>>> var.register_output("metabolism", 95.0)
>>> var.value # Deterministic value
95.0
>>> # Multiple sources with different weights
>>> var.add_source("diet", weight=0.5)
Variable('glucose', value=95.0, sources=2)
>>> var.register_output("diet", 105.0)
>>> var.value # Weighted average: (1.0*95 + 0.5*105) / 1.5
98.333...
Notes
-----
This method updates the variable's deterministic value immediately
based on all registered source values. The computation is purely
deterministic and does not involve any stochastic components.
The method performs full domain validation to ensure mathematical
consistency. If validation fails, the variable's state remains
unchanged.
Sources are automatically registered if not already present (with
default weight 1.0). This allows flexible usage patterns.
"""
if not self.domain.validate(value):
raise ValueError(
f"Value {value!r} from source '{source}' violates domain "
f"constraints: {self.domain}"
)
# Auto-register source if not already registered
if source not in self._sources:
self.add_source(source)
self._source_values[source] = value
# Update deterministic value
self._value = self._resolve_deterministic()
def _resolve_deterministic(self, strategy: str = "weighted") -> Any:
"""
Internal method to compute deterministic value from multiple sources.
Parameters
----------
strategy : str, default="weighted"
Resolution strategy for handling multiple source values:
- "weighted": Weighted average (default, recommended for most cases)
- "latest": Most recently registered value
- "priority": Value from source with highest weight
Returns
-------
Any
Deterministic value according to the specified strategy.
Raises
------
ValueError
If an unknown resolution strategy is specified.
Notes
-----
This method implements pure deterministic computation. No stochastic
components are involved, ensuring causal consistency.
The weighted strategy computes:
result = Σ(weight_i * numeric_value_i) / Σ(weight_i)
For non-numeric values or conversion failures, the method falls back
to the "latest" strategy to maintain consistency.
If no sources have provided values, returns the base value (which may
be None for uninitialized variables).
This method is internal; users should access values through the
public `value` property or `resolve()` method.
"""
if not self._source_values:
return self._value
if strategy == "weighted":
total_weight = 0.0
weighted_sum = 0.0
for src, val in self._source_values.items():
weight = self._weights.get(src, 1.0)
try:
# Attempt numeric conversion for weighted average
weighted_sum += float(val) * weight
total_weight += weight
except (ValueError, TypeError):
# Non-numeric value: fall back to latest strategy
return list(self._source_values.values())[-1]
if total_weight == 0:
return None
return weighted_sum / total_weight
elif strategy == "latest":
# Most recently registered value (last in insertion order)
return list(self._source_values.values())[-1]
elif strategy == "priority":
# Value from source with highest weight
best_source = max(
self._source_values.keys(),
key=lambda s: self._weights.get(s, 0.0)
)
return self._source_values[best_source]
else:
raise ValueError(
f"Unknown resolution strategy: {strategy!r}. "
f"Valid options: 'weighted', 'latest', 'priority'"
)
@property
def value(self) -> Any:
"""
Get the deterministic value of the variable.
Returns
-------
Any
Deterministic value resolved from all registered source values
using the default "weighted" strategy.
Notes
-----
This property returns the pure deterministic value without any
stochastic perturbations. It represents the variable's "true" value
in the causal model, free from measurement error or process noise.
The value is recomputed on each access to reflect the current state
of all source values. This ensures consistency in dynamic systems.
For variables with noise models, this property does NOT apply noise.
Use `apply_noise()` or `sample_noise()` for stochastic values.
Examples
--------
>>> from causaloop import Variable, Interval
>>> var = Variable("concentration", Interval(0, 100))
>>> var.register_output("source1", 30.0)
>>> var.register_output("source2", 40.0)
>>> var.value # Weighted average: (30 + 40) / 2
35.0
>>> # Even with noise model, value is deterministic
>>> noisy_var = Variable("measurement", Interval(0, 10),
... noise=..., rng=...)
>>> noisy_var.register_output("sensor", 5.0)
>>> noisy_var.value # Still 5.0, not 5.0 + noise
5.0
"""
return self._resolve_deterministic()
[docs]
def resolve(self, strategy: str = "weighted") -> Any:
"""
Get deterministic value using specified resolution strategy.
Parameters
----------
strategy : str, default="weighted"
Resolution strategy for handling multiple source values:
- "weighted": Weighted average (recommended for most applications)
- "latest": Most recently registered value
- "priority": Value from source with highest weight
Returns
-------
Any
Deterministic value according to the specified strategy.
Raises
------
ValueError
If an unknown resolution strategy is specified.
Examples
--------
>>> from causaloop import Variable, Interval
>>> var = Variable("score", Interval(0, 100))
>>> var.register_output("judge1", 85)
>>> var.register_output("judge2", 90)
>>> var.add_source("judge1", weight=2.0)
Variable('score', value=86.66..., sources=2)
>>> # Different resolution strategies
>>> var.resolve("weighted") # (2*85 + 1*90) / 3
86.666...
>>> var.resolve("latest") # Most recent value
90
>>> var.resolve("priority") # Judge1 has higher weight
85
Notes
-----
This method provides explicit control over the resolution strategy,
which can be important in specific modeling contexts:
- "weighted": Appropriate when sources have different reliabilities
- "latest": Useful for temporal sequences or state updates
- "priority": When certain sources should dominate others
All strategies produce deterministic results. For stochastic
perturbations, use `apply_noise()` on the resolved value.
"""
return self._resolve_deterministic(strategy)
[docs]
def sample_noise(self, n: int = 1) -> np.ndarray:
"""
Sample noise from the configured noise model.
Parameters
----------
n : int, default=1
Number of independent noise samples to generate.
Returns
-------
np.ndarray
Array of noise samples with shape (n,).
Returns zeros if no noise model is configured.
Raises
------
RuntimeError
If noise model is configured but RNG is not available.
Examples
--------
>>> from causaloop import Variable, Interval
>>> import numpy as np
>>> def gaussian_noise(rng, n):
... return rng.normal(0, 0.1, n)
>>>
>>> rng = np.random.default_rng(seed=42)
>>> var = Variable("signal", Interval(-1, 1),
... noise=gaussian_noise, rng=rng)
>>>
>>> # Single noise sample
>>> var.sample_noise()
array([0.03...])
>>>
>>> # Multiple samples
>>> print(var.sample_noise(5))
[-0.10... 0.07... 0.09... -0.19... -0.13...]
>>>
>>> # Variable without noise model
>>> deterministic_var = Variable("constant", Interval(0, 1), 0.5)
>>> deterministic_var.sample_noise(3) # Returns zeros
array([0., 0., 0.])
Notes
-----
This method samples noise independently of the variable's current
value. The noise samples can be applied to any value using addition:
>>> value = 5.0
>>> noise = var.sample_noise()
>>> noisy_value = value + float(noise[0])
For variables without noise models, this method returns zeros,
allowing uniform handling of both deterministic and stochastic
variables in algorithms.
The RNG state advances with each call, ensuring proper stochastic
behavior in sequential sampling.
"""
if not self.has_noise:
return np.zeros(n)
try:
return self.noise(self.rng, n)
except Exception as e:
raise RuntimeError(
f"Failed to sample noise from model for variable '{self.name}': {e}"
)
[docs]
def apply_noise(self, value: Optional[Any] = None) -> Tuple[Any, Any]:
"""
Apply noise to a value, returning both deterministic and noisy results.
Parameters
----------
value : Any, optional
Value to apply noise to. If None, uses the variable's current
deterministic value.
Returns
-------
Tuple[Any, Any]
Tuple containing (deterministic_value, noisy_value).
If no noise model is configured, returns (value, value).
Raises
------
ValueError
If the provided value violates domain constraints.
Examples
--------
>>> from causaloop import Variable, Interval
>>> import numpy as np
>>> def uniform_noise(rng, n):
... return rng.uniform(-0.1, 0.1, n)
>>>
>>> rng = np.random.default_rng(seed=42)
>>> var = Variable("measurement", Interval(0, 10),
... noise=uniform_noise, rng=rng)
>>> var.register_output("sensor", 5.0)
>>>
>>> # Apply noise to current value
>>> deterministic, noisy = var.apply_noise()
>>> deterministic
5.0
>>> noisy # 5.0 + noise
5.05...
>>>
>>> # Apply noise to specific value
>>> var.apply_noise(7.5)
(7.5, 7.48...)
>>>
>>> # Deterministic variable
>>> det_var = Variable("constant", Interval(0, 1), 0.5)
>>> det_var.apply_noise() # No noise model
(0.5, 0.5)
Notes
-----
This method follows the scientific convention of clearly separating
deterministic values from stochastic perturbations. It always returns
both components, allowing users to:
1. Track the underlying deterministic signal
2. Apply domain-specific handling of noisy values
3. Compute error statistics (noise = noisy - deterministic)
4. Implement custom noise handling (clipping, rejection, etc.)
The method validates that the input value satisfies domain constraints
but does NOT validate the noisy result. Users are responsible for
handling cases where noise pushes values outside valid domains.
This design supports flexible modeling strategies:
- Measurement error: deterministic = true value, noisy = observed
- Process noise: deterministic = expected, noisy = realized
- Stochastic processes: deterministic = mean, noisy = sample
"""
# Use provided value or current deterministic value
target_value = value if value is not None else self.value
# Validate the target value
if target_value is not None and not self.domain.validate(target_value):
raise ValueError(
f"Value {target_value!r} violates domain constraints: {self.domain}"
)
# Return both values (identical if no noise)
if not self.has_noise or target_value is None:
return (target_value, target_value)
try:
# Sample and apply noise
noise_sample = float(self.sample_noise(1)[0])
# Ensure numeric types for addition
try:
numeric_value = float(target_value)
noisy_value = numeric_value + noise_sample
except (ValueError, TypeError):
# Non-numeric value: cannot apply additive noise
return (target_value, target_value)
return (target_value, noisy_value)
except Exception as e:
# Noise application failed
import warnings
warnings.warn(
f"Noise application failed for variable '{self.name}': {e}. "
"Returning deterministic value only."
)
return (target_value, target_value)
[docs]
def get_deterministic_and_noisy(self) -> Tuple[Any, Any]:
"""
Get both deterministic value and noisy version in one call.
Returns
-------
Tuple[Any, Any]
Tuple containing (deterministic_value, noisy_value).
Convenience method equivalent to apply_noise(None).
Examples
--------
>>> from causaloop import Variable, Interval
>>> var = Variable("signal", Interval(-1, 1), noise=..., rng=...)
>>> var.register_output("source", 0.5)
>>> det, noisy = var.get_deterministic_and_noisy()
>>> det # Deterministic value
0.5
Notes
-----
This is a convenience method for the common pattern of wanting both
the deterministic value and a noisy version. It's equivalent to:
>>> det, noisy = var.apply_noise(var.value)
The method always returns a tuple, even for deterministic variables
(where both elements are identical).
"""
return self.apply_noise(self.value)
[docs]
def clear_sources(self) -> None:
"""
Clear all source values while preserving source registrations.
Examples
--------
>>> from causaloop import Variable, Interval
>>> var = Variable("temperature", Interval(0, 100))
>>> var.register_output("sensor1", 25.0)
>>> var.register_output("sensor2", 26.0)
>>> var.value
25.5
>>> var.clear_sources()
>>> var.value is None
True
>>> var._sources # Sources still registered
['sensor1', 'sensor2']
>>> var._source_values # Values cleared
{}
Notes
-----
This method clears only the current values from sources, not the
source registrations themselves. This is useful for:
- Resetting variable state between simulation runs
- Clearing temporary measurements while keeping configurations
- Reusing variable structures with different input data
Source weights and noise models remain unchanged. To completely
reset a variable, create a new instance.
"""
self._source_values.clear()
self._value = None
[docs]
def get_source_value(self, source: str) -> Optional[Any]:
"""
Get the current value from a specific source.
Parameters
----------
source : str
Source identifier.
Returns
-------
Optional[Any]
Current deterministic value from the specified source,
or None if the source hasn't provided a value.
Examples
--------
>>> from causaloop import Variable, Interval
>>> var = Variable("pressure", Interval(0, 100))
>>> var.register_output("sensor_A", 85.5)
>>> var.get_source_value("sensor_A")
85.5
>>> var.get_source_value("sensor_B") is None # Not provided
True
Notes
-----
This method retrieves the raw value provided by a specific source,
before any resolution or noise application. It's useful for:
- Debugging source contributions
- Implementing custom resolution logic
- Monitoring individual component outputs
- Validating source behavior
"""
return self._source_values.get(source)
[docs]
def get_source_weight(self, source: str) -> float:
"""
Get the weight of a registered source.
Parameters
----------
source : str
Source identifier.
Returns
-------
float
Current weight of the source.
Returns 0.0 if source is not registered.
Examples
--------
>>> from causaloop import Variable, Interval
>>> var = Variable("output", Interval(0, 100))
>>> var.add_source("process1", weight=2.0)
Variable('output', value=None, sources=1)
>>> var.get_source_weight("process1")
2.0
>>> var.get_source_weight("unknown_process") # Not registered
0.0
Notes
-----
Source weights determine influence in weighted averaging. A weight
of 0.0 effectively disables a source's contribution (unless using
other strategy).
Weights are relative, not normalized. Changing one source's weight
doesn't automatically adjust others.
"""
return self._weights.get(source, 0.0)
[docs]
def set_source_weight(self, source: str, weight: float) -> None:
"""
Set or update the weight of a source.
Parameters
----------
source : str
Source identifier.
weight : non-negative float
New weight for the source.
Raises
------
ValueError
If weight is negative.
KeyError
If source is not registered.
Examples
--------
>>> from causaloop import Variable, Interval
>>> var = Variable("output", Interval(0, 100))
>>> var.add_source("process1", weight=1.0)
Variable('output', value=None, sources=1)
>>> var.set_source_weight("process1", 3.0)
>>> var.get_source_weight("process1")
3.0
Notes
-----
Changing source weights affects future value resolution but doesn't
change the variable's current value. The new weight takes effect on
the next value computation.
To immediately update the variable's value with new weights, call
`register_output` again or access the `value` property.
"""
if weight < 0:
raise ValueError(
f"Weight for source {source} is negative. Weights should be always non-negative."
)
if source not in self._sources:
raise KeyError(
f"Source '{source}' not registered. Use add_source() first."
)
self._weights[source] = weight
# Recompute value if this source has a current value
if source in self._source_values:
self._value = self._resolve_deterministic()
@property
def sources(self) -> List[str]:
"""
Get list of registered sources.
Returns
-------
List[str]
List of source identifiers in registration order.
Examples
--------
>>> from causaloop import Variable, Interval
>>> var = Variable("temperature", Interval(0, 100))
>>> var.add_source("indoor_sensor")
Variable('temperature', value=None, sources=1)
>>> var.add_source("outdoor_sensor")
Variable('temperature', value=None, sources=2)
>>> var.sources
['indoor_sensor', 'outdoor_sensor']
Notes
-----
The order of sources in this list reflects registration order, which
may be relevant for "latest" resolution strategy or for understanding
the variable's dependency structure.
"""
return self._sources.copy()
@property
def active_sources(self) -> List[str]:
"""
Get list of sources that have provided current values.
Returns
-------
List[str]
List of source identifiers that have non-None values.
Examples
--------
>>> from causaloop import Variable, Interval
>>> var = Variable("score", Interval(0, 100))
>>> var.add_source("judge1")
Variable('score', value=None, sources=1)
>>> var.add_source("judge2")
Variable('score', value=None, sources=2)
>>> var.register_output("judge1", 85)
>>> var.active_sources
['judge1']
Notes
-----
This property helps distinguish between potential influences (all
registered sources) and actual current influences (active sources).
Inactive sources (registered but not providing values) don't affect
the variable's current deterministic value.
"""
return [s for s in self._sources if s in self._source_values]
def __repr__(self) -> str:
"""
Return string representation of a Variable.
Returns
-------
str
String showing class name, variable name, deterministic value,
number of sources, and noise indicator.
Examples
--------
>>> from causaloop import Variable, Interval
>>> var = Variable("temperature", Interval(0, 100), 25.0)
>>> repr(var)
"Variable('temperature', value=25.0, sources=0)"
>>> noisy_var = Variable("signal", Interval(-1, 1), 0.0,
... noise=..., rng=...)
>>> repr(noisy_var)
"Variable('signal', value=0.0, sources=0, +noise)"
"""
src_count = len(self._sources)
noise_indicator = " +noise" if self.has_noise else ""
return (
f"{self.__class__.__name__}("
f"{self.name!r}, value={self.value}, "
f"sources={src_count}{noise_indicator})"
)
def __str__(self) -> str:
"""
Return human-readable string representation.
Returns
-------
str
Formatted string with name, deterministic value, units, type,
source count, and noise indicator.
Examples
--------
>>> from causaloop import Variable, Interval
>>> var = Variable("glucose", Interval(50, 300), 95.0, "mg/dL")
>>> var.register_output("metabolism", 95.0)
>>> str(var)
'glucose = 95.0 mg/dL (continuous) [1/1 source(s)]'
>>> noisy_var = Variable("measurement", Interval(0, 10), 5.0, "V",
... noise=..., rng=...)
>>> str(noisy_var)
'measurement = 5.0 V (continuous) [0/0 source(s) +noise]'
"""
base_str = super().__str__()
active_count = len(self.active_sources)
total_count = len(self._sources)
noise_indicator = " +noise" if self.has_noise else ""
source_info = f" [{active_count}/{total_count} source(s){noise_indicator}]"
return base_str + source_info
[docs]
def to_dict(self) -> Dict:
"""
Convert variable to detailed dictionary representation.
Returns
-------
Dict
Dictionary containing variable metadata, source information,
deterministic value, and noise configuration.
Examples
--------
>>> from causaloop import Variable, Interval
>>> var = Variable("output", Interval(0, 100))
>>> var.add_source("proc1", weight=2.0)
Variable('output', value=None, sources=1)
>>> var.add_source("proc2", weight=1.0)
Variable('output', value=None, sources=2)
>>> var.register_output("proc1", 75.0)
>>> var.to_dict()
{
'name': 'output',
'type': 'continuous',
'value': 75.0,
'units': None,
'domain': 'Interval(0, 100)',
'sources': ['proc1', 'proc2'],
'weights': {'proc1': 2.0, 'proc2': 1.0},
'source_values': {'proc1': 75.0},
'has_noise': False
}
Notes
-----
This representation is suitable for serialization (JSON, YAML),
persistence, and configuration. It captures all necessary information
to reconstruct the variable's state, though noise models and RNGs
may require special handling for complete serialization.
The dictionary includes both structural information (sources, weights)
and current state (value, source_values).
"""
base_dict = super().to_dict()
base_dict.update({
"sources": self._sources.copy(),
"weights": self._weights.copy(),
"source_values": self._source_values.copy(),
"has_noise": self.has_noise
})
return base_dict