chain_simulator.utilities module#

Various utilities for validating and processing simulation output.

This module provides various utilities for validating transition matrices and for processing output from a simulation. Validation utilities can be used to reduce chances of simulating a faulty transition matrix. The accumulator can be used to process state vector from a simulation, e.g. to calculate phenotype data.

exception chain_simulator.utilities.AccumulationError#

Bases: Exception

Something went wrong while accumulating callback values.

exception chain_simulator.utilities.CallbackError#

Bases: Exception

Something went wrong inside a callback function.

exception chain_simulator.utilities.NoCallbackWarning#

Bases: Warning

There are no callback functions provided.

exception chain_simulator.utilities.TransitionMatrixNegativeWarning#

Bases: TransitionMatrixWarning

Some probabilities in a transition matrix are negative.

exception chain_simulator.utilities.TransitionMatrixSumWarning#

Bases: TransitionMatrixWarning

Not all rows in a transition matrix sum to 1.

chain_simulator.utilities.simulation_accumulator(vector_processor, **callbacks)#

Accumulate simulation data using callback functions.

Accumulate data from intermediate/final state vectors using callback functions. Callback functions must accept a state vector, the current step in time and the step size (compated to the previous step). The function can return either nothing, an int or float. These returned values are accumulated and summed per callback function and are returned after the simulation has finished.

Parameters:

vector_processorIterator[Tuple[numpy.ndarray, int]]: Iterator yielding a state vector and the current step in time.
**callbacksCallable[[numpy.ndarray, int], None | int | float]: Callable accepting a state vector and the current step in time and returning either nothing, an int or float.

Returns:

Dict[str, None | int | float]: Dictionary with the name of the callback function and corresponding accumulated result.

Raises:

TypeError: If the processor or callback functions have incorrect types.
CallbackError: If the callback function raises an exception.
AccumulationError: If the return value from callback function cannot be summed.

Warns:

NoCallbackWarning: If there are no callback specified.

Parameters:

vector_processor (Iterator[Tuple[STATE_VECTOR, int]]) –
callbacks (Callable[[STATE_VECTOR, int, int], None | int | float]) –

Return type:

Dict[str, None | int | float]

Warning

This function serves as a wrapper for chain_simulator.simulation.state_vector_processor(). The creation of this processor should be handled by the caller of this wrapper function. It is important that the processor was not iterated upon before calling this wrapper function. Otherwise, some state vectors cannot be processed using this accumulator.

See also

chain_simulator.simulation.state_vector_processor: Simulate a Markov chain and return intermediate/final state vector(s).

Notes

Callback functions are accepted as keyword-arguments, meaning that they can be provided as key-value pairs or unpacked from a dictionary.

Callback functions are not required to return anything. If nothing is returned, no accumulator is created. Make sure to access accumulators using .get(<callback_name>) instead of [<callback_name>] to check whether an accumulator was created.

Callback functions should have the following signature:

def callback_function(
    state_vector: numpy.ndarray, step_in_time: int, step_size: int
) -> None | int | float:
    ...

Intermediate/final state vectors are provided as-is from the simulation.

Examples

Provide callbacks as keyword arguments:

>>> from numpy import array, sum
>>> from chain_simulator.simulation import state_vector_processor
>>> state_vector = array([1, 0, 0])
>>> transition_matrix = array(
...     [[0.0, 1.0, 0.0], [0.0, 0.5, 0.5], [0.0, 0.0, 1.0]]
... )
>>> processor = state_vector_processor(
...     state_vector, transition_matrix, 4, 1
... )
>>> accumulated = simulation_accumulator(
...     processor,
...     time_cumulative=lambda x, y, z: sum(x),
...     vector_sum=lambda x, y, z: y
... )
>>> accumulated
{'time_cumulative': 4.0, 'vector_sum': 10}

Or add callbacks to a dictionary and unpack them in the accumulator:

>>> from numpy import array, sum
>>> from chain_simulator.simulation import state_vector_processor
>>> state_vector = array([1, 0, 0])
>>> transition_matrix = array(
...     [[0.0, 1.0, 0.0], [0.0, 0.5, 0.5], [0.0, 0.0, 1.0]]
... )
>>> processor = state_vector_processor(
...     state_vector, transition_matrix, 4, 1
... )
>>> callback_functions = {
...     "time_cumulative": lambda x, y, z: sum(x),
...     "vector_sum": lambda x, y, z: y
... }
>>> accumulated = simulation_accumulator(
...     processor, **callback_functions
... )
>>> accumulated
{'time_cumulative': 4.0, 'vector_sum': 10}

Callbacks using constant extra parameters can be passed using functools.partial:

>>> from numpy import array
>>> from chain_simulator.simulation import state_vector_processor
>>> from functools import partial
>>> from typing import Callable
>>> state_vector = array([1, 0, 0])
>>> transition_matrix = array(
...     [[0.0, 1.0, 0.0], [0.0, 0.5, 0.5], [0.0, 0.0, 1.0]]
... )
>>> processor = state_vector_processor(
...     state_vector, transition_matrix, 4, 1
... )
>>> partial_callable: Callable = partial(lambda w, x, y, z: x * z, z=5)
>>> accumulated = simulation_accumulator(
...     processor, callable_with_constant=partial_callable
... )
>>> accumulated
{'callable_with_constant': 50}

chain_simulator.utilities.validate_matrix_negative(transition_matrix)#

Validate probabilities in a transition matrix for negative signs.

Checks whether all probabilities in transition_matrix are positive. In case all probabilities are positive, this function evaluates TRUE. In case one or more probabilities are negative, this function evaluates FALSE.

Parameters:

transition_matrixAny 2d NumPy array or SciPy COO/CSC/CSR array/matrix: A Markov chain transition matrix.

Returns:

bool: Indication whether all probabilities in transition_matrix are positive.

Warns:

TransitionMatrixNegativeWarning: If there are any negative probabilities.

Parameters:

transition_matrix (_T) –

Return type:

bool