Runtime

Run systems frame by frame with clocks, live parameter changes, observers, and trajectories.

AsFastAsPossibleClock

Bases: BaseClock

Clock that never waits.

Examples:

>>> from skcomplex.engine.runtime import AsFastAsPossibleClock
>>> clock = AsFastAsPossibleClock().check().prepare().run(10.0)
>>> clock.max_lag_
0.0

BaseClock

Bases: BaseConfig[None, float]

Base class of clocks, which decide how long to wait between frames.

A clock follows the common API of the engine. It is checked and prepared at the start of a run with prepare(), which marks the wall time, and run(elapsed) waits until a frame that is elapsed simulated time units after the start is due. Subclasses implement _run.

Attributes:

Name	Type	Description
started_		The wall time at which the run started, in seconds of time.perf_counter.
max_lag_		The largest delay of a frame behind its due time, in seconds.

prepare(context=None)

Mark the wall time at which the run of the checked clock starts.

Parameters:

Name	Type	Description	Default
context	None	Nothing: a clock prepares on no context.	None

Returns:

Name	Type	Description
self	Self	The prepared clock.

Source code in src/skcomplex/engine/runtime/_base.py

def prepare(self, context: None = None) -> Self:
    """Mark the wall time at which the run of the checked clock starts.

    Args:
        context:
            Nothing: a clock prepares on no context.

    Returns:
        self:
            The prepared clock.
    """
    return super().prepare(context)

run(elapsed=None, final=False)

Wait until a frame that is elapsed simulated time units after the start is due.

Parameters:

Name	Type	Description	Default
elapsed	float | None	The simulated time since the start of the run.	None
final	bool	Whether this is the last call of the run.	False

Returns:

Name	Type	Description
self	Self	The clock.

Source code in src/skcomplex/engine/runtime/_base.py

def run(self, elapsed: float | None = None, final: bool = False) -> Self:
    """Wait until a frame that is `elapsed` simulated time units after the start is due.

    Args:
        elapsed:
            The simulated time since the start of the run.

        final:
            Whether this is the last call of the run.

    Returns:
        self:
            The clock.
    """
    super().run(elapsed, final)
    return self

BaseObserver

Bases: BaseConfig[BaseProblemSpec, FrameSpec]

Base class of objects that receive the frames of a simulation.

An observer follows the common API of the engine. A simulation checks it, calls prepare(problem) with the compiled problem before the first frame, run(frame) with every frame, and run(final=True) without a frame once the run stops, which releases what preparing acquired. Subclasses override any of the hooks _prepare, _run, and _release.

Examples:

>>> from skcomplex.core.system import System
>>> from skcomplex.engine.runtime import BaseObserver, Simulation
>>> from skcomplex.models.biology.common.populations import PopulationElement
>>> class CountingObserver(BaseObserver):
...     def _prepare(self, context):
...         self.count_ = 0
...
...     def _run(self, inputs):
...         self.count_ += inputs is not None
>>> observer = CountingObserver().check()
>>> simulation = Simulation(System(elements=[PopulationElement('a')]), dt=1, observers=[observer])
>>> _ = simulation.check().prepare().run(until=3)
>>> observer.count_
4

prepare(problem=None)

Prepare the checked observer for a run of a compiled problem.

Parameters:

Name	Type	Description	Default
problem	BaseProblemSpec | None	The compiled problem the run advances.	None

Returns:

Name	Type	Description
self	Self	The prepared observer.

Source code in src/skcomplex/engine/runtime/_base.py

def prepare(self, problem: BaseProblemSpec | None = None) -> Self:
    """Prepare the checked observer for a run of a compiled problem.

    Args:
        problem:
            The compiled problem the run advances.

    Returns:
        self:
            The prepared observer.
    """
    return super().prepare(problem)

run(frame=None, final=False)

React to a frame, and release what preparing acquired at the last call.

Parameters:

Name	Type	Description	Default
frame	FrameSpec | None	The frame, or None for the call that ends the run.	None
final	bool	Whether this is the last call of the run.	False

Returns:

Name	Type	Description
self	Self	The observer.

Source code in src/skcomplex/engine/runtime/_base.py

def run(self, frame: FrameSpec | None = None, final: bool = False) -> Self:
    """React to a frame, and release what preparing acquired at the last call.

    Args:
        frame:
            The frame, or `None` for the call that ends the run.

        final:
            Whether this is the last call of the run.

    Returns:
        self:
            The observer.
    """
    super().run(frame, final)
    return self

ConservationObserver(*, rtol=1e-06, atol=1e-09)

Bases: BaseObserver

Observer that checks at every frame that each quantity changes only by its flows with the environment.

The total of a quantity is the sum of the states that hold it. The monitor compares its change since the first frame with the accumulated net inflow from the environment, which the compiler integrates as a hidden state, and raises an error when they differ by more than the tolerance, such as after a boundary law clips a state.

Parameters:

Name	Type	Description	Default
rtol	float	The relative tolerance, a fraction of the largest absolute total seen.	1e-06
atol	float	The absolute tolerance.	1e-09

Attributes:

Name	Type	Description
problem_		The compiled problem of the run.
transfer_network_		The quantities of the problem, with the states that hold them and their flows with the environment.
totals_		The total of every quantity at the latest frame.
initial_		The total of every quantity minus its inflow from the environment at the first frame, or None.
scale_		The largest absolute total of every quantity seen so far.

Examples:

>>> from skcomplex.engine.runtime import ConservationObserver, Simulation
>>> from skcomplex.models import create
>>> monitor = ConservationObserver()
>>> _ = Simulation(create('bathtub'), dt=1, observers=[monitor]).check().prepare().run(10)
>>> round(monitor.totals_['water'], 6)
25.0

Source code in src/skcomplex/engine/runtime/_conservation.py

def __init__(self, *, rtol: float = 1e-6, atol: float = 1e-9) -> None:
    self.rtol = rtol
    self.atol = atol

InvariantObserver(invariants)

Bases: BaseObserver

Observer that checks at every frame that every named observable stays within its tolerance of its first value.

A simulation attaches one when it is given invariants.

Parameters:

Name	Type	Description	Default
invariants	Mapping[str, float]	The tolerance of every invariant, by the path of its observable.	required

Attributes:

Name	Type	Description
initial_		The value of every invariant at the first frame.

Examples:

>>> from skcomplex.core.elements import GroupElement
>>> from skcomplex.specs import ObservableSpec
>>> from skcomplex.core.system import System
>>> from skcomplex.engine.runtime import InvariantObserver, Simulation
>>> from skcomplex.models.biology.common.populations import PopulationElement
>>> system = System(
...     elements=[GroupElement('herd', element=PopulationElement, size=2, values={'n': [2.0, 6.0]})]
... )
>>> total = ObservableSpec('total', group='herd', reduce='sum', variable='n')
>>> monitor = InvariantObserver({'total': 1e-9})
>>> _ = Simulation(system, dt=1, observables=[total], observers=[monitor]).check().prepare().run(3)
>>> monitor.initial_
{'total': 8.0}

Source code in src/skcomplex/engine/runtime/_invariants.py

def __init__(self, invariants: Mapping[str, float]) -> None:
    self.invariants = invariants

RealTimeClock(speed=1.0)

Bases: BaseClock

Clock that advances simulated time at a fixed speed relative to wall time.

The largest delay of a frame behind its due time is recorded in max_lag_.

Parameters:

Name	Type	Description	Default
speed	float	The number of simulated time units per second of wall time.	1.0

Examples:

>>> from skcomplex.engine.runtime import RealTimeClock
>>> clock = RealTimeClock(speed=100.0).check().prepare().run(1.0)
>>> clock.max_lag_ < 0.1
True

Source code in src/skcomplex/engine/runtime/_real_time.py

def __init__(self, speed: float = 1.0) -> None:
    self.speed = speed

RecorderObserver

Bases: BaseObserver

Observer that keeps every frame it receives.

Attributes:

Name	Type	Description
problem_		The compiled problem of the run.
frames_		The frames received so far.

Examples:

>>> from skcomplex.core.system import System
>>> from skcomplex.models.biology.common.populations import PopulationElement
>>> from skcomplex.engine.runtime import RecorderObserver, Simulation
>>> recorder = RecorderObserver()
>>> simulation = Simulation(System(elements=[PopulationElement('a')]), dt=1, observers=[recorder])
>>> _ = simulation.check().prepare().run(until=2)
>>> len(recorder.trajectory_.times)
3

trajectory_ property

Trajectory of the frames received so far.

Simulation(system, *, compiler=None, solver=None, backend=None, dt=None, clock=None, observers=(), record=True, random_state=None, checked=False, schedule=None, observables=(), invariants=None, initial_state=None)

Bases: BaseConfig[None, float]

Simulation of a system with the solver of its form.

Each frame advances simulated time by dt and is handed to every observer. Parameters can change between frames through update_parameters, also while a background run started with start is going. Values set between runs stay in force for the later runs of the simulation, and extend continues a run from its last frame.

A simulation composes the pipeline of the engine and follows its common API: check() validates the system, prepare() compiles it and chooses the solver, and run(until) prepares and runs from time 0 until a time. A run prepares the solver for its end time and every observer on the problem before the first frame, runs every observer at every frame, and runs it with final=True at the end.

Parameters:

Name	Type	Description	Default
system	System | BaseProblemSpec	The system to simulate, or a problem already compiled from one, which skips compiling for repeated runs that change only parameters or initial values.	required
compiler	BaseCompiler | None	The compiler of the system, whose backend, solver, and observables the simulation sets, or None for PassCompiler. An operator system takes OperatorCompiler(), and a model package can bring the compiler of a form of its own, which names the solver of its problems.	None
solver	BaseSolver | None	The solver, or None for the solver the compiler names, or else the default solver of the system's form, CoSimulationSolver for a system that runs as partitions. The default ODE solver is ScipyODESolver with LSODA, which switches to BDF methods while the system is stiff.	None
backend	BaseBackend | None	The backend that compiles the system, or None for NumbaBackend on systems with at least 1000 members and NumpyBackend on the others.	None
dt	float | None	The simulated time between frames. It defaults to 1 for discrete maps and agent-based models. Otherwise it defaults to 0.01, raised to the smallest of 0.02, 0.05, 0.1, 0.2, and so on, in steps of 1, 2, and 5 times a power of ten, that keeps a run at most 1000 frames long.	None
clock	BaseClock | None	The clock that paces frames against wall time. It defaults to running as fast as possible.	None
observers	Sequence[BaseObserver]	The observers that receive every frame.	()
record	bool	Whether to record the frames as the trajectory trajectory_.	True
random_state	int | Generator | None	The seed or generator of stochastic solvers. A seed starts a stream spawned from np.random.SeedSequence(random_state), independent of np.random.default_rng(random_state), so a model can draw its initial state from the same seed; a generator is used as it is.	None
checked	bool	Whether to check after every frame that every state lies in its domain, such as an alphabet of labels, and raise an error naming the state when it does not.	False
schedule	Mapping[str, Callable[[float], float | ArrayLike]] | None	A mapping from dotted parameter name to a function of time, such as a ramp of a control parameter. At every frame the parameter takes the value of its function at the frame's time and keeps it until the next frame. An adaptive parameter, such as a bounded control input, can be scheduled too, and the trajectory records its scheduled values.	None
observables	Sequence[ObservableSpec]	The observables to compute and record at every frame, labelled by their path from the root, such as village/mean_wealth for an observable of the nested system village.	()
invariants	Mapping[str, float] | None	The tolerance of every observable that must stay constant, by the path of the observable. At every frame the run checks that each stays within its tolerance of its value at the first frame, and raises a ConservationError that names the observable and the first frame that breaks it.	None
initial_state	Mapping[str, object] | None	Initial values that replace those of the elements, by dotted state name, such as {'rabbits.n': 20}. The whole state of an operator system is <system label>.state, which takes a vector, a density matrix, or 'ground' for the ground state.	None

Attributes:

Name	Type	Description
problem_		The compiled problem.
stepper_		The stepper that advances the problem.
time_		The simulated time of the last frame.
state_		The state vector of the last frame.
parameters_		The current parameter vector.

Examples:

>>> from skcomplex.core.system import System
>>> from skcomplex.models.biology.common.populations import LogisticGrowthInteraction, PopulationElement
>>> from skcomplex.engine.runtime import Simulation
>>> system = System(
...     elements=[PopulationElement('rabbits', n=10)],
...     interactions=[LogisticGrowthInteraction('growth', target='rabbits', r=1.0, K=100)],
... )
>>> simulation = Simulation(system, dt=1.0).check().prepare().run(until=20)
>>> round(simulation.trajectory_.values('rabbits.n')[-1])
100

Source code in src/skcomplex/engine/runtime/_simulations.py

def __init__(
    self,
    system: System | BaseProblemSpec,
    *,
    compiler: BaseCompiler | None = None,
    solver: BaseSolver | None = None,
    backend: BaseBackend | None = None,
    dt: float | None = None,
    clock: BaseClock | None = None,
    observers: Sequence[BaseObserver] = (),
    record: bool = True,
    random_state: int | np.random.Generator | None = None,
    checked: bool = False,
    schedule: Mapping[str, Callable[[float], float | npt.ArrayLike]] | None = None,
    observables: Sequence[ObservableSpec] = (),
    invariants: Mapping[str, float] | None = None,
    initial_state: Mapping[str, object] | None = None,
) -> None:
    self.system = system
    self.compiler = compiler
    self.solver = solver
    self.backend = backend
    self.dt = dt
    self.clock = clock
    self.observers = observers
    self.record = record
    self.random_state = random_state
    self.checked = checked
    self.schedule = schedule
    self.observables = observables
    self.invariants = invariants
    self.initial_state = initial_state

trajectory_ property

Trajectory of the frames recorded so far.

extend(until)

Continue a finished run from its last frame to until in the calling thread.

The run keeps its state, parameters, stepper, clock, observers, and frame duration, applies pending parameter changes, and appends its frames to the trajectory. A solver that closes an external process at the end of a run, such as LammpsSolver, cannot continue it.

Parameters:

Name	Type	Description	Default
until	float	The simulated time to stop at, later than the time of the last frame.	required

Returns:

Name	Type	Description
simulation	Self	The simulation itself.

Raises:

Type	Description
SimulationError	The simulation has not run, is running in the background, ended at a terminal event, or until is not later than its time.

Examples:

>>> from skcomplex.core.system import System
>>> from skcomplex.models.biology.common.populations import LogisticGrowthInteraction, PopulationElement
>>> from skcomplex.engine.runtime import Simulation
>>> system = System(
...     elements=[PopulationElement('rabbits', n=10)],
...     interactions=[LogisticGrowthInteraction('growth', target='rabbits', r=1.0, K=100)],
... )
>>> simulation = Simulation(system, dt=1.0).check().prepare().run(until=5)
>>> simulation.update_parameters({'growth.r': 0.0})
>>> trajectory = simulation.extend(until=10).trajectory_
>>> trajectory.times.tolist()
[0.0, 1.0, 2.0, 3.0, 4.0, 5.0, 6.0, 7.0, 8.0, 9.0, 10.0]
>>> bool(trajectory.values('rabbits.n')[5] == trajectory.values('rabbits.n')[10])
True

Source code in src/skcomplex/engine/runtime/_simulations.py

def extend(self, until: float) -> Self:
    """Continue a finished run from its last frame to `until` in the calling thread.

    The run keeps its state, parameters, stepper, clock, observers, and frame duration, applies pending parameter
    changes, and appends its frames to the trajectory. A solver that closes an external process at the end of a
    run, such as `LammpsSolver`, cannot continue it.

    Args:
        until:
            The simulated time to stop at, later than the time of the last frame.

    Returns:
        simulation:
            The simulation itself.

    Raises:
        SimulationError:
            The simulation has not run, is running in the background, ended at a terminal event, or `until` is not
            later than its time.

    Examples:
        >>> from skcomplex.core.system import System
        >>> from skcomplex.models.biology.common.populations import LogisticGrowthInteraction, PopulationElement
        >>> from skcomplex.engine.runtime import Simulation
        >>> system = System(
        ...     elements=[PopulationElement('rabbits', n=10)],
        ...     interactions=[LogisticGrowthInteraction('growth', target='rabbits', r=1.0, K=100)],
        ... )
        >>> simulation = Simulation(system, dt=1.0).check().prepare().run(until=5)
        >>> simulation.update_parameters({'growth.r': 0.0})
        >>> trajectory = simulation.extend(until=10).trajectory_
        >>> trajectory.times.tolist()
        [0.0, 1.0, 2.0, 3.0, 4.0, 5.0, 6.0, 7.0, 8.0, 9.0, 10.0]
        >>> bool(trajectory.values('rabbits.n')[5] == trajectory.values('rabbits.n')[10])
        True
    """
    self._check_prepared()
    if not hasattr(self, 'time_'):
        error_msg = 'The simulation has been prepared but not run. Call `run` before `extend`.'
        raise SimulationError(error_msg)
    if self._thread is not None and self._thread.is_alive():
        error_msg = 'The simulation is running in the background. Call `wait` or `stop` before `extend`.'
        raise SimulationError(error_msg)
    if not until > self.time_ + TIME_TOLERANCE:
        error_msg = f'Cannot extend the run to t = {until}, which is not later than its current time {self.time_}.'
        raise SimulationError(error_msg)
    if self.stepper_.stopped_at_ is not None:
        error_msg = f'A terminal event ended the run at t = {self.time_}, so it cannot be extended.'
        raise SimulationError(error_msg)
    self._stop_requested.clear()
    self._running.set()
    self._loop(until)
    return self

pause()

Pause a background run before its next frame.

Source code in src/skcomplex/engine/runtime/_simulations.py

def pause(self) -> None:
    """Pause a background run before its next frame."""
    self._check_prepared()
    self._running.clear()

prepare(context=None)

Compile the system of the checked simulation and choose the solver.

Parameters:

Name	Type	Description	Default
context	None	Nothing: a simulation prepares on its own system.	None

Returns:

Name	Type	Description
self	Self	The prepared simulation, with the compiled problem problem_ and the solver stepper_.

Source code in src/skcomplex/engine/runtime/_simulations.py

def prepare(self, context: None = None) -> Self:
    """Compile the system of the checked simulation and choose the solver.

    Args:
        context:
            Nothing: a simulation prepares on its own system.

    Returns:
        self:
            The prepared simulation, with the compiled problem `problem_` and the solver `stepper_`.
    """
    return super().prepare(context)

resume()

Resume a paused background run.

Source code in src/skcomplex/engine/runtime/_simulations.py

def resume(self) -> None:
    """Resume a paused background run."""
    self._check_prepared()
    self._running.set()

run(until=None, final=False)

Run the prepared simulation from time 0 to until in the calling thread.

Parameters:

Name	Type	Description	Default
until	float | None	The simulated time to stop at.	None
final	bool	Whether this is the last call, after which the simulation releases what preparing acquired.	False

Returns:

Name	Type	Description
simulation	Self	The simulation itself.

Raises:

Type	Description
NotPreparedError	The simulation was not prepared with prepare.
ValidationError	No end time is passed.

Source code in src/skcomplex/engine/runtime/_simulations.py

def run(self, until: float | None = None, final: bool = False) -> Self:
    """Run the prepared simulation from time 0 to `until` in the calling thread.

    Args:
        until:
            The simulated time to stop at.

        final:
            Whether this is the last call, after which the simulation releases what preparing acquired.

    Returns:
        simulation:
            The simulation itself.

    Raises:
        NotPreparedError:
            The simulation was not prepared with `prepare`.
        ValidationError:
            No end time is passed.
    """
    super().run(until, final)
    return self

start(until=math.inf)

Start the simulation from time 0 in a background thread.

Parameters:

Name	Type	Description	Default
until	float	The simulated time to stop at. The default runs until stop is called.	inf

Returns:

Name	Type	Description
simulation	Self	The simulation itself.

Source code in src/skcomplex/engine/runtime/_simulations.py

def start(self, until: float = math.inf) -> Self:
    """Start the simulation from time 0 in a background thread.

    Args:
        until:
            The simulated time to stop at. The default runs until `stop` is called.

    Returns:
        simulation:
            The simulation itself.
    """
    self.prepare()
    self._start(until)
    self._thread = threading.Thread(target=self._run_in_background, args=(until,), daemon=True)
    self._thread.start()
    return self

stop()

Stop a background run after its current frame and wait for its thread to end.

Source code in src/skcomplex/engine/runtime/_simulations.py

def stop(self) -> None:
    """Stop a background run after its current frame and wait for its thread to end."""
    self._check_prepared()
    self._stop_requested.set()
    self._running.set()
    self.wait()

update_parameters(values)

Set parameter values by dotted name, applied before the next frame.

Parameters:

Name	Type	Description	Default
values	Mapping[str, float | ArrayLike]	A mapping from dotted parameter name, such as growth.r, or name of one component, such as regulation.weights[1], to its new value.	required

Raises:

Type	Description
ParameterError	A name is not a parameter or a parameter component of the problem, or a value would make a transition or emission matrix negative or its rows not sum to 1.
ValidationError	A value has the wrong shape or is outside the domain of its parameter.

Source code in src/skcomplex/engine/runtime/_simulations.py

def update_parameters(self, values: Mapping[str, float | npt.ArrayLike]) -> None:
    """Set parameter values by dotted name, applied before the next frame.

    Args:
        values:
            A mapping from dotted parameter name, such as `growth.r`, or name of one component, such as
            `regulation.weights[1]`, to its new value.

    Raises:
        ParameterError:
            A name is not a parameter or a parameter component of the problem, or a value would make a
            transition or emission matrix negative or its rows not sum to 1.

        ValidationError:
            A value has the wrong shape or is outside the domain of its parameter.
    """
    self._check_prepared()
    layout = self.problem_.parameters
    changes = []
    for name, value in values.items():
        try:
            entry, indices = layout.locate(name)
        except KeyError as error:
            error_msg = f'No parameter named {name!r}. Available names are: {", ".join(layout.names)}.'
            raise ParameterError(error_msg) from error
        changes.append((name, indices, self._check_value(name, entry, value)))
    self._check_matrices(changes)
    for _, indices, value in changes:
        self._pending.put((indices, value))
    if self._thread is None or not self._thread.is_alive():
        self._apply_pending()

wait(timeout=None)

Wait for a background run to finish and raise any error it met.

Parameters:

Name	Type	Description	Default
timeout	float | None	The longest time in seconds to wait, or None to wait until the run ends.	None

Raises:

Type	Description
BaseException	The error that ended the background run, if any.

Source code in src/skcomplex/engine/runtime/_simulations.py

def wait(self, timeout: float | None = None) -> None:
    """Wait for a background run to finish and raise any error it met.

    Args:
        timeout:
            The longest time in seconds to wait, or `None` to wait until the run ends.

    Raises:
        BaseException:
            The error that ended the background run, if any.
    """
    self._check_prepared()
    if self._thread is not None:
        self._thread.join(timeout)
    if self._error is not None:
        error, self._error = self._error, None
        raise error

Trajectory(problem, times, states, parameters, *, agents=(), reporters=(), observables=(), events=(), structures=())

Recorded states and parameters of a simulation over time.

Parameters:

Name	Type	Description	Default
problem	BaseProblemSpec	The compiled problem, which gives the names of the vector components.	required
times	NDArray[float64]	The time of each recorded frame.	required
states	NDArray[float64]	One state vector per row.	required
parameters	NDArray[float64]	One parameter vector per row.	required
agents	Sequence[Mapping[str, AgentSnapshotSpec]]	One mapping from agent group path to agents per frame, empty for problems without agents.	()
reporters	Sequence[Mapping[str, float]]	One mapping from dotted reporter name to value per frame, empty for problems without reporters.	()
observables	Sequence[Mapping[str, object]]	One mapping from observable path to value per frame, empty for systems without observables.	()
events	Sequence[BaseRecordSpec]	The events that fired during the run, in order, each with its time and the states before and after its reset, the cascades of lattices with their sizes, durations, and areas, and the renewals of renewal laws with their members.	()
structures	Sequence[Mapping[str, object]]	One mapping from rewritten element path to its structure per frame, empty for other systems.	()

Examples:

>>> from skcomplex.core.system import System
>>> from skcomplex.models.biology.common.populations import ExponentialGrowthInteraction, PopulationElement
>>> from skcomplex.engine.runtime import Simulation
>>> system = System(
...     elements=[PopulationElement('a', n=1)],
...     interactions=[ExponentialGrowthInteraction('g', target='a', r=1)],
... )
>>> trajectory = Simulation(system, dt=0.5).check().prepare().run(until=1).trajectory_
>>> trajectory.times.tolist()
[0.0, 0.5, 1.0]
>>> trajectory.values('a.n').round(3).tolist()
[1.0, 1.649, 2.718]

Source code in src/skcomplex/engine/runtime/_trajectories.py

def __init__(
    self,
    problem: BaseProblemSpec,
    times: npt.NDArray[np.float64],
    states: npt.NDArray[np.float64],
    parameters: npt.NDArray[np.float64],
    *,
    agents: Sequence[Mapping[str, AgentSnapshotSpec]] = (),
    reporters: Sequence[Mapping[str, float]] = (),
    observables: Sequence[Mapping[str, object]] = (),
    events: Sequence[BaseRecordSpec] = (),
    structures: Sequence[Mapping[str, object]] = (),
) -> None:
    self.problem = problem
    self.times = times
    self.states = states
    self.parameters = parameters
    self.agents_by_frame = agents
    self.reporters = reporters
    self.observables = observables
    self.events = list(events)
    self.structures = structures

names property

Names of the states, parameters, reporters, and observables that values reads.

agent_locations(group)

Return the locations of the agents of a group, one array per frame with one row per agent alive.

Parameters:

Name	Type	Description	Default
group	str	The path of the agent group.	required

Returns:

Name	Type	Description
locations	list[AnyArray]	The locations at every frame, empty arrays for frames without locations.

Examples:

>>> from skcomplex.engine.runtime import Simulation
>>> from skcomplex.models import create
>>> trajectory = Simulation(create('boids'), random_state=0).check().prepare().run(until=2).trajectory_
>>> [locations.shape[1] for locations in trajectory.agent_locations('flock')]
[2, 2, 2]

Source code in src/skcomplex/engine/runtime/_trajectories.py

def agent_locations(self, group: str) -> list[AnyArray]:
    """Return the locations of the agents of a group, one array per frame with one row per agent alive.

    Args:
        group:
            The path of the agent group.

    Returns:
        locations:
            The locations at every frame, empty arrays for frames without locations.

    Examples:
        >>> from skcomplex.engine.runtime import Simulation
        >>> from skcomplex.models import create
        >>> trajectory = Simulation(create('boids'), random_state=0).check().prepare().run(until=2).trajectory_
        >>> [locations.shape[1] for locations in trajectory.agent_locations('flock')]
        [2, 2, 2]
    """
    frames = [self.agents(group, frame) for frame in range(len(self.times))]
    return [
        np.asarray(snapshot.locations) if snapshot.locations is not None else np.zeros((0, 0))
        for snapshot in frames
    ]

agents(group, frame)

Return the agents of a group alive at a frame, with their ids, states, and locations.

Parameters:

Name	Type	Description	Default
group	str	The path of the agent group.	required
frame	int	The index of the frame.	required

Returns:

Name	Type	Description
agents	AgentSnapshotSpec	The agents of the group at that frame.

Raises:

Type	Description
KeyError	The trajectory has no agent group of that path.

Source code in src/skcomplex/engine/runtime/_trajectories.py

def agents(self, group: str, frame: int) -> AgentSnapshotSpec:
    """Return the agents of a group alive at a frame, with their ids, states, and locations.

    Args:
        group:
            The path of the agent group.

        frame:
            The index of the frame.

    Returns:
        agents:
            The agents of the group at that frame.

    Raises:
        KeyError:
            The trajectory has no agent group of that path.
    """
    groups = self.agents_by_frame[frame] if self.agents_by_frame else {}
    if group not in groups:
        error_msg = f'No agent group {group!r}. Available groups are: {", ".join(groups)}.'
        raise KeyError(error_msg)
    return groups[group]

lattice(name)

Return the values of a lattice state over time in lattice shape.

Parameters:

Name	Type	Description	Default
name	str	The dotted name of a state of a lattice group, such as cells.alive.	required

Returns:

Name	Type	Description
values	AnyArray	One array of the lattice shape per frame, with the components of a vector or a tensor state on a last axis.

Raises:

Type	Description
KeyError	The name is not a state of a lattice group.

Source code in src/skcomplex/engine/runtime/_trajectories.py

def lattice(self, name: str) -> AnyArray:
    """Return the values of a lattice state over time in lattice shape.

    Args:
        name:
            The dotted name of a state of a lattice group, such as `cells.alive`.

    Returns:
        values:
            One array of the lattice shape per frame, with the components of a vector or a tensor state on a last
            axis.

    Raises:
        KeyError:
            The name is not a state of a lattice group.
    """
    shape = self.problem.lattice_shape(name)
    if shape is None:
        error_msg = f'{name!r} is not a state of a lattice group.'
        raise KeyError(error_msg)
    return self.values(name).reshape(len(self.times), *shape, *self.problem.states.entry(name).member_shape)

member(index, problem)

Return the trajectory of one member of a replicated system, as a trajectory of the original system.

Parameters:

Name	Type	Description	Default
index	int	The member index, the same in every group of the replicated system.	required
problem	BaseProblemSpec	The compiled original system, whose single elements the replicated groups hold.	required

Returns:

Name	Type	Description
trajectory	Trajectory	The times, states, and parameters of the member, laid out as the original problem lays them out.

Examples:

>>> import numpy as np
>>> from skcomplex.engine.compile import compile_system
>>> from skcomplex.engine.runtime import Simulation
>>> from skcomplex.experiments import BaseExperiment
>>> from skcomplex.models.mathematics.dynamical_systems import logistic_map
>>> ensemble = BaseExperiment._replicate(logistic_map(), 3, {'map.x': [0.1, 0.2, 0.3]})
>>> trajectory = Simulation(ensemble, dt=1).check().prepare().run(until=2).trajectory_
>>> float(trajectory.member(1, compile_system(logistic_map())).values('map.x')[0])
0.2

Source code in src/skcomplex/engine/runtime/_trajectories.py

def member(self, index: int, problem: BaseProblemSpec) -> 'Trajectory':
    """Return the trajectory of one member of a replicated system, as a trajectory of the original system.

    Args:
        index:
            The member index, the same in every group of the replicated system.

        problem:
            The compiled original system, whose single elements the replicated groups hold.

    Returns:
        trajectory:
            The times, states, and parameters of the member, laid out as the original problem lays them out.

    Examples:
        >>> import numpy as np
        >>> from skcomplex.engine.compile import compile_system
        >>> from skcomplex.engine.runtime import Simulation
        >>> from skcomplex.experiments import BaseExperiment
        >>> from skcomplex.models.mathematics.dynamical_systems import logistic_map
        >>> ensemble = BaseExperiment._replicate(logistic_map(), 3, {'map.x': [0.1, 0.2, 0.3]})
        >>> trajectory = Simulation(ensemble, dt=1).check().prepare().run(until=2).trajectory_
        >>> float(trajectory.member(1, compile_system(logistic_map())).values('map.x')[0])
        0.2
    """
    frames = len(self.times)
    states = np.zeros((frames, problem.states.size))
    parameters = np.zeros((frames, problem.parameters.size))
    for layout, source, target, values in (
        (problem.states, self.problem.states, states, self.states),
        (problem.parameters, self.problem.parameters, parameters, self.parameters),
    ):
        for entry in layout.entries:
            origin = source.entry(entry.name)
            if origin.size == entry.size:
                target[:, entry.start : entry.start + entry.size] = np.asarray(values)[
                    :,
                    origin.start : origin.start + origin.size,
                ]
                continue
            columns = np.asarray(values)[:, origin.start : origin.start + origin.size].reshape(
                frames,
                -1,
                entry.size,
            )
            target[:, entry.start : entry.start + entry.size] = columns[:, index]
    return Trajectory(problem, self.times, states, parameters)

network(frame, nodes=None, layer=None)

Return the nodes and edges of a network whose topology changes, as they are at a frame.

Parameters:

Name	Type	Description	Default
frame	int	The index of the frame.	required
nodes	str | None	The path of a variable group on the nodes of the network, or None for the first such network.	None
layer	str | None	The layer, or None for the graph of the network.	None

Returns:

Name	Type	Description
snapshot	NetworkSnapshotSpec	The rows of the nodes present and the ends and rows of the edges present.

Raises:

Type	Description
KeyError	The system has no network whose topology changes.

Source code in src/skcomplex/engine/runtime/_trajectories.py

def network(self, frame: int, nodes: str | None = None, layer: str | None = None) -> NetworkSnapshotSpec:
    """Return the nodes and edges of a network whose topology changes, as they are at a frame.

    Args:
        frame:
            The index of the frame.

        nodes:
            The path of a variable group on the nodes of the network, or `None` for the first such network.

        layer:
            The layer, or `None` for the graph of the network.

    Returns:
        snapshot:
            The rows of the nodes present and the ends and rows of the edges present.

    Raises:
        KeyError:
            The system has no network whose topology changes.
    """
    topology = self.problem.topology
    if topology is None:
        error_msg = 'The system has no network whose topology changes. Give its groups `variable=True`.'
        raise KeyError(error_msg)
    return topology.read_network(self.parameters[frame], nodes, layer)

structure(path, frame)

Return the word or patch of a rewritten element at a frame.

Parameters:

Name	Type	Description	Default
path	str	The path of the rewritten element.	required
frame	int	The index of the frame.	required

Returns:

Name	Type	Description
structure	object	The structure recorded at that frame, which later generations do not change.

Raises:

Type	Description
KeyError	The path is not a rewritten element of the system.

Examples:

>>> from skcomplex.models import run
>>> trajectory = run('cantor_set', settings={'until': 4}).trajectory_
>>> trajectory.structure('cantor', 2).parts
4

Source code in src/skcomplex/engine/runtime/_trajectories.py

def structure(self, path: str, frame: int) -> object:
    """Return the word or patch of a rewritten element at a frame.

    Args:
        path:
            The path of the rewritten element.

        frame:
            The index of the frame.

    Returns:
        structure:
            The structure recorded at that frame, which later generations do not change.

    Raises:
        KeyError:
            The path is not a rewritten element of the system.

    Examples:
        >>> from skcomplex.models import run
        >>> trajectory = run('cantor_set', settings={'until': 4}).trajectory_
        >>> trajectory.structure('cantor', 2).parts
        4
    """
    structures = self.structures[frame] if self.structures else {}
    if path not in structures:
        names = ', '.join(structures) or 'none'
        error_msg = f'No rewritten element {path!r}. The rewritten elements of the system are: {names}.'
        raise KeyError(error_msg)
    return structures[path]

values(name)

Return the values of a state or a parameter over time by dotted name.

Parameters:

Name	Type	Description	Default
name	str	The dotted name, such as rabbits.n or growth.r.	required

Returns:

Name	Type	Description
values	AnyArray	One value per frame for a scalar, or one row per frame for a vector, NaN in the rows of a variable group on a network that hold no member at a frame.

Raises:

Type	Description
KeyError	The name is neither a state nor a parameter.

Source code in src/skcomplex/engine/runtime/_trajectories.py

def values(self, name: str) -> AnyArray:
    """Return the values of a state or a parameter over time by dotted name.

    Args:
        name:
            The dotted name, such as `rabbits.n` or `growth.r`.

    Returns:
        values:
            One value per frame for a scalar, or one row per frame for a vector, NaN in the rows of a variable
            group on a network that hold no member at a frame.

    Raises:
        KeyError:
            The name is neither a state nor a parameter.
    """
    if self.problem.states.has(name):
        entry = self.problem.states.entry(name)
        return self._present(name, entry.read(np.asarray(self.problem.states.read(self.states, name))))
    if self.problem.parameters.has(name):
        entry = self.problem.parameters.entry(name)
        values = entry.read(np.asarray(self.problem.parameters.read(self.parameters, name)))
        return self._present(name, values)
    if self.reporters and name in self.reporters[0]:
        found = [frame[name] for frame in self.reporters]
        if all(isinstance(item, numbers.Real) for item in found):
            return np.asarray(found, dtype=float)
        return np.fromiter(found, dtype=object, count=len(found))
    if self.observables and name in self.observables[0]:
        return np.asarray([frame[name] for frame in self.observables])
    reporter_names = list(self.reporters[0]) if self.reporters else []
    observable_names = list(self.observables[0]) if self.observables else []
    names = [*self.problem.states.names, *self.problem.parameters.names, *reporter_names, *observable_names]
    error_msg = f'No state or parameter named {name!r}. Available names are: {", ".join(names)}.'
    raise KeyError(error_msg)