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 forsolve();solve_forced()accepts callables, constants, or sampled input trajectories and converts them to the solver’s sampled input matrix. The system has state dimensionnand input dimensionm(fromsys.nandsys.m).compute_trajectoryis 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_backendas below.x0 (array_like, optional) – Initial state in \(\mathbb{R}^n\). If
None, usessys.x0.t0 (float) – Start and end time.
tf (float) – Start and end time.
n_steps (int, optional) – Number of time samples (including endpoints) when
dtis not set.dt (float, optional) – Step size when
n_stepsis not set. The grid ends attfwhendtdividestf - t0; otherwise its last sample is the first step pasttf.solver (str, optional) – Solver mode; if
None, chosen byselect_solver(). Use"euler"for variabledtper 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
backendtocompile(). Typical values arenumpy(default) orjax. UseCOMPILE_BACKEND_AUTO(the stringauto) to try JAX if importable, then fall back to NumPy.solver_warnings (str) –
"warn"(default),"error", or"ignore"for discontinuous-loopUserWarningmessages fromminilink.simulation.solver_warnings.
- solve()[source]¶
Run the simulation over
self.twith 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:
- solve_forced(u, input_port_id=None)[source]¶
Run simulation with a prescribed input.
umay be a full sampled input matrix, a callableu(t), a constant vector, or a scalar for one-dimensional inputs. Ifinput_port_idis given,udescribes only that port and the other ports use nominal values.- Returns:
State and input time series.
- Return type:
- 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_stepsordtwin (n_stepsif 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, sodtcomes from the system’s smallest time constant scaled byAUTO_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 withDEFAULT_N_STEPSpoints.
- 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_backendis"jax", the grid is uniform, and it has at leastRK4_AUTO_MIN_TIME_POINTSpoints, 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.signalsvia the compiled static evaluator’soutputsmethod.
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 samplesu, shape(m, n_pts), attimes, read at anyt."zoh"holds each sample until the next one; any otherschemeis aninterp1dkind, 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 andf_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) dswithf_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), columnkattimes[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 thetimesgrid.- integrate(evaluator, times, x0, args=None)[source]¶
Nominal input:
x(t) = x₀ + ∫ f_ivp(x(s), s) dswithf_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), columnkattimes[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) dswithf_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), columnkattimes[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
timesonly).- integrate(evaluator, times, x0, args=None)[source]¶
Nominal input:
x(t) = x₀ + ∫ f_ivp(x(s), s) dswithf_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), columnkattimes[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
timesgrid.- integrate(evaluator, times, x0, args=None)[source]¶
Nominal input:
x(t) = x₀ + ∫ f_ivp(x(s), s) dswithf_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), columnkattimes[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
StepExecutionPlanas the synchronouscompute_rollout, then advances withtick()underscheduledivisors (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:
diagram (StepDiagramSystem)
schedule (StepSchedule)
- 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
- class minilink.simulation.computer.StepSchedule(dt_base, fire=<factory>)[source]¶
Firing schedule for a
Computerruntime.- Parameters:
dt_base (float)
fire (dict[str, int])
- minilink.simulation.computer.as_computer(side, schedule)[source]¶
Build a
Computerfrom a step block orStepDiagramSystem.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 checksscheduleagainst itsdt_mpc).- Parameters:
side (System | StepDiagramSystem)
schedule (StepSchedule | float)
- Return type:
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
HybridDiagramon the computer base tick grid.The API mirrors
Simulator: construct witht0/tf, thensolve()orsolve_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
Trajectorysample spacing.compile_backend (str) – Backend for plant compile (default
numpy).verbose (bool) – Print setup information.
- solve_forced(u, *, input_port_id=None)[source]¶
Simulate with prescribed computer-boundary input.
umay be a callableu(t), constant vector, or sampled matrix on the tick grid (m × n_ticks). Withinput_port_id,udrives that one input port and the other ports hold their nominal values.
- 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(ordt_basewhen 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_cmdcommand is drawn asu_cmd);uhere 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)(nominaluwhen no input), thenx = 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 ofrun(): time a short ZOH probe, then choose as many substeps perframe_dtas fit inAUTO_SIM_DT_BUDGET_FRACof the frame budget (default 25%, so most of the frame is left for draw / I/O). The fine-end cap is the coarser offrame_dt / DEFAULT_SIM_STEPS_PER_FRAMEand the offlineSimulatorauto-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 exactlyframe_dtof simulation time per frame — reproducible and JAX-friendly (constant ZOH shape)."realtime": advance by the measured wall elapsed, quantized to a multiple ofsim_dtand clamped toMAX_DT_HOLD_FRAMESframe periods.renderer (str or None) – Live view backend name for
make_renderer()(default"pygame");Noneruns headless.is_3d (bool) – Open the live scene in 3-D (renderer dependent).
input (RealtimeInput, optional) – Live input source;
Noneholds the nominal port values.output (RealtimeOutput, optional) – External per-frame consumer;
Nonepublishes nothing.compile_backend (str or None) – Evaluator backend.
Noneor"auto"(defaultNone): 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
Simulatorstyle, plus a live in-place per-frame line of compute time vsframe_dtbudget (ok/LATE).
Notes
One sample is recorded per frame: sample
k > 0stores the state att_kand the input held over the frame ending att_k; sample 0 is the initial state with the nominal input.When a frame’s wall cost exceeds
frame_dt, aUserWarningis 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_stepsframes, or simulation time reachingtf.- Parameters:
x0 (np.ndarray, optional) – Initial state (defaults to
sys.x0).t0 (float) – Initial simulation time.