"""
Causal variable implementation with temporal consistency and counterfactual reasoning.
This module provides the CausalVariable class, which extends Variable with
causal-temporal tracking capabilities. CausalVariable maintains a complete
history of updates with causal metadata, enabling temporal consistency
enforcement, counterfactual reasoning, and causal pathway analysis.
Key concepts:
- CausalUpdate: Timestamped record of a value change with causal dependencies
- Temporal consistency: Prevention of retroactive causation and causal loops
- Counterfactual branches: Alternative causal histories for what-if analysis
- Causal resolution: Value determination considering temporal recency and confidence
"""
from __future__ import annotations
from typing import Optional, Any, List, Dict, Callable
from dataclasses import dataclass, field
import hashlib
import numpy as np
from datetime import datetime
from .standard import Variable
from .base import Domain
from .types import TemporalStatus, Distribution
[docs]
@dataclass
class CausalUpdate:
"""
Record of a causal update with full temporal and dependency metadata.
A CausalUpdate represents a single causal event where a variable's value
changes due to a specific mechanism. It captures not just the new value,
but the complete causal context: when it happened, what caused it, and
with what confidence.
Parameters
----------
source : str
Identifier of the mechanism or process that caused this update
(e.g., "metabolism", "insulin_injection", "sensor_reading").
value : Any
The new value assigned by this update.
timestamp : float
Simulation time when this update occurred. Must be monotonically
non-decreasing for each source to maintain causal consistency.
id : str, optional
Unique identifier for this update. Auto-generated if not provided.
deps : List[str], optional
List of causal IDs that this update depends on. Forms the causal
dependency graph when combined across updates.
status : TemporalStatus, default=TemporalStatus.CONSEQUENT
Temporal status indicating the update's position in causal ordering.
confidence : float, default=1.0
Confidence level (0.0 to 1.0) in this update's value. Used in
weighted resolution strategies.
meta : Dict[str, Any], optional
Additional metadata for domain-specific information or analysis.
Attributes
----------
source : str
Source mechanism identifier.
value : Any
The update value.
timestamp : float
Update timestamp.
id : str
Unique causal identifier.
deps : List[str]
Causal dependencies.
status : TemporalStatus
Temporal status.
confidence : float
Update confidence.
meta : Dict[str, Any]
Additional metadata.
Examples
--------
>>> from causaloop import CausalUpdate
>>> update = CausalUpdate(
... source="metabolism",
... value=95.0,
... timestamp=10.5,
... deps=["prev_update_id"],
... confidence=0.95
... )
>>> update.id # Auto-generated
'a1b2c3d4e5f6g7h8'
>>> update.timestamp
10.5
Notes
-----
CausalUpdates form the building blocks of causal reasoning. By tracking
dependencies between updates, the system can reconstruct causal pathways
and detect inconsistencies.
The auto-generated ID uses SHA-256 hashing for uniqueness while keeping
the identifier reasonably short. For extremely high-frequency systems,
consider providing custom IDs to avoid collisions.
"""
source: str
value: Any
timestamp: float
id: str = ""
deps: List[str] = field(default_factory=list)
status: TemporalStatus = TemporalStatus.CONSEQUENT
confidence: float = 1.0
meta: Dict[str, Any] = field(default_factory=dict)
def __post_init__(self) -> None:
"""
Auto-generate unique ID if not provided.
The ID is generated from a hash of the source, timestamp, value,
and current time to ensure uniqueness even for identical updates
from the same source at the same time.
"""
if not self.id:
content = (
f"{self.source}:{self.timestamp}:{self.value}:"
f"{datetime.now().timestamp()}"
)
self.id = hashlib.sha256(content.encode()).hexdigest()[:16]
[docs]
class CausalError(Exception):
"""
Exception raised for causal consistency violations.
This exception indicates that a causal operation would violate temporal
consistency, create causal loops, or otherwise break the causal integrity
of the system.
Examples
--------
>>> from causaloop import Interval, CausalVariable, CausalError
>>> var = CausalVariable("temp", Interval(0, 100))
>>> var.update("heater", 25.0, time=10.0)
'update_id_1'
>>> try:
... var.update("heater", 30.0, time=9.0) # Earlier time!
... except CausalError as e:
... print(e)
Temporal violation: heater at 9.0 (last update at 10.0)
"""
pass
[docs]
class CausalVariable(Variable):
"""
Variable with causal-temporal tracking and counterfactual reasoning.
CausalVariable extends Variable with the ability to track the complete
causal history of value changes, enforce temporal consistency, and support
counterfactual reasoning. It maintains a causal graph of updates and
provides methods for analyzing causal pathways.
Key features:
1. Temporal consistency: Prevents retroactive causation and causal loops
2. Causal history: Complete record of all updates with metadata
3. Counterfactual reasoning: Alternative causal scenarios for what-if analysis
4. Causal resolution: Value determination considering temporal context
5. Dependency tracking: Explicit causal dependencies between updates
Parameters
----------
name : str
Variable identifier.
domain : Domain
Value domain with validation rules.
value : Any, optional
Initial deterministic value.
units : str, optional
Measurement units.
noise : Distribution, optional
Noise model for stochastic perturbations.
rng : np.random.Generator, optional
Random number generator for noise (required if noise specified).
max_history : int, default=1000
Maximum number of causal updates to retain in history.
Older updates are discarded when this limit is exceeded.
Attributes
----------
max_history : int
Maximum causal history length.
_history : List[CausalUpdate]
Complete causal history in chronological order.
_constraints : List[Callable]
Temporal consistency constraints.
_pending : Dict[str, CausalUpdate]
Updates pending causal resolution.
_last_time : float
Timestamp of most recent update.
_branches : Dict[str, List[CausalUpdate]]
Counterfactual causal branches.
Examples
--------
>>> import numpy as np
>>> from causaloop import CausalVariable, Interval
>>> # Create causal variable
>>> var = CausalVariable("glucose", Interval(50, 300), 95.0, "mg/dL")
>>> # Record causal updates
>>> update_id = var.update(
... source="metabolism",
... value=95.0,
... time=10.0,
... deps=["previous_event"],
... confidence=0.95
... )
>>> # Query causal history
>>> history = var.history(start=0, end=20)
>>> len(history)
1
>>> # Create counterfactual scenario
>>> var.branch(
... branch_id="insulin_intervention",
... time=10.0,
... value=85.0,
... source="intervention"
... )
>>> # Query counterfactual value
>>> cf_value = var.get_branch_value("insulin_intervention", time=11.0)
See Also
--------
Variable : Parent class with deterministic resolution and noise handling.
CausalUpdate : Individual causal event record.
"""
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,
max_history: int = 1000,
):
"""
Initialize a CausalVariable with causal tracking capabilities.
Parameters
----------
name : str
Variable identifier.
domain : Domain
Domain of valid values.
value : Any, optional
Initial deterministic value.
units : str, optional
Measurement units.
noise : Distribution, optional
Noise model for stochastic perturbations.
rng : np.random.Generator, optional
Random number generator for noise (required if noise specified).
max_history : int, default=1000
Maximum causal updates to retain. Set based on memory constraints
and analysis requirements.
Notes
-----
The causal history is stored as a list of CausalUpdate objects.
When max_history is exceeded, the oldest updates are removed to
maintain constant memory usage. Consider setting max_history based
on your temporal analysis needs and memory constraints.
For long-running simulations with frequent updates, you may need
to increase max_history or implement external storage for complete
historical analysis.
"""
super().__init__(name, domain, value, units, noise, rng)
self.max_history = max_history
self._history: List[CausalUpdate] = []
self._constraints: List[Callable] = []
self._pending: Dict[str, CausalUpdate] = {}
self._last_time: float = 0.0
self._branches: Dict[str, List[CausalUpdate]] = {}
[docs]
def update(
self,
source: str,
value: Any,
time: float,
deps: Optional[List[str]] = None,
confidence: float = 1.0,
meta: Optional[Dict[str, Any]] = None
) -> str:
"""
Record a causal update with temporal consistency checking.
Parameters
----------
source : str
Source mechanism causing this update.
value : Any
New deterministic value (must satisfy domain constraints).
time : float
Simulation time when this update occurs.
deps : List[str], optional
List of causal IDs that this update depends on.
confidence : float, default=1.0
Confidence in this update (0.0 to 1.0).
meta : Dict[str, Any], optional
Additional metadata for analysis or domain-specific information.
Returns
-------
str
Unique causal ID of the created update.
Raises
------
CausalError
If the update violates temporal consistency constraints.
ValueError
If the value violates domain constraints.
Examples
--------
>>> from causaloop import CausalVariable, Interval
>>> var = CausalVariable("temperature", Interval(0, 100))
>>> # Simple update
>>> update_id = var.update("heater", 25.0, time=10.0)
>>> # Update with dependencies
>>> update_id2 = var.update(
... source="thermostat",
... value=26.0,
... time=11.0,
... deps=[update_id], # Depends on previous update
... confidence=0.9,
... meta={"room": "living_room"}
... )
>>> # Invalid: retroactive update
>>> try:
... var.update("heater", 30.0, time=9.0)
... except CausalError as e:
... print(e)
Temporal violation: heater at 9.0 (last update at 10.0)
Notes
-----
This method enforces causal consistency by:
1. Checking that time >= last update time for this source
2. Validating against all registered temporal constraints
3. Recording dependencies to build the causal graph
The update is stored in causal history and also registered as a
normal source output for deterministic value resolution.
Dependencies should reference valid causal IDs from previous updates.
Circular dependencies are not automatically detected but will cause
issues in causal analysis.
"""
# Check temporal consistency
if not self._check_time(source, time):
raise CausalError(
f"Temporal violation: {source} at {time} "
f"(last update at {self._last_time})"
)
# Create causal update record
update = CausalUpdate(
source=source,
value=value,
timestamp=time,
deps=deps or [],
confidence=confidence,
meta=meta or {}
)
# Store in history with size limit
self._history.append(update)
if len(self._history) > self.max_history:
self._history.pop(0)
# Add to pending updates for current resolution
self._pending[source] = update
self._last_time = time
# Register as normal source output (validates domain)
super().register_output(source, value)
return update.id
def _check_time(self, source: str, time: float) -> bool:
"""
Validate temporal ordering for a potential update.
Parameters
----------
source : str
Source mechanism identifier.
time : float
Proposed update time.
Returns
-------
bool
True if the update satisfies all temporal constraints,
False otherwise.
Notes
-----
Basic checks:
1. Time must not be earlier than last update time
2. All registered temporal constraints must be satisfied
Temporal constraints are user-defined callables with signature:
constraint(variable: CausalVariable, source: str, time: float) -> bool
This method is called before recording any update to prevent
causal consistency violations.
"""
# Basic monotonicity check
if time < self._last_time:
return False
# Check all registered constraints
for constraint in self._constraints:
if not constraint(self, source, time):
return False
return True
[docs]
def add_constraint(
self, constraint: Callable[[CausalVariable, str, float], bool]
) -> None:
"""
Add a temporal consistency constraint.
Parameters
----------
constraint : Callable
Constraint function with signature:
constraint(variable: CausalVariable, source: str, time: float) -> bool
Should return True if the update is allowed, False otherwise.
Examples
--------
>>> from causaloop import CausalVariable, Interval
>>> var = CausalVariable("pressure", Interval(0, 100))
>>> # Constraint: updates from "valve" can only occur at integer times
>>> def integer_time_constraint(var, source, time):
... if source == "valve":
... return time.is_integer()
... return True
>>> var.add_constraint(integer_time_constraint)
>>> # This will work
>>> var.update("valve", 50.0, time=10.0)
'update_id'
>>> # This will fail the constraint
>>> try:
... var.update("valve", 60.0, time=10.5)
... except CausalError:
... print("Constraint violation")
Constraint violation
Notes
-----
Constraints allow domain-specific temporal logic to be enforced.
Common uses include:
- Minimum time between updates from certain sources
- Synchronization requirements between variables
- Domain-specific timing patterns
- Resource availability constraints
Constraints are checked in registration order. The first constraint
that returns False prevents the update.
For performance, keep constraint functions simple and fast.
"""
self._constraints.append(constraint)
@property
def value(self) -> Any:
"""
Get current value using causal-aware resolution.
Returns
-------
Any
Current deterministic value resolved using causal strategy.
Notes
-----
The causal resolution strategy considers:
1. Temporal recency: More recent updates have higher weight
2. Confidence: Updates with higher confidence have higher weight
3. Source-specific weighting: As configured via add_source()
The resolution uses exponential decay for temporal weighting:
weight = confidence * exp(-decay_rate * time_since_update)
This gives a smooth transition between updates while respecting
causal ordering.
For the pure deterministic value without causal weighting,
use the parent class's resolution directly:
>>> det_value = super(CausalVariable, var).value
Or access the most recent update:
>>> recent = var.get_most_recent()
>>> det_value = recent.value if recent else None
"""
return self._resolve_causal()
def _resolve_causal(self, strategy: str = "causal") -> Any:
"""
Internal method for causal-aware value resolution.
Parameters
----------
strategy : str, default="causal"
Resolution strategy:
- "causal": Weighted by temporal recency and confidence
- "most_recent": Value from most recent update
- "most_confident": Value from highest confidence update
- Other strategies inherited from Variable
Returns
-------
Any
Resolved value according to specified strategy.
Raises
------
ValueError
If unknown strategy specified.
Notes
-----
The "causal" strategy implements exponential decay weighting:
decay = exp(-decay_rate * (current_time - update_time))
weight = update.confidence * decay
Where decay_rate controls how quickly older updates lose influence.
The current implementation uses decay_rate = 0.1.
If no pending updates exist, falls back to parent class resolution.
"""
if not self._pending:
return super()._resolve_deterministic()
if strategy == "causal":
# Use most recent update time as reference
now = max((u.timestamp for u in self._pending.values()), default=0.0)
total_weight = 0.0
weighted_sum = 0.0
for update in self._pending.values():
# Exponential temporal decay
time_diff = max(0.0, now - update.timestamp)
decay = np.exp(-0.1 * time_diff) # Decay rate = 0.1
# Combine temporal decay with confidence
weight = update.confidence * decay
try:
weighted_sum += float(update.value) * weight
total_weight += weight
except (ValueError, TypeError):
# Non-numeric value: return most recent
return update.value
if total_weight > 0:
return weighted_sum / total_weight
return None
elif strategy == "most_recent":
# Value from most recent update
most_recent = max(
self._pending.values(),
key=lambda u: u.timestamp
)
return most_recent.value
elif strategy == "most_confident":
# Value from highest confidence update
most_confident = max(
self._pending.values(),
key=lambda u: u.confidence
)
return most_confident.value
else:
# Fall back to parent strategies
return super()._resolve_deterministic(strategy)
[docs]
def resolve_causal(self, strategy: str = "causal") -> Any:
"""
Get value using specified causal resolution strategy.
Parameters
----------
strategy : str, default="causal"
Resolution strategy (see _resolve_causal for options).
Returns
-------
Any
Resolved value according to specified strategy.
Examples
--------
>>> from causaloop import CausalVariable, Interval
>>> var = CausalVariable("signal", Interval(0, 10))
>>> var.update("source1", 5.0, time=1.0, confidence=0.8)
'id1'
>>> var.update("source2", 7.0, time=2.0, confidence=0.9)
'id2'
>>> # Different resolution strategies
>>> var.resolve_causal("causal") # Weighted by time and confidence
np.float64(6.10...)
>>> var.resolve_causal("most_recent") # Most recent update
7.0
>>> var.resolve_causal("most_confident") # Highest confidence
7.0
Notes
-----
This method provides explicit control over causal resolution,
which can be important for different analysis purposes:
- "causal": For simulation where recent, confident updates dominate
- "most_recent": For real-time monitoring or state tracking
- "most_confident": When reliability is more important than recency
"""
return self._resolve_causal(strategy)
[docs]
def history(
self,
start: float = 0.0,
end: Optional[float] = None,
source: Optional[str] = None
) -> List[CausalUpdate]:
"""
Retrieve causal history with optional filtering.
Parameters
----------
start : float, default=0.0
Start time (inclusive) for history query.
end : float, optional
End time (inclusive) for history query.
If None, includes all updates from start onward.
source : str, optional
If provided, only returns updates from this source.
Returns
-------
List[CausalUpdate]
List of causal updates matching the filter criteria,
in chronological order.
Examples
--------
>>> from causaloop import CausalVariable, Interval
>>> var = CausalVariable("pressure", Interval(0, 100))
>>> var.update("valve", 50.0, time=10.0)
'id1'
>>> var.update("pump", 60.0, time=12.0)
'id2'
>>> var.update("valve", 55.0, time=15.0)
'id3'
>>> # All history
>>> all_history = var.history()
>>> len(all_history)
3
>>> # Time window
>>> window = var.history(start=11.0, end=14.0)
>>> len(window)
1
>>> window[0].source
'pump'
>>> # Source-specific
>>> valve_history = var.history(source="valve")
>>> len(valve_history)
2
Notes
-----
The history is stored in chronological order, so this method
performs linear filtering. For large histories with frequent
queries, consider maintaining additional indexing structures.
If max_history is set and old updates have been discarded,
queries for times before the oldest retained update will return
empty results.
"""
if end is None:
end = float('inf')
filtered = [
u for u in self._history
if start <= u.timestamp <= end
]
if source is not None:
filtered = [u for u in filtered if u.source == source]
return filtered
[docs]
def get_most_recent(self, source: Optional[str] = None) -> Optional[CausalUpdate]:
"""
Get the most recent causal update.
Parameters
----------
source : str, optional
If provided, returns most recent update from this specific source.
Returns
-------
Optional[CausalUpdate]
Most recent update matching criteria, or None if no updates exist.
Examples
--------
>>> from causaloop import CausalVariable, Interval
>>> var = CausalVariable("temperature", Interval(0, 100))
>>> var.update("sensor1", 25.0, time=10.0)
'id1'
>>> var.update("sensor2", 26.0, time=11.0)
'id2'
>>> # Most recent overall
>>> recent = var.get_most_recent()
>>> recent.source
'sensor2'
>>> recent.timestamp
11.0
>>> # Most recent from specific source
>>> sensor1_recent = var.get_most_recent(source="sensor1")
>>> sensor1_recent.value
25.0
Notes
-----
This method is optimized for frequent access by tracking the
most recent update time internally. For source-specific queries,
it performs a reverse linear search through history.
For variables with very long histories, consider maintaining
source-specific pointers for O(1) access.
"""
if not self._history:
return None
if source is None:
# Most recent overall (history is chronological)
return self._history[-1]
# Most recent from specific source
for update in reversed(self._history):
if update.source == source:
return update
return None
[docs]
def branch(
self,
branch_id: str,
time: float,
value: Any,
source: str = "counterfactual"
) -> None:
"""
Create a counterfactual causal branch.
Parameters
----------
branch_id : str
Unique identifier for this counterfactual branch.
time : float
Intervention time where the counterfactual diverges from actual history.
value : Any
Counterfactual intervention value at the divergence point.
source : str, default="counterfactual"
Source identifier for the counterfactual intervention.
Examples
--------
>>> from causaloop import CausalVariable, Interval
>>> var = CausalVariable("glucose", Interval(50, 300), 95.0, "mg/dL")
>>> var.update("metabolism", 95.0, time=9.0)
'id1'
>>> var.update("meal", 150.0, time=10.0)
'id2'
>>> # Create counterfactual: what if insulin was given instead of meal?
>>> var.branch(
... branch_id="insulin_intervention",
... time=10.0, # Same time as the meal
... value=85.0, # Insulin would lower glucose
... source="insulin_injection"
... )
>>> # Query counterfactual value at later time
>>> cf_value = var.get_branch_value("insulin_intervention", time=11.0)
Notes
-----
Counterfactual branches allow what-if analysis by creating alternative
causal histories. Each branch:
1. Copies all actual history up to the intervention time
2. Adds the counterfactual intervention as a new update
3. Can be extended with additional hypothetical updates
Branches are stored separately from actual history and don't affect
the variable's current value or normal operations.
Multiple branches can be created for different intervention scenarios.
Branch IDs must be unique within a variable.
"""
# Get actual history before intervention
pre_intervention = [u for u in self._history if u.timestamp < time]
# Create counterfactual intervention
last_dep = pre_intervention[-1].id if pre_intervention else None
intervention = CausalUpdate(
source=source,
value=value,
timestamp=time,
deps=[last_dep] if last_dep else [],
status=TemporalStatus.COUNTERFACTUAL,
meta={"branch_id": branch_id, "intervention": True}
)
# Store branch
self._branches[branch_id] = pre_intervention + [intervention]
[docs]
def get_branch_value(
self,
branch_id: str,
time: float,
strategy: str = "causal"
) -> Optional[Any]:
"""
Get value in a counterfactual branch at specified time.
Parameters
----------
branch_id : str
Counterfactual branch identifier.
time : float
Time to query in the counterfactual scenario.
strategy : str, default="causal"
Resolution strategy for determining the value.
See _resolve_causal for available strategies.
Returns
-------
Optional[Any]
Value in the counterfactual scenario at the specified time,
or None if the branch doesn't exist or has no relevant updates.
Examples
--------
>>> from causaloop import CausalVariable, Interval
>>> var = CausalVariable("temperature", Interval(0, 100), 20.0)
>>> var.update("heater", 25.0, time=10.0)
'id1'
>>> # Create counterfactual: what if heater was stronger?
>>> var.branch("stronger_heater", time=10.0, value=30.0)
>>> # Query counterfactual value
>>> cf_value = var.get_branch_value("stronger_heater", time=11.0)
>>> cf_value
30.0
Notes
-----
This method reconstructs the counterfactual state by:
1. Finding all updates in the branch up to the query time
2. Applying the specified resolution strategy to those updates
The resolution considers only updates within the branch, ignoring
actual history after the intervention point.
For complex counterfactuals with multiple hypothetical updates
after the intervention, additional updates would need to be added
to the branch manually.
"""
if branch_id not in self._branches:
return None
branch_updates = self._branches[branch_id]
# Get updates up to query time
relevant = [u for u in branch_updates if u.timestamp <= time]
if not relevant:
return None
# Simple resolution: most recent in branch
if strategy == "most_recent":
return relevant[-1].value
# For more complex strategies, we'd need to reconstruct
# a temporary variable state. For now, return most recent.
return relevant[-1].value
[docs]
def get_causal_path(
self,
start_time: float,
end_time: float
) -> List[CausalUpdate]:
"""
Get the causal pathway between two times.
Parameters
----------
start_time : float
Start time for causal pathway.
end_time : float
End time for causal pathway.
Returns
-------
List[CausalUpdate]
Updates in the causal pathway, in chronological order.
Examples
--------
>>> from causaloop import CausalVariable, Interval
>>> var = CausalVariable("signal", Interval(0, 10))
>>> var.update("A", 1.0, time=1.0, deps=[])
'id1'
>>> var.update("B", 2.0, time=2.0, deps=["id1"])
'id2'
>>> var.update("C", 3.0, time=3.0, deps=["id2"])
'id3'
>>> # Get causal pathway
>>> path = var.get_causal_path(start_time=1.0, end_time=3.0)
>>> [u.source for u in path]
['A', 'B', 'C']
Notes
-----
A causal pathway is a sequence of updates where each update
(after the first) depends on the previous one. This method
reconstructs such pathways by following dependency links.
If multiple dependency paths exist, this method returns one
valid path (not necessarily the shortest or only one).
Circular dependencies will cause infinite recursion.
"""
# Get updates in time window
updates = self.history(start=start_time, end=end_time)
if not updates:
return []
# Build dependency graph
by_id = {u.id: u for u in updates}
pathways = []
# Start from each update and trace back dependencies
for update in updates:
path = []
current = update
visited = set()
while current and current.id not in visited:
path.append(current)
visited.add(current.id)
# Follow first dependency that's in our set
next_id = next((dep for dep in current.deps if dep in by_id), None)
current = by_id.get(next_id) if next_id else None
# Keep the longest valid path (no cycles)
if len(path) > len(pathways):
pathways = path
# Return in chronological order
pathways.sort(key=lambda u: u.timestamp)
return pathways
[docs]
def clear_pending(self) -> None:
"""
Clear pending updates without affecting history.
Examples
--------
>>> from causaloop import CausalVariable, Interval
>>> var = CausalVariable("output", Interval(0, 100))
>>> var.update("process", 50.0, time=10.0)
'id1'
>>> len(var._pending)
1
>>> var.clear_pending()
>>> len(var._pending)
0
>>> len(var._history) # History preserved
1
Notes
-----
Pending updates are used for current value resolution but can be
cleared to reset the variable's "current state" while preserving
historical records. This is useful between simulation runs or
analysis phases.
Clearing pending updates does not affect the causal history or
counterfactual branches.
"""
self._pending.clear()
[docs]
def reset(self) -> None:
"""
Reset all causal tracking (history, branches, pending updates).
Examples
--------
>>> from causaloop import CausalVariable, Interval
>>> var = CausalVariable("signal", Interval(0, 10))
>>> var.update("source", 5.0, time=1.0)
'id1'
>>> var.branch("counterfactual", time=1.0, value=6.0)
>>> len(var._history)
1
>>> len(var._branches)
1
>>> var.reset()
>>> len(var._history)
0
>>> len(var._branches)
0
>>> var._last_time
0.0
Notes
-----
This method completely resets causal tracking while preserving
the variable's basic configuration (domain, units, noise model).
Use this method to start fresh causal tracking, such as when
beginning a new simulation or analysis from time zero.
Source registrations and weights from the parent class are
preserved. Only causal-specific state is reset.
"""
self._history.clear()
self._constraints.clear()
self._pending.clear()
self._branches.clear()
self._last_time = 0.0
def __repr__(self) -> str:
"""
Return string representation.
Returns
-------
str
String showing class name, variable name, current value,
history size, pending updates, and noise indicator.
Examples
--------
>>> from causaloop import CausalVariable, Interval
>>> var = CausalVariable("temperature", Interval(0, 100), 25.0)
>>> repr(var)
"CausalVariable('temperature', value=25.0, hist=0, pend=0)"
>>> var.update("heater", 30.0, time=10.0)
'id1'
>>> repr(var)
"CausalVariable('temperature', value=30.0, hist=1, pend=1)"
"""
hist_len = len(self._history)
pend_len = len(self._pending)
noise_indicator = " +noise" if self.has_noise else ""
return (
f"{self.__class__.__name__}("
f"{self.name!r}, value={self.value}, "
f"hist={hist_len}, pend={pend_len}{noise_indicator})"
)
def __str__(self) -> str:
"""
Return human-readable string representation.
Returns
-------
str
Formatted string with name, value, units, history information,
and noise indicator.
"""
base_str = super().__str__()
hist_len = len(self._history)
branches = len(self._branches)
causal_info = f" [causal: {hist_len} updates, {branches} branches]"
return base_str + causal_info
[docs]
def to_dict(self) -> Dict:
"""
Convert variable to detailed dictionary representation.
Returns
-------
Dict
Dictionary containing variable metadata, source information,
causal history summary, and configuration.
Notes
-----
The full causal history can be large, so this method only includes
summary statistics. For complete serialization of causal state,
additional methods would be needed to export/import the full
history and branches.
Noise models and RNGs require special handling for serialization
and are not included in the basic dictionary representation.
"""
base_dict = super().to_dict()
base_dict.update({
"max_history": self.max_history,
"history_count": len(self._history),
"pending_count": len(self._pending),
"branch_count": len(self._branches),
"last_time": self._last_time,
"constraint_count": len(self._constraints),
"is_causal": True
})
return base_dict