API reference

Auto-generated from the radCAD source docstrings.

radcad.wrappers

The user-facing modelling classes: Model, Simulation, and Experiment.

radcad.wrappers

User-facing classes for describing and running radCAD simulations.

This module provides the three classes that make up the radCAD modelling hierarchy: Model > Simulation > Experiment. A Model captures what the system is, a Simulation captures how long and how many times to run it, and an Experiment groups one or more simulations so they can be executed together by the Engine.

Model

Model(initial_state={}, state_update_blocks=[], params={})

A dynamical-system model: an initial state plus the rules that evolve it.

A Model bundles together the three things needed to describe a system: the initial_state of its State Variables, the state_update_blocks (Partial State Update Blocks) that transition state from one timestep to the next, and the params (System Parameters) those rules depend on.

A Model is also iterable: iterating over it advances the system one timestep at a time, which is useful for live digital twins, gradient descent, or composing one model inside another. See __iter__ and __call__.

Parameters:

Name	Type	Description	Default
initial_state	dict	Mapping of State Variable name to initial value.	{}
state_update_blocks	list	Ordered list of Partial State Update Blocks, each a dict with "policies" and "variables" keys.	[]
params	dict | dataclass	System Parameters, either as a dict of lists (for sweeps) or a (possibly nested) dataclass.	{}

Source code in radcad/wrappers.py

def __init__(self, initial_state={}, state_update_blocks=[], params={}):
    self.substeps = []
    self.state = {
        **copy.deepcopy(initial_state),
        'simulation': 0,
        'subset': 0,
        'run': 1,
        'substep': 0,
        'timestep': 0
    }
    self.initial_state = copy.deepcopy(initial_state)
    self.state_update_blocks = state_update_blocks
    self.params = copy.deepcopy(params)
    self.exceptions = []
    self._raise_exceptions = True
    self._deepcopy = True
    self._deepcopy_method = SimulationExecution.deepcopy_method
    self._drop_substeps = False

__iter__

__iter__()

Advance the model one timestep per iteration, yielding self.

Each iteration runs a single-timestep simulation from the current state, updates state and substeps, and yields the model so the caller can read the new state. Only single-subset models are supported (no parameter sweeps when iterating).

Source code in radcad/wrappers.py

def __iter__(self):
    """Advance the model one timestep per iteration, yielding ``self``.

    Each iteration runs a single-timestep simulation from the current
    `state`, updates `state` and `substeps`, and yields
    the model so the caller can read the new state. Only single-subset
    models are supported (no parameter sweeps when iterating).
    """
    param_sweep = generate_parameter_sweep(self.params)
    while True:
        _params = param_sweep[0] if param_sweep else {}
        simulation_execution = SimulationExecution(
            # Model / simulation settings
            timesteps = 1,
            initial_state = copy.deepcopy(self.state),
            state_update_blocks = self.state_update_blocks,
            params = _params,
            # Execution settings
            enable_deepcopy = self._deepcopy,
            drop_substeps = self._drop_substeps,
            raise_exceptions = self._raise_exceptions,
        )
        simulation_execution.deepcopy_method = self._deepcopy_method
        result, exception = multiprocess_wrapper(simulation_execution)
        if exception: self.exceptions.append(exception)
        self.substeps = result.pop()
        self.state = self.substeps[-1] if self.substeps else self.state
        yield self

__call__

__call__(**kwargs)

Configure execution options for iteration, then return self.

Lets you set engine-style options before iterating, e.g. next(model(deepcopy=False, drop_substeps=True)).

Parameters:

Name	Type	Description	Default
**raise_exceptions	bool	Raise on failure rather than collecting exceptions. Defaults to True.	required
**deepcopy	bool	Deepcopy state to prevent unintended mutation. Defaults to True.	required
**deepcopy_method	Callable	Custom deepcopy implementation.	required
**drop_substeps	bool	Keep only the final substep per timestep. Defaults to False.	required

Returns:

Type	Description
Model	self, ready to be iterated.

Source code in radcad/wrappers.py

def __call__(self, **kwargs):
    """Configure execution options for iteration, then return ``self``.

    Lets you set engine-style options before iterating, e.g.
    ``next(model(deepcopy=False, drop_substeps=True))``.

    Args:
        **raise_exceptions (bool): Raise on failure rather than collecting
            exceptions. Defaults to ``True``.
        **deepcopy (bool): Deepcopy state to prevent unintended mutation.
            Defaults to ``True``.
        **deepcopy_method (Callable): Custom deepcopy implementation.
        **drop_substeps (bool): Keep only the final substep per timestep.
            Defaults to ``False``.

    Returns:
        Model: ``self``, ready to be iterated.
    """
    self._raise_exceptions = kwargs.pop("raise_exceptions", True)
    self._deepcopy = kwargs.pop("deepcopy", True)
    self._deepcopy_method = kwargs.pop("deepcopy_method", SimulationExecution.deepcopy_method)
    self._drop_substeps = kwargs.pop("drop_substeps", False)
    return self

Simulation

Simulation(model: Model, timesteps=100, runs=1, **kwargs)

Bases: Executable

A Model together with a number of timesteps and runs.

A Simulation answers "run this model for how long, and how many times?". The number of runs drives Monte Carlo analysis, while the model's System Parameters drive parameter sweeps. Call run (or wrap it in an Experiment) to execute it via the Engine.

Parameters:

Name	Type	Description	Default
model	Model	The model to simulate.	required
timesteps	int	Number of timesteps per run. Defaults to 100.	100
runs	int	Number of (Monte Carlo) runs. Defaults to 1.	1

Additional Executable keyword arguments (engine and the lifecycle hooks) are also accepted.

Source code in radcad/wrappers.py

def __init__(self, model: Model, timesteps=100, runs=1, **kwargs):
    super().__init__(**kwargs)

    self.model = model
    self.timesteps = timesteps
    self.runs = runs

    self.index = kwargs.pop("index", 0)

    if kwargs:
        raise Exception(f"Invalid Simulation option in {kwargs}")

run

run()

Execute the simulation and return its results.

Returns:

Type	Description
SimulationResults	A flat list of state dictionaries, one per substep, suitable for pandas.DataFrame(...).

Source code in radcad/wrappers.py

def run(self):
    """Execute the simulation and return its results.

    Returns:
        SimulationResults: A flat list of state dictionaries, one per
        substep, suitable for ``pandas.DataFrame(...)``.
    """
    return self.engine._run(executable=self)

Experiment

Experiment(simulations=[], **kwargs)

Bases: Executable

A collection of Simulation objects run together.

Grouping simulations into an Experiment lets the Engine execute them in parallel (e.g. for A/B tests) and collects their results and exceptions in one place.

Parameters:

Name	Type	Description	Default
simulations	Simulation | list	A simulation or list of simulations.	[]

Additional Executable keyword arguments (engine and the lifecycle hooks) are also accepted.

Source code in radcad/wrappers.py

def __init__(self, simulations=[], **kwargs):
    super().__init__(**kwargs)

    # Add and validate simulations
    self.simulations = []
    self.add_simulations(simulations)

    if kwargs:
        raise Exception(f"Invalid Experiment option in {kwargs}")

run

run()

Execute every simulation in the experiment and return the results.

Returns:

Type	Description
SimulationResults	A flat list of state dictionaries across all simulations, suitable for pandas.DataFrame(...).

Source code in radcad/wrappers.py

def run(self):
    """Execute every simulation in the experiment and return the results.

    Returns:
        SimulationResults: A flat list of state dictionaries across all
        simulations, suitable for ``pandas.DataFrame(...)``.
    """
    return self.engine._run(executable=self)

add_simulations

add_simulations(simulations)

Append one or more Simulation objects to the experiment.

Parameters:

Name	Type	Description	Default
simulations	Simulation | list	Simulation(s) to add.	required

Raises:

Type	Description
Exception	If any item is not a Simulation.

Source code in radcad/wrappers.py

def add_simulations(self, simulations):
    """Append one or more [Simulation][radcad.wrappers.Simulation] objects to the experiment.

    Args:
        simulations (Simulation | list): Simulation(s) to add.

    Raises:
        Exception: If any item is not a [Simulation][radcad.wrappers.Simulation].
    """
    if not isinstance(simulations, list):
        simulations = [simulations]
    if any(not isinstance(sim, Simulation) for sim in simulations):
        raise Exception("Invalid simulation added")
    self.simulations.extend(simulations)

clear_simulations

clear_simulations()

Remove all simulations.

Returns:

Type	Description
bool	True if there were simulations to clear, else False.

Source code in radcad/wrappers.py

def clear_simulations(self):
    """Remove all simulations.

    Returns:
        bool: ``True`` if there were simulations to clear, else ``False``.
    """
    cleared = True if self.simulations else False
    self.simulations = []
    return cleared

get_simulations

get_simulations()

Return the list of simulations in the experiment.

Source code in radcad/wrappers.py

def get_simulations(self):
    """Return the list of simulations in the experiment."""
    return self.simulations

Executable

Executable(**kwargs)

Base class for things the Engine can run.

Holds the shared machinery for both Simulation and Experiment: the Engine instance, the results and exceptions populated after a run, and the lifecycle hooks used to extend behaviour without modifying the core. Not intended to be instantiated directly.

Accepts the following keyword arguments: engine (an Engine, defaulting to a new instance) and the optional lifecycle hooks before_experiment, after_experiment, before_simulation, after_simulation, before_run, after_run, before_subset, and after_subset.

Source code in radcad/wrappers.py

def __init__(self, **kwargs) -> None:
    self.engine = kwargs.pop("engine", Engine())

    self.results: SimulationResults = []
    self.exceptions = []

    # Hooks
    self.before_experiment = kwargs.pop("before_experiment", None)
    self.after_experiment = kwargs.pop("after_experiment", None)
    self.before_simulation = kwargs.pop("before_simulation", None)
    self.after_simulation = kwargs.pop("after_simulation", None)
    self.before_run = kwargs.pop("before_run", None)
    self.after_run = kwargs.pop("after_run", None)
    self.before_subset = kwargs.pop("before_subset", None)
    self.after_subset = kwargs.pop("after_subset", None)

__deepcopy__

__deepcopy__(memo={})

Deepcopy the executable, resetting the engine's run generator first.

The engine's _run_generator is not picklable/copyable once a run has started, so it is reset to an empty iterator before copying.

Source code in radcad/wrappers.py

def __deepcopy__(self, memo={}):
    """Deepcopy the executable, resetting the engine's run generator first.

    The engine's ``_run_generator`` is not picklable/copyable once a run
    has started, so it is reset to an empty iterator before copying.
    """
    # Reset iterators to enable deepcopy after simulation run
    self.engine._run_generator = iter(())

    cls = self.__class__
    result = cls.__new__(cls)
    memo[id(self)] = result
    for k, v in self.__dict__.items():
        setattr(result, k, copy.deepcopy(v, memo))
    return result

run

run()

Run the executable and return its results. Implemented by subclasses.

Source code in radcad/wrappers.py

def run(self):
    """Run the executable and return its results. Implemented by subclasses."""
    raise NotImplementedError("Method run() not implemented for class that extends Base")

radcad.engine

The execution engine.

radcad.engine

The execution engine that runs simulations and experiments.

The Engine turns the declarative description in a Simulation or Experiment into actual computation. It expands parameter sweeps and runs, dispatches the work to a chosen parallel-processing Backend, fires lifecycle hooks, and gathers the results and any exceptions.

Engine

Engine(**kwargs)

Configures and executes experiments and simulations.

The Engine owns all execution-time settings (which parallel-processing backend to use, how many processes, whether to deepcopy state, how to handle exceptions) and drives the run loop over every simulation, run, and parameter subset. An Engine can be passed to a Simulation or Experiment to customise execution.

See __init__ for the full list of options.

Handles configuration and execution of experiments and simulations.

Parameters:

Name	Type	Description	Default
**backend	Backend	Which execution backend to use (e.g. Pathos, Multiprocessing, etc.). Defaults to Backend.DEFAULT / Backend.PATHOS. Can be set via the RADCAD_BACKEND environment variable which takes precedence.	required
**processes	int	Number of system CPU processes to spawn. Defaults to multiprocessing.cpu_count() - 1 or 1	required
**raise_exceptions	bool	Whether to raise exceptions, or catch them and return exceptions along with partial results. Default to True.	required
**deepcopy	bool	Whether to enable deepcopy of State Variables to avoid unintended state mutation. Defaults to True.	required
**deepcopy_method	Callable	Method to use for deepcopy of State Variables. By default uses Pickle for improved performance, use copy.deepcopy for an alternative to Pickle.	required
**drop_substeps	bool	Whether to drop simulation result substeps during runtime to save memory and improve performance. Defaults to False.	required
**simulation_execution	SimulationExecution	Set a custom radCAD SimulationExecution class instance. Overrides setting of raise_exceptions, deepcopy, deepcopy_method, and drop_substeps.	required
**_run_generator	tuple_iterator	Generator to generate simulation runs, used to implement custom execution backends. Defaults to iter(()).	required

Source code in radcad/engine.py

def __init__(self, **kwargs):
    """
    Handles configuration and execution of experiments and simulations.

    Args:
        **backend (Backend): Which execution backend to use (e.g. Pathos, Multiprocessing, etc.). Defaults to `Backend.DEFAULT` / `Backend.PATHOS`. Can be set via the `RADCAD_BACKEND` environment variable which takes precedence.
        **processes (int, optional): Number of system CPU processes to spawn. Defaults to `multiprocessing.cpu_count() - 1 or 1`
        **raise_exceptions (bool): Whether to raise exceptions, or catch them and return exceptions along with partial results. Default to `True`.
        **deepcopy (bool): Whether to enable deepcopy of State Variables to avoid unintended state mutation. Defaults to `True`.
        **deepcopy_method (Callable): Method to use for deepcopy of State Variables. By default uses Pickle for improved performance, use `copy.deepcopy` for an alternative to Pickle.
        **drop_substeps (bool): Whether to drop simulation result substeps during runtime to save memory and improve performance. Defaults to `False`.
        **simulation_execution (SimulationExecution): Set a custom radCAD SimulationExecution class instance. Overrides setting of `raise_exceptions`, `deepcopy`, `deepcopy_method`, and `drop_substeps`.
        **_run_generator (tuple_iterator): Generator to generate simulation runs, used to implement custom execution backends. Defaults to  `iter(())`.
    """
    self.executable = None
    self.processes = kwargs.pop("processes", cpu_count)

    ENV_RADCAD_BACKEND = os.getenv('RADCAD_BACKEND', None)
    if ENV_RADCAD_BACKEND:
        kwargs.pop("backend", None)
        self.backend = Backend[ENV_RADCAD_BACKEND]
    else:
        self.backend = kwargs.pop("backend", Backend.DEFAULT)

    _simulation_execution = kwargs.pop("simulation_execution", None)
    if _simulation_execution:
        self.simulation_execution = _simulation_execution
    else:
        self.raise_exceptions = kwargs.pop("raise_exceptions", True)
        self.deepcopy = kwargs.pop("deepcopy", True)
        self.deepcopy_method = kwargs.pop("deepcopy_method", core.SimulationExecution.deepcopy_method)
        self.drop_substeps = kwargs.pop("drop_substeps", False)

        self.simulation_execution = core.SimulationExecution(
            enable_deepcopy=self.deepcopy,
            drop_substeps=self.drop_substeps,
            raise_exceptions=self.raise_exceptions,
        )
        self.simulation_execution.deepcopy_method = self.deepcopy_method

    self._run_generator = iter(())

    if kwargs:
        raise Exception(f"Invalid Engine option in {kwargs}")

radcad.backends

Parallel-processing backends.

radcad.backends

Parallel-processing backends that execute simulation runs.

A backend determines how the independent units of work produced by the Engine are executed: in a single process, across local CPU cores, or on a remote cluster. Select one via the Engine backend option or the RADCAD_BACKEND environment variable.

Backend

Bases: Enum

Available execution backends.

• DEFAULT: alias for PATHOS.
• PATHOS: local multiprocessing via pathos (dill-based pickling).
• MULTIPROCESSING: local multiprocessing via the standard library.
• RAY: local execution via Ray.
• RAY_REMOTE: distributed execution on a remote Ray cluster.
• SINGLE_PROCESS: no parallelisation; useful for debugging.

Executor

Executor(engine)

Bases: object

Abstract base for backend executors.

Subclasses implement execute_runs to consume the engine's run generator and return a list of (result, exception) tuples.

Source code in radcad/backends/__init__.py

def __init__(self, engine):
    self.engine = engine

execute_runs

execute_runs()

Execute every run from the engine's generator. Implemented by subclasses.

Source code in radcad/backends/__init__.py

def execute_runs(self):
    """Execute every run from the engine's generator. Implemented by subclasses."""
    raise Exception("Backend executor not implemented")

radcad.core

The core simulation loop. Most users never use this directly, but it defines radCAD's execution semantics.

radcad.core

The core simulation loop that evaluates a single run of a model.

A SimulationExecution represents one fully-specified unit of work, a single run of a single parameter subset. Stepping through it produces the raw simulation result. The Engine creates one of these per run/subset and hands it to a backend, which calls multiprocess_wrapper to execute it (with exception handling) in a worker process.

Most models never touch this module directly; it is documented here because its state-transition logic defines radCAD's execution semantics.

SimulationExecution dataclass

SimulationExecution(
    timesteps: int = 0,
    timestep: int = None,
    substep_index: int = 0,
    substeps: List = default([]),
    state_update_blocks: List[StateUpdateBlock] = default(
        []
    ),
    result: SimulationResults = default([]),
    initial_state: StateVariables = default({}),
    params: SystemParameters = default({}),
    enable_deepcopy: bool = True,
    drop_substeps: bool = False,
    raise_exceptions: bool = True,
    simulation_index: int = 0,
    run_index: int = 1,
    subset_index: int = 0,
    previous_state: StateVariables = None,
)

Bases: SimulationExecutionSpecification

One executable run of a model: a single run of a single parameter subset.

Implements radCAD's state-transition semantics on top of SimulationExecutionSpecification. For each substep it executes the block's Policy Functions, aggregates their signals, applies the State Update Functions, and appends the new state to the result. Execution options such as enable_deepcopy, drop_substeps, and raise_exceptions control performance and error behaviour.

Subclass it and override deepcopy_method (or other methods) to customise execution; see the README for an example.

logger class-attribute instance-attribute

logger = logging.getLogger('radCAD')

Use "radCAD" logging instance to avoid conflict with other projects

subset_index class-attribute instance-attribute

subset_index: int = 0

Index starts at +1 to remain compatible with cadCAD implementation

initialise_state

initialise_state()

Stamp bookkeeping keys onto the initial state and seed the result.

Sets the simulation/subset/run/substep/timestep keys and appends the initial state as the first entry of the result.

Source code in radcad/core.py

def initialise_state(self):
    """Stamp bookkeeping keys onto the initial state and seed the result.

    Sets the ``simulation``/``subset``/``run``/``substep``/``timestep``
    keys and appends the initial state as the first entry of the result.
    """
    self.initial_state["simulation"] = self.simulation_index
    self.initial_state["subset"] = self.subset_index
    self.initial_state["run"] = self.run_index
    self.initial_state["substep"] = 0
    # If first timestep is not explicitely set in initial state, start from zero
    # Needed to properly handle model generator
    if not "timestep" in self.initial_state:
        self.initial_state["timestep"] = 0
    assert (
        isinstance(self.initial_state["timestep"], int) and self.initial_state["timestep"] >= 0
    ), f"Simulation initial timestep should be a positive integer, not {self.initial_state['timestep']}"
    self.timestep = self.initial_state["timestep"]
    self.result.append([self.initial_state])

substep

substep(state_update_block: StateUpdateBlock) -> None

Evaluate a single Partial State Update Block.

Runs the block's policies to produce aggregated signals, applies each State Update Function to compute the new State Variable values, and records the resulting substate.

Source code in radcad/core.py

def substep(self, state_update_block: StateUpdateBlock) -> None:
    """Evaluate a single Partial State Update Block.

    Runs the block's policies to produce aggregated signals, applies each
    State Update Function to compute the new State Variable values, and
    records the resulting substate.
    """
    substate: StateVariables = (
        self.previous_state.copy() if self.substep_index == 0 else self.substeps[self.substep_index - 1].copy()
    )

    signals: PolicySignal = self.execute_policies(substate, state_update_block)

    updated_state: StateVariables = dict([
        self.update_state(substate, signals, state_update)
        for state_update in state_update_block["variables"].items()
    ])
    self.process_substep(substate, updated_state)

execute_policies

execute_policies(
    substate: StateVariables,
    state_update_block: StateUpdateBlock,
) -> PolicySignal

Run a block's Policy Functions and merge their signals into one dict.

Each policy receives (params, substep, state_history, substate). Where multiple policies emit the same signal key, their values are summed (see add_signals).

Source code in radcad/core.py

def execute_policies(
    self,
    substate: StateVariables,
    state_update_block: StateUpdateBlock,
) -> PolicySignal:
    """Run a block's Policy Functions and merge their signals into one dict.

    Each policy receives ``(params, substep, state_history, substate)``.
    Where multiple policies emit the same signal key, their values are
    summed (see `add_signals`).
    """
    policy_results: List[PolicySignal] = list(
        map(lambda function: function(
            self.params,
            self.substep,
            self.result,
            self.deepcopy_method(substate) if self.enable_deepcopy else substate.copy()
        ), state_update_block["policies"].values())
    )

    result: dict = {}
    result_length = len(policy_results)
    if result_length == 0:
        return result
    elif result_length == 1:
        return policy_results[0]
    else:
        return reduce(SimulationExecution.add_signals, policy_results, result)

update_state

update_state(
    substate: StateVariables,
    signals: PolicySignal,
    state_update: StateUpdate,
) -> StateUpdateResult

Apply one State Update Function and validate its returned key.

Calls function(params, substep, state_history, substate, signals) and checks that the returned key matches the declared State Variable and exists in the initial state.

Returns:

Type	Description
StateUpdateResult	The (key, value) produced by the function.

Raises:

Type	Description
KeyError	If the declared or returned key is unknown, or they differ.

Source code in radcad/core.py

def update_state(
    self,
    substate: StateVariables,
    signals: PolicySignal,
    state_update: StateUpdate,
) -> StateUpdateResult:
    """Apply one State Update Function and validate its returned key.

    Calls ``function(params, substep, state_history, substate, signals)``
    and checks that the returned key matches the declared State Variable
    and exists in the initial state.

    Returns:
        StateUpdateResult: The ``(key, value)`` produced by the function.

    Raises:
        KeyError: If the declared or returned key is unknown, or they differ.
    """
    _substate = self.deepcopy_method(substate) if self.enable_deepcopy else substate.copy()
    _signals = self.deepcopy_method(signals) if self.enable_deepcopy else signals.copy()

    state, function = state_update
    if state not in self.initial_state:
        raise KeyError(f"Invalid state key {state} in partial state update block")
    state_key, state_value = function(
        self.params, self.substep, self.result, _substate, _signals
    )
    if state_key not in self.initial_state:
        raise KeyError(
            f"Invalid state key {state} returned from state update function"
        )
    if state == state_key:
        return state_key, state_value
    else:
        raise KeyError(
            f"PSU state key {state} doesn't match function state key {state_key}"
        )

add_signals staticmethod

add_signals(
    acc: PolicySignal, a: PolicySignal
) -> PolicySignal

Reduce two Policy Signals by summing values that share a key.

Source code in radcad/core.py

@staticmethod
def add_signals(acc: PolicySignal, a: PolicySignal) -> PolicySignal:
    """Reduce two Policy Signals by summing values that share a key."""
    for (key, value) in a.items():
        if acc.get(key, None):
            acc[key] += value
        else:
            acc[key] = value
    return acc

deepcopy_method staticmethod

deepcopy_method(obj: Any) -> Any

The method used for simulation deepcopy operations

Source code in radcad/core.py

@staticmethod
def deepcopy_method(obj: Any) -> Any:
    """
    The method used for simulation deepcopy operations
    """
    return pickle.loads(pickle.dumps(obj=obj, protocol=-1))

SimulationExecutionSpecification dataclass

SimulationExecutionSpecification(
    timesteps: int = 0,
    timestep: int = None,
    substep_index: int = 0,
    substeps: List = default([]),
    state_update_blocks: List[StateUpdateBlock] = default(
        []
    ),
    result: SimulationResults = default([]),
)

Bases: ABC

Abstract template for the simulation loop, independent of state semantics.

Defines the nested execution structure (execution > timestep > substep) as a sequence of overridable hook methods (before_step, substep, after_step, ...). SimulationExecution implements these hooks with radCAD's concrete state-transition behaviour. Separating the two makes the loop structure explicit and the behaviour easy to customise.

execute

execute() -> SimulationResults

Run the full simulation loop and return the accumulated result.

Calls before_execution, iterates over each timestep (which in turn iterates over substeps via step), then after_execution.

Returns:

Type	Description
SimulationResults	The per-timestep, per-substep state history.

Source code in radcad/core.py

def execute(
    self,
) -> SimulationResults:
    """Run the full simulation loop and return the accumulated result.

    Calls `before_execution`, iterates over each timestep (which in
    turn iterates over substeps via `step`), then
    `after_execution`.

    Returns:
        SimulationResults: The per-timestep, per-substep state history.
    """
    self.before_execution()

    initial_timestep = self.timestep if self.timestep else 0
    for timestep in range(initial_timestep, initial_timestep + self.timesteps):
        self.timestep = timestep
        self.before_step()
        self.step()
        self.after_step()

    self.after_execution()
    return self.result

step

step() -> None

Advance the system by one timestep, evaluating every substep.

Each Partial State Update Block in state_update_blocks is one substep; they are evaluated in order, each building on the previous.

Source code in radcad/core.py

def step(self) -> None:
    """Advance the system by one timestep, evaluating every substep.

    Each Partial State Update Block in ``state_update_blocks`` is one
    substep; they are evaluated in order, each building on the previous.
    """
    self.substeps: List = []
    for (substep, state_update_block) in enumerate(self.state_update_blocks):
        self.substep_index = substep
        self.before_substep()
        self.substep(state_update_block)
        self.after_substep()

multiprocess_wrapper

multiprocess_wrapper(
    simulation_execution: SimulationExecution,
)

Execute one SimulationExecution, handling exceptions per policy.

This is the function dispatched to worker processes by every backend. If raise_exceptions is True the exception propagates; otherwise the partial result up to the failing timestep is returned alongside an exception/traceback metadata dict.

Parameters:

Name	Type	Description	Default
simulation_execution	SimulationExecution	The run to execute.	required

Returns:

Type	Description
tuple	(result, exception_metadata), the per-substep result and a dict describing any exception (with exception/traceback set to None on success).

Source code in radcad/core.py

def multiprocess_wrapper(simulation_execution: SimulationExecution):
    """Execute one [SimulationExecution][radcad.core.SimulationExecution], handling exceptions per policy.

    This is the function dispatched to worker processes by every backend. If
    ``raise_exceptions`` is ``True`` the exception propagates; otherwise the
    partial result up to the failing timestep is returned alongside an
    exception/traceback metadata dict.

    Args:
        simulation_execution (SimulationExecution): The run to execute.

    Returns:
        tuple: ``(result, exception_metadata)``, the per-substep result and a
        dict describing any exception (with ``exception``/``traceback`` set to
        ``None`` on success).
    """
    exception = None
    trace = None
    try:
        try:
            simulation_execution.execute()
        except Exception as _exception:
            trace = traceback.format_exc()
            exception = _exception
            print(trace)
            raise_exceptions_message = (
              'Catching exception and returning partial results because option Engine.raise_exceptions == False.'
              if simulation_execution.raise_exceptions
              else 'Raising exception because option Engine.raise_exceptions == True (set to False to catch exception and return partial results).'
            )
            simulation_execution.logger.warning(
                f"""Simulation {
                    simulation_execution.simulation_index
                } / run {
                    simulation_execution.run_index
                } / subset {
                    simulation_execution.subset_index
                } failed!
                {raise_exceptions_message}"""
            )
        if simulation_execution.raise_exceptions and exception:
            raise exception
        else:
            return simulation_execution.result, {
                'exception': exception,
                'traceback': trace,
                'simulation': simulation_execution.simulation_index,
                'run': simulation_execution.run_index,
                'subset': simulation_execution.subset_index,
                'timesteps': simulation_execution.timesteps,
                'parameters': simulation_execution.params,
                'initial_state': simulation_execution.initial_state,
            }
    except Exception as e:
        """
        Fail safe by catching catch any outer exceptions
        """
        if simulation_execution.raise_exceptions:
            raise e
        else:
            return [], e

radcad.types

Type aliases describing radCAD's core data structures.

radcad.types

Type aliases describing radCAD's core data structures.

These aliases name the data shapes that flow through a simulation: the State Variables, System Parameters, Policy Signals, and Partial State Update Blocks. Use them as type hints in model functions for clearer, self-documenting models.

StateUpdateBlock module-attribute

StateUpdateBlock = Dict[str, Dict[str, Callable]]

A Partial State Update Block: {"policies": {...}, "variables": {...}}.

SystemParameters module-attribute

SystemParameters = Union[Dict[str, List[Any]], Dataclass]

System Parameters as a dict of lists (sweepable) or a (nested) dataclass.

StateVariables module-attribute

StateVariables = Dict[str, Any]

A mapping of State Variable name to value at a single substep.

SimulationResults module-attribute

SimulationResults = List[List[Dict[str, Any]]]

Raw nested results: a list of timesteps, each a list of substep states.

PolicySignal module-attribute

PolicySignal = Dict[str, Any]

The aggregated output of a block's Policy Functions for one substep.

StateUpdate module-attribute

StateUpdate = Tuple[str, Callable]

A (state_variable_name, state_update_function) pair.

StateUpdateResult module-attribute

StateUpdateResult = Tuple[str, Any]

The (key, value) returned by a State Update Function.

Dataclass

Bases: Protocol

Structural type matching any @dataclass instance.

Used so System Parameters can be either a plain dict or a (nested) dataclass. Membership is detected via the __dataclass_fields__ attribute, currently the most reliable check.

Context dataclass

Context(
    simulation: int = None,
    run: int = None,
    subset: int = None,
    timesteps: int = None,
    initial_state: StateVariables = None,
    parameters: SystemParameters = None,
    state_update_blocks: List[StateUpdateBlock] = None,
)

Mutable context passed to simulation hooks

radcad.utils

Helpers for parameter sweeps, common State Update Functions, and dataclass parameters.

radcad.utils

Helper functions for parameter sweeps, common State Update Functions, and dataclass params.

The most commonly used helpers here are the generic State Update Functions (update_from_signal, accumulate_from_signal, update_timestamp) that remove boilerplate from models, and default, required when giving a dataclass field a mutable default such as a list or nested dataclass. The remaining functions implement radCAD's parameter-sweep expansion and are used internally by the Engine.

update_from_signal

update_from_signal(
    state_variable, signal_key=None, optional_update=False
)

A generic State Update Function to update a State Variable directly from a Policy Signal, avoiding boilerplate.

Parameters:

Name	Type	Description	Default
state_variable	str	State Variable key.	required
signal_key	str	Policy Signal key. Defaults to the State Variable key.	None
optional_update	bool	If True, only update the State Variable when the Policy Signal key exists.	False

Returns:

Type	Description
Callable	A generic State Update Function.

Source code in radcad/utils.py

def update_from_signal(state_variable, signal_key=None, optional_update=False):
    """A generic State Update Function to update a State Variable directly from a
    Policy Signal, avoiding boilerplate.

    Args:
        state_variable (str): State Variable key.
        signal_key (str, optional): Policy Signal key. Defaults to the State Variable key.
        optional_update (bool, optional): If True, only update the State Variable when the Policy Signal key exists.

    Returns:
        Callable: A generic State Update Function.
    """
    if not signal_key:
        signal_key = state_variable
    return partial(_update_from_signal, state_variable, signal_key, optional_update)

accumulate_from_signal

accumulate_from_signal(state_variable, signal_key=None)

A generic State Update Function to accumulate a State Variable directly from a Policy Signal, useful to avoid boilerplate code.

Source code in radcad/utils.py

def accumulate_from_signal(state_variable, signal_key=None):
    """
    A generic State Update Function to accumulate a State Variable directly from a Policy Signal,
    useful to avoid boilerplate code.
    """
    if not signal_key:
        signal_key = state_variable
    return partial(_accumulate_from_signal, state_variable, signal_key)

update_timestamp

update_timestamp(
    params,
    substep,
    state_history,
    previous_state,
    policy_input,
)

A radCAD State Update Function used to calculate and update the current timestamp given a timestep and starting date parameter.

Source code in radcad/utils.py

def update_timestamp(params, substep, state_history, previous_state, policy_input):
    """
    A radCAD State Update Function used to calculate and update the current timestamp
    given a timestep and starting date parameter.
    """
    # Parameters
    dt = params["dt"]
    date_start = params["date_start"]

    # State Variables
    timestep = previous_state["timestep"]

    # Calculate current timestamp from timestep
    timestamp = date_start + datetime.timedelta(days=timestep * dt)

    return "timestamp", timestamp

generate_parameter_sweep

generate_parameter_sweep(
    params: SystemParameters,
) -> List[SystemParameters]

Expand System Parameters into one parameter set per sweep subset.

Each parameter is a list of values; the sweep length is the longest such list, and shorter lists are extended by repeating their last value. Works with both dict and (nested) dataclass parameters.

Parameters:

Name	Type	Description	Default
params	SystemParameters	The parameters to expand.	required

Returns:

Type	Description
list	One concrete parameter set (same type as params) per subset.

Source code in radcad/utils.py

def generate_parameter_sweep(params: SystemParameters) -> List[SystemParameters]:
    """Expand System Parameters into one parameter set per sweep subset.

    Each parameter is a list of values; the sweep length is the longest such
    list, and shorter lists are extended by repeating their last value. Works
    with both dict and (nested) dataclass parameters.

    Args:
        params (SystemParameters): The parameters to expand.

    Returns:
        list: One concrete parameter set (same type as ``params``) per subset.
    """
    max_len = _get_sweep_length(params)
    param_sweep = []
    for sweep_index in range(0, max_len):
        param_set = _traverse_sweep_params(params, max_len, sweep_index)
        param_sweep.append(param_set)
    return param_sweep

generate_cartesian_product_parameter_sweep

generate_cartesian_product_parameter_sweep(params)

Source code in radcad/utils.py

def generate_cartesian_product_parameter_sweep(params):
    cartesian_product = list(itertools.product(*params.values()))
    param_sweep = {key: [x[i] for x in cartesian_product] for i, key in enumerate(params.keys())}
    return param_sweep

default

default(obj)

Used and necessary when setting the default value of a dataclass field to a list.

Source code in radcad/utils.py

def default(obj):
    """Used and necessary when setting the default value of a dataclass field to a list."""
    return field(default_factory=lambda: copy.copy(obj))

flatten

flatten(nested_list)

Flatten a list one level deep, leaving non-list items in place.

Source code in radcad/utils.py

def flatten(nested_list):
    """Flatten a list one level deep, leaving non-list items in place."""
    def generator(nested_list):
        for sublist in nested_list:
            if isinstance(sublist, list):
                for item in sublist:
                    yield item
            else:
                yield sublist

    return list(generator(nested_list))

extend_list

extend_list(list, target_length)

Extend list to length by repeating last element

Source code in radcad/utils.py

def extend_list(list, target_length):
    """Extend list to length by repeating last element"""
    if len(list) >= target_length:
        return list
    else:
        extension = [list[-1] if list else None] * (target_length - len(list))
        return list + extension

extract_exceptions

extract_exceptions(results_with_exceptions)

Split backend output into a flat results list and an exceptions list.

Source code in radcad/utils.py

def extract_exceptions(results_with_exceptions):
    """Split backend output into a flat results list and an exceptions list."""
    results, exceptions = zip(*results_with_exceptions)
    return (list(flatten(flatten(list(results)))), list(exceptions))

local_variables

local_variables(_locals)

Return a dictionary of all local variables, useful for debugging.

Source code in radcad/utils.py

def local_variables(_locals):
    """Return a dictionary of all local variables, useful for debugging."""
    return {key: _locals[key] for key in [_key for _key in _locals.keys() if "__" not in _key]}