Simulation

The continuous-time simulator: integrate dx = f(x, u, t) of a DynamicSystem or a diagram on a time grid.

class minilink.simulation.simulator.Simulator(sys, x0=None, t0=0, tf=10, n_steps=None, dt=None, solver=None, verbose=False, compile_backend='numpy', solver_warnings='warn')[source]

Integrate a compiled system along a discrete time grid.

The dynamics follow dx/dt = f(x, u, t). Inputs are nominal for solve(); solve_forced() accepts callables, constants, or sampled input trajectories and converts them to the solver’s sampled input matrix. The system has state dimension n and input dimension m (from sys.n and sys.m). compute_trajectory is the facade; use the class directly for repeated runs with the same solver, grid and backend.

Parameters:
  • sys (System) – Model to simulate; compiled with compile_backend as below.

  • x0 (array_like, optional) – Initial state in \(\mathbb{R}^n\). If None, uses sys.x0.

  • t0 (float) – Start and end time.

  • tf (float) – Start and end time.

  • n_steps (int, optional) – Number of time samples (including endpoints) when dt is not set.

  • dt (float, optional) – Step size when n_steps is not set. The grid ends at tf when dt divides tf - t0; otherwise its last sample is the first step past tf.

  • solver (str, optional) – Solver mode; if None, chosen by select_solver(). Use "euler" for variable dt per output knot, or "euler_fixedsteps" for uniform grids via compiled Euler rollouts (same pattern as "rk4_fixedsteps").

  • verbose (bool) – Print setup information (default quiet).

  • compile_backend (str) – Name passed as backend to compile(). Typical values are numpy (default) or jax. Use COMPILE_BACKEND_AUTO (the string auto) to try JAX if importable, then fall back to NumPy.

  • solver_warnings (str) – "warn" (default), "error", or "ignore" for discontinuous-loop UserWarning messages from minilink.simulation.solver_warnings.

solve()[source]

Run the simulation over self.t with the nominal input.

The input is the constant \(\bar{u}\) of the compiled evaluator.

Returns:

State and input time series, (n, n_pts) and (m, n_pts).

Return type:

Trajectory

solve_forced(u, input_port_id=None)[source]

Run simulation with a prescribed input.

u may be a full sampled input matrix, a callable u(t), a constant vector, or a scalar for one-dimensional inputs. If input_port_id is given, u describes only that port and the other ports use nominal values.

Returns:

State and input time series.

Return type:

Trajectory

select_time_vector(t0, tf, n_steps, dt, sys, solver_mode=None)[source]

Build time samples on [t0, tf] and return (time_vector, dt, len).

Explicit n_steps or dt win (n_steps if both are set, with a logged warning). With neither, the automatic grid depends on the solver: fixed-step solvers (FIXED_STEP_SOLVERS) integrate on the grid, so dt comes from the system’s smallest time constant scaled by AUTO_DT_SCALE, rounded down to a divisor of the sample period when a block holds a signal; adaptive solvers pick their own steps, so the grid is only a reporting resolution with DEFAULT_N_STEPS points.

select_solver(sys, user_solver=None)[source]

Choose the solver backend label from the system and options.

  • If the user has specified a solver, return it.

  • If the system has discontinuous behavior, return "euler".

  • If a block holds a signal over a sample period (solver_info ["sample_period"], a noise source), return "rk4_fixedsteps": the automatic grid then steps at a divisor of that period.

  • If the time grid was given explicitly, compile_backend is "jax", the grid is uniform, and it has at least RK4_AUTO_MIN_TIME_POINTS points, return "rk4_fixedsteps" (fast JIT rollout). The automatic grid never triggers this rule.

  • Otherwise, return "scipy".

Static simulation: a memoryless System leaf (n = 0) sampled on a time grid.

class minilink.simulation.static_simulator.StaticSimulator(sys, x0=None, t0=0, tf=10, n_steps=None, dt=None, solver=None, verbose=False, compile_backend='numpy')[source]

Evaluate static IO blocks on a time grid (no ODE integration).

Records boundary outputs in Trajectory.signals via the compiled static evaluator’s outputs method.

solve()[source]

Sample the outputs on the grid under the nominal input.

solve_forced(u, input_port_id=None)[source]

Sample the outputs on the grid under a prescribed input, as Simulator.solve_forced.

Input interpolation: a sampled input signal read at any time of a forced simulation.

minilink.simulation.input_interpolation.build_u_at_t(times, u, scheme='linear')[source]

Return u(t): the samples u, shape (m, n_pts), at times, read at any t.

"zoh" holds each sample until the next one; any other scheme is an interp1d kind, one interpolant per channel. Before the first sample and after the last, the end samples hold.

Parameters:
  • times (ndarray)

  • u (ndarray)

  • scheme (str)

Return type:

Callable[[float], ndarray]

ODE solver backends: the integrator contract a Simulator steps through.

class minilink.simulation.solvers.solver.SolverBackend[source]

Integrate ẋ = f(x, u, t) through an evaluator of the dynamics primitives.

The evaluator provides f(x, u, t) for a forced input and f_ivp(x, t) = f(x, u_nom, t) for the nominal one.

abstract integrate(evaluator, times, x0, args=None)[source]

Nominal input: x(t) = x₀ + ∫ f_ivp(x(s), s) ds with f_ivp(x, t) = f(x, u_nom, t).

Parameters:
  • evaluator (evaluator) – Provides f_ivp(x, t).

  • times (ndarray) – 1D time grid, shape (n_pts,).

  • x0 (ndarray) – Initial state, shape (n,).

  • args (optional) – Backend options.

Returns:

The states x, shape (n, n_pts).

Return type:

ndarray

abstract integrate_forced(evaluator, times, u, x0, args=None)[source]

Forced input: x(t) = x₀ + ∫ f(x(s), u(s), s) ds.

Parameters:
  • evaluator (evaluator) – Provides f(x, u, t).

  • times (ndarray) – 1D time grid, shape (n_pts,).

  • u (ndarray) – Sampled input, shape (m, n_pts), column k at times[k].

  • x0 (ndarray) – Initial state, shape (n,).

  • args (optional) – Backend options.

Returns:

The states x, shape (n, n_pts).

Return type:

ndarray

Adaptive SciPy scipy.integrate.solve_ivp() on the simulation time grid.

class minilink.simulation.solvers.scipy_ivp.SciPySolverBackend[source]

Adaptive solve_ivp, its solution reported on the times grid.

integrate(evaluator, times, x0, args=None)[source]

Nominal input: x(t) = x₀ + ∫ f_ivp(x(s), s) ds with f_ivp(x, t) = f(x, u_nom, t).

Parameters:
  • evaluator (evaluator) – Provides f_ivp(x, t).

  • times (ndarray) – 1D time grid, shape (n_pts,).

  • x0 (ndarray) – Initial state, shape (n,).

  • args (optional) – Backend options.

Returns:

The states x, shape (n, n_pts).

Return type:

ndarray

integrate_forced(evaluator, times, u, x0, args=None)[source]

Forced input: x(t) = x₀ + ∫ f(x(s), u(s), s) ds.

Parameters:
  • evaluator (evaluator) – Provides f(x, u, t).

  • times (ndarray) – 1D time grid, shape (n_pts,).

  • u (ndarray) – Sampled input, shape (m, n_pts), column k at times[k].

  • x0 (ndarray) – Initial state, shape (n,).

  • args (optional) – Backend options.

Returns:

The states x, shape (n, n_pts).

Return type:

ndarray

Variable-step explicit Euler on the simulation time grid.

class minilink.simulation.solvers.euler.EulerSolverBackend[source]

Explicit Euler with per-interval dt = times[i+1] - times[i].

integrate(evaluator, times, x0, args=None)[source]

Nominal input: x(t) = x₀ + ∫ f_ivp(x(s), s) ds with f_ivp(x, t) = f(x, u_nom, t).

Parameters:
  • evaluator (evaluator) – Provides f_ivp(x, t).

  • times (ndarray) – 1D time grid, shape (n_pts,).

  • x0 (ndarray) – Initial state, shape (n,).

  • args (optional) – Backend options.

Returns:

The states x, shape (n, n_pts).

Return type:

ndarray

integrate_forced(evaluator, times, u, x0, args=None)[source]

Forced input: x(t) = x₀ + ∫ f(x(s), u(s), s) ds.

Parameters:
  • evaluator (evaluator) – Provides f(x, u, t).

  • times (ndarray) – 1D time grid, shape (n_pts,).

  • u (ndarray) – Sampled input, shape (m, n_pts), column k at times[k].

  • x0 (ndarray) – Initial state, shape (n,).

  • args (optional) – Backend options.

Returns:

The states x, shape (n, n_pts).

Return type:

ndarray

Fixed-step explicit Euler on a uniform simulation time grid.

class minilink.simulation.solvers.euler_fixed.EulerFixedStepSolverBackend[source]

Fixed-step Euler via compiled rollout primitives (uniform times only).

integrate(evaluator, times, x0, args=None)[source]

Nominal input: x(t) = x₀ + ∫ f_ivp(x(s), s) ds with f_ivp(x, t) = f(x, u_nom, t).

Parameters:
  • evaluator (evaluator) – Provides f_ivp(x, t).

  • times (ndarray) – 1D time grid, shape (n_pts,).

  • x0 (ndarray) – Initial state, shape (n,).

  • args (optional) – Backend options.

Returns:

The states x, shape (n, n_pts).

Return type:

ndarray

integrate_forced(evaluator, times, u, x0, args=None)[source]

Forced input: x(t) = x₀ + ∫ f(x(s), u(s), s) ds.

Parameters:
  • evaluator (evaluator) – Provides f(x, u, t).

  • times (ndarray) – 1D time grid, shape (n_pts,).

  • u (ndarray) – Sampled input, shape (m, n_pts), column k at times[k].

  • x0 (ndarray) – Initial state, shape (n,).

  • args (optional) – Backend options.

Returns:

The states x, shape (n, n_pts).

Return type:

ndarray

Fixed-step fourth-order Runge–Kutta on the simulation time grid.

class minilink.simulation.solvers.rk4_fixed.RK4SolverBackend[source]

Fixed-step RK4 on the times grid.

integrate(evaluator, times, x0, args=None)[source]

Nominal input: x(t) = x₀ + ∫ f_ivp(x(s), s) ds with f_ivp(x, t) = f(x, u_nom, t).

Parameters:
  • evaluator (evaluator) – Provides f_ivp(x, t).

  • times (ndarray) – 1D time grid, shape (n_pts,).

  • x0 (ndarray) – Initial state, shape (n,).

  • args (optional) – Backend options.

Returns:

The states x, shape (n, n_pts).

Return type:

ndarray

integrate_forced(evaluator, times, u, x0, args=None)[source]

Forced input: x(t) = x₀ + ∫ f(x(s), u(s), s) ds.

Parameters:
  • evaluator (evaluator) – Provides f(x, u, t).

  • times (ndarray) – 1D time grid, shape (n_pts,).

  • u (ndarray) – Sampled input, shape (m, n_pts), column k at times[k].

  • x0 (ndarray) – Initial state, shape (n,).

  • args (optional) – Backend options.

Returns:

The states x, shape (n, n_pts).

Return type:

ndarray

Scheduled step simulation: a compiled step diagram ticked on a multi-rate schedule.

class minilink.simulation.computer.Computer(diagram, schedule)[source]

Stateful scheduled runtime for a StepDiagramSystem.

Compiles the same StepExecutionPlan as the synchronous compute_rollout, then advances with tick() under schedule divisors (multi-rate firing) and double-buffered internal port signals.

Within each tick, fired blocks gather inputs from the committed read buffer only (parallel / order-agnostic semantics). Skipped blocks hold via the write-buffer copy taken at tick start.

Parameters:
compile(backend='numpy', *, bind_params=False, verbose=False)[source]

Build the step execution plan and per-subsystem lookup tables.

Parameters:
  • backend (str)

  • bind_params (bool)

  • verbose (bool)

Return type:

None

reset(x0=None)[source]

Reset tick index, state, and internal signal buffers.

Return type:

None

tick(u)[source]

Advance one base tick.

Parameters:

u (array_like) – Diagram boundary input vector (layout diagram.m).

Returns:

Boundary output port values after the tick commit.

Return type:

dict[str, ndarray]

divisor(sys_id)[source]

Return the fire divisor for sys_id (default 1).

Parameters:

sys_id (str)

Return type:

int

class minilink.simulation.computer.StepSchedule(dt_base, fire=<factory>)[source]

Firing schedule for a Computer runtime.

Parameters:
  • dt_base (float)

  • fire (dict[str, int])

classmethod from_rates(dt_base, rates_hz)[source]

Build fire divisors from subsystem rates in Hz.

Parameters:
  • dt_base (float)

  • rates_hz (dict[str, float])

Return type:

StepSchedule

minilink.simulation.computer.as_computer(side, schedule)[source]

Build a Computer from a step block or StepDiagramSystem.

Wraps a leaf in a step diagram, exposes standard boundary ports, and attaches schedule. Use the % operator as shorthand: computer = block % dt. A block whose class defines its own % builds the Computer there (an MPC block checks schedule against its dt_mpc).

Parameters:
Return type:

Computer

Hybrid simulation: a computer ticking on its schedule, the plant held between ticks.

class minilink.simulation.hybrid_simulator.HybridSimulator(hybrid, *, x0_plant=None, x0_computer=None, t0=0, tf=10, n_steps=None, plant_dt_inner=None, compile_backend='numpy', verbose=False)[source]

Advance a HybridDiagram on the computer base tick grid.

The API mirrors Simulator: construct with t0 / tf, then solve() or solve_forced().

Parameters:
  • hybrid (HybridDiagram) – Computer + plant + boundary connections.

  • x0_plant (array_like, optional) – Initial plant / computer states (defaults from diagram x0).

  • x0_computer (array_like, optional) – Initial plant / computer states (defaults from diagram x0).

  • t0 (float) – Simulation interval in seconds. Duration defaults to tf - t0.

  • tf (float) – Simulation interval in seconds. Duration defaults to tf - t0.

  • n_steps (int, optional) – Override number of base ticks (advanced).

  • plant_dt_inner (float, optional) – RK4 sub-step inside each base tick for the plant; also sets plant Trajectory sample spacing.

  • compile_backend (str) – Backend for plant compile (default numpy).

  • verbose (bool) – Print setup information.

solve()[source]

Simulate with nominal computer-boundary inputs.

solve_forced(u, *, input_port_id=None)[source]

Simulate with prescribed computer-boundary input.

u may be a callable u(t), constant vector, or sampled matrix on the tick grid (m × n_ticks). With input_port_id, u drives that one input port and the other ports hold their nominal values.

run(u_traj)[source]

Tick the computer on u_traj (m × n_ticks), the plant held between ticks.

Parameters:

u_traj (ndarray)

Return type:

HybridSimResult

class minilink.simulation.hybrid_simulator.HybridSimResult(computer, plant, hybrid=None)[source]

Hybrid rollout with separate computer and plant views.

Parameters:
  • computer (StepRollout) – Tick-indexed computer diagram state and boundary signals.

  • plant (Trajectory) – Continuous-time plant state and boundary channels at plant_dt_inner (or dt_base when inner sub-stepping is unset).

  • hybrid (HybridDiagram, optional) – Source diagram for plot labels (optional).

property n_samples: int

Number of plant trajectory samples.

plot(*, signals=None, show=True, backend='matplotlib', **kwargs)[source]

Plot plant channels (continuous-time view).

Parameters:
  • signals (tuple[str, ...] | None)

  • show (bool)

plot_computer(*, signals=None, show=True, backend='matplotlib', **kwargs)[source]

Plot computer boundary channels on the tick index k.

Channels keep their computer port names (a u_cmd command is drawn as u_cmd); u here is the computer’s input vector, not the plant command.

Parameters:
  • signals (tuple[str, ...] | None)

  • show (bool)

Real-time simulation orchestrator.

RealtimeSimulator advances a continuous system in sync with the wall clock: poll a live RealtimeInput, hold u for one frame (ZOH), advance with the same compiled evaluator stack as offline simulation, draw the frame live through the existing Animator pipeline, publish to an optional RealtimeOutput, and return a Trajectory when the session stops.

Sibling orchestrators: Simulator (offline time grid) and HybridSimulator (computer ticks). The stepping math lives in the evaluators; this module owns only the loop, the clock policy, and the recording.

TODO: User Architectural Review — first realtime slice (game-mode parity); the loop structure and I/O contracts may still move.

class minilink.simulation.realtime.simulator.RealtimeSimulator(sys, *, frame_dt=0.03333333333333333, sim_dt=None, sync='locked', renderer='pygame', is_3d=False, input=None, output=None, compile_backend=None, max_steps=None, tf=None, verbose=False)[source]

Advance a continuous system live, one rendered frame at a time.

Each frame: u, stop = input.poll(t, x) (nominal u when no input), then x = evaluator.integrate_zoh(x, u, t, dt_hold, dt_inner=sim_dt), live draw, optional external publish, then sleep out the frame budget.

Parameters:
  • sys (DynamicSystem) – Continuous system or diagram to simulate.

  • frame_dt (float) – Target wall time between frames in seconds (default 1/30).

  • sim_dt (float, optional) – Internal integration step for the ZOH hold. None (default) picks a live-friendly step at the start of run(): time a short ZOH probe, then choose as many substeps per frame_dt as fit in AUTO_SIM_DT_BUDGET_FRAC of the frame budget (default 25%, so most of the frame is left for draw / I/O). The fine-end cap is the coarser of frame_dt / DEFAULT_SIM_STEPS_PER_FRAME and the offline Simulator auto-dt (smallest_time_constant × scale) — never finer than offline would use, and never more than the live substep ceiling. Pass an explicit value to skip calibration.

  • sync (str) – "locked" (default): advance exactly frame_dt of simulation time per frame — reproducible and JAX-friendly (constant ZOH shape). "realtime": advance by the measured wall elapsed, quantized to a multiple of sim_dt and clamped to MAX_DT_HOLD_FRAMES frame periods.

  • renderer (str or None) – Live view backend name for make_renderer() (default "pygame"); None runs headless.

  • is_3d (bool) – Open the live scene in 3-D (renderer dependent).

  • input (RealtimeInput, optional) – Live input source; None holds the nominal port values.

  • output (RealtimeOutput, optional) – External per-frame consumer; None publishes nothing.

  • compile_backend (str or None) – Evaluator backend. None or "auto" (default None): try JAX when available and the system is JAX-compatible, otherwise NumPy. Pass "numpy" or "jax" to force a backend.

  • max_steps (int, optional) – Stop after this many frames (headless smoke / CI).

  • tf (float, optional) – Stop once the simulation time reaches tf.

  • verbose (bool) – Print framed setup / completion panels (default quiet), matching the offline Simulator style, plus a live in-place per-frame line of compute time vs frame_dt budget (ok / LATE).

Notes

One sample is recorded per frame: sample k > 0 stores the state at t_k and the input held over the frame ending at t_k; sample 0 is the initial state with the nominal input.

When a frame’s wall cost exceeds frame_dt, a UserWarning is raised (throttled) so lagging sessions are visible without flooding the console.

run(x0=None, t0=0.0)[source]

Run the live loop until stop; return the recorded Trajectory.

The run stops on the first of: input should_stop, renderer quit event, max_steps frames, or simulation time reaching tf.

Parameters:
  • x0 (np.ndarray, optional) – Initial state (defaults to sys.x0).

  • t0 (float) – Initial simulation time.