Skip to content

Controllers

shinro.controllers

Control algorithms for state regulation and trajectory tracking.

Provides discrete-time controllers that compute control actions from state feedback. All controllers implement the Controller ABC.

Available controllers: LQR — Linear Quadratic Regulator (DARE-based optimal gain) PIDController — Proportional-Integral-Derivative with anti-windup MPC_LTI — Linear Time-Invariant MPC with OSQP QP solver MPC_LTI_DeltaU — MPC with Δu (control rate) regularization MPPIController — Model Predictive Path Integral (sampling-based) SlidingModeController — Sliding Mode Control (robust nonlinear)

LQR

Bases: Controller

Linear Quadratic Regulator for discrete-time systems.

Computes the optimal state-feedback control law \(u = -K (x - x_t)\) that minimizes:

\[ J = \sum_k \left( x_k^T Q x_k + u_k^T R u_k \right) \]

The gain K is computed once via the Discrete Algebraic Riccati Equation (DARE) and applied online as a single matrix-vector multiply.

The DARE solve uses scipy (numpy-only) since there is no equivalent in PyTorch. The conversion is handled transparently via bk.to_numpy / bk.from_numpy.

Parameters:

Name Type Description Default
state_cost_matrix

Q — penalizes state deviation (n_x, n_x).

 required
control_cost_matrix

R — penalizes control effort (n_u, n_u).

 required
dynamics_state_matrix

A — discrete-time state transition (n_x, n_x).

 required
dynamics_control_matrix

B — control input matrix (n_x, n_u).

 required
backend ArrayBackend | None

Array backend. Defaults to NumpyBackend.

 None
Source code in src/shinro/controllers/lqr.py 
54
55
56
57
58
59
60
61
62
63
64
65
66
67
def __init__(
    self,
    state_cost_matrix,
    control_cost_matrix,
    dynamics_state_matrix,
    dynamics_control_matrix,
    backend: ArrayBackend | None = None,
):
    self.bk = backend or NumpyBackend()
    self.A = dynamics_state_matrix
    self.B = dynamics_control_matrix
    self.Q = state_cost_matrix
    self.R = control_cost_matrix
    self.gain_calculation()

gain_calculation 

gain_calculation()

Solve DARE and compute the optimal LQR gain matrix K.

Solves \(P = A^T P A - A^T P B (R + B^T P B)^{-1} B^T P A + Q\) via scipy.linalg.solve_discrete_are, then computes:

\[ K = (R + B^T P B)^{-1} B^T P A \]

The gain K is stored as self.K and used in compute().

Source code in src/shinro/controllers/lqr.py 
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
def gain_calculation(self):
    """Solve DARE and compute the optimal LQR gain matrix K.

    Solves :math:`P = A^T P A - A^T P B (R + B^T P B)^{-1} B^T P A + Q`
    via ``scipy.linalg.solve_discrete_are``, then computes:

    .. math::

        K = (R + B^T P B)^{-1} B^T P A

    The gain K is stored as ``self.K`` and used in ``compute()``.
    """
    A_np = self.bk.to_numpy(self.A)
    B_np = self.bk.to_numpy(self.B)
    P_np = solve_discrete_are(A_np, B_np, self.bk.to_numpy(self.Q), self.bk.to_numpy(self.R))
    P = self.bk.from_numpy(P_np)
    self.K = self.bk.inv(self.R + self.B.T @ P @ self.B) @ (self.B.T @ P @ self.A)

compute 

compute(current_state, target_state: Any | None = None)

Compute the optimal control input \(u = -K (x - x_t)\).

Parameters:

Name Type Description Default
current_state

Current state vector (n_x,).

 required
target_state Any | None

Desired state vector (n_x,). Defaults to zeros.

 None

Returns:

Type Description

Control input vector (n_u,).
Source code in src/shinro/controllers/lqr.py 
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
def compute(self, current_state, target_state: Any | None = None):
    """Compute the optimal control input :math:`u = -K (x - x_t)`.

    Args:
        current_state: Current state vector (n_x,).
        target_state: Desired state vector (n_x,). Defaults to zeros.

    Returns:
        Control input vector (n_u,).
    """
    if target_state is None:
        target_state = self.bk.zeros_like(current_state)
    error = target_state - current_state
    return self.K @ error

reset 

reset()

No internal state to reset for LQR.

Source code in src/shinro/controllers/lqr.py 
102
103
def reset(self):
    """No internal state to reset for LQR."""

from_config classmethod 

from_config(config, backend: ArrayBackend | None = None)

Create an LQR controller from a TOML config dict or LQRConfig.

Config fields

state_cost: Diagonal Q weights (n_x,) or full Q matrix (n_x, n_x). control_cost: Diagonal R weights (n_u,) or full R matrix (n_u, n_u). dt: Time step (used for validation against the plant). A_dynamics: Full A matrix (n_x, n_x). Required unless a scenario with a plant derives it. B_dynamics: Full B matrix (n_x, n_u). Required unless a scenario with a plant derives it.

Parameters:

Name Type Description Default
config

TOML config dict or LQRConfig.

 required
backend ArrayBackend | None

Array backend. Defaults to NumpyBackend.

 None

Returns:

Type Description

LQR instance.
Source code in src/shinro/controllers/lqr.py 
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
@classmethod
def from_config(cls, config, backend: ArrayBackend | None = None):
    """Create an LQR controller from a TOML config dict or :class:`LQRConfig`.

    Config fields:
        state_cost: Diagonal Q weights (n_x,) or full Q matrix (n_x, n_x).
        control_cost: Diagonal R weights (n_u,) or full R matrix (n_u, n_u).
        dt: Time step (used for validation against the plant).
        A_dynamics: Full A matrix (n_x, n_x). Required unless a scenario with a
            plant derives it.
        B_dynamics: Full B matrix (n_x, n_u). Required unless a scenario with a
            plant derives it.

    Args:
        config: TOML config dict or LQRConfig.
        backend: Array backend. Defaults to NumpyBackend.

    Returns:
        LQR instance.
    """
    bk = backend or NumpyBackend()
    cfg = cls.parse_config(config)
    if cfg.A_dynamics is None or cfg.B_dynamics is None:
        raise ValueError(
            "LQR: A_dynamics and B_dynamics are required — declare them in the config, or "
            "build the controller through a scenario with a plant (which derives them)."
        )
    return cls(
        state_cost_matrix=parse_matrix(bk, cfg.state_cost),
        control_cost_matrix=parse_matrix(bk, cfg.control_cost),
        dynamics_state_matrix=bk.array(cfg.A_dynamics),
        dynamics_control_matrix=bk.array(cfg.B_dynamics),
        backend=bk,
    )

PIDController

Bases: Controller

Proportional-Integral-Derivative controller with anti-windup.

Computes control effort as:

\[ u(t) = K_p e(t) + K_i \int e(\tau) d\tau + K_d \frac{de}{dt} \]

Features: - Independent gains per channel (Kp, Ki, Kd as vectors). - Output clamping with integral anti-windup back-calculation. - Derivative on error (standard form).

When output is clamped, the integral term is back-calculated on saturated channels only to prevent integral windup.

Parameters:

Name Type Description Default
kp

Proportional gain vector (n,).

 required
ki

Integral gain vector (n,).

 required
kd

Derivative gain vector (n,).

 required
dt float

Time step in seconds.

 required
output_limits tuple | None

Optional (min_limits, max_limits) for output clamping. Each is an array of shape (n,).

 None
backend ArrayBackend | None

Array backend. Defaults to NumpyBackend.

 None
Source code in src/shinro/controllers/pid.py 
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
def __init__(
    self,
    kp,
    ki,
    kd,
    dt: float,
    output_limits: tuple | None = None,
    backend: ArrayBackend | None = None,
):
    self.bk = backend or NumpyBackend()
    self.kp = kp
    self.kd = kd
    self.ki = ki
    self.dt = dt

    self.min_limits = output_limits[0] if output_limits else None
    self.max_limits = output_limits[1] if output_limits else None
    self._integral = self.bk.zeros_like(self.ki)
    self._prev_error = self.bk.zeros_like(self.kd)
    # First-tick gate as a 0/1 array (not a Python bool): under tracing a
    # bool branch would bake the first-tick path into the graph forever.
    # As a recurrent state port the graph gates the D-term at runtime.
    self._has_run = self.bk.zeros_like(self.ki)

compute 

compute(current_state, target_state)

Compute the PID control effort.

Parameters:

Name Type Description Default
current_state

Measured current state (n,).

 required
target_state

Desired target state (n,).

 required

Returns:

Type Description

Control effort vector (n,).
Source code in src/shinro/controllers/pid.py 
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
def compute(self, current_state, target_state):
    """Compute the PID control effort.

    Args:
        current_state: Measured current state (n,).
        target_state: Desired target state (n,).

    Returns:
        Control effort vector (n,).
    """
    error = target_state - current_state
    p_term = self.kp * error
    self._integral = self._integral + error * self.dt
    i_term = self.ki * self._integral

    # Where-gate: der = has_run ? (e - e_prev)/dt : 0. A Python branch
    # would bake the first-tick path into the traced graph forever; the
    # 0/1 flag is a recurrent state port the host feeds back (0 on tick
    # 0, 1 afterwards). where (not multiply): on tick 0 the candidate
    # divides by dt — with dt=0 that's inf, and inf*0 = NaN under a
    # multiply gate, while where simply discards the unselected branch.
    # The condition is an explicit `!= 0` so torch gets a boolean tensor
    # (torch.where rejects float conditions); under tracing it emits the
    # same ne node.
    der = self.bk.where(
        self._has_run != 0,
        (error - self._prev_error) / self.dt,
        self.bk.zeros_like(error),
    )
    d_term = self.kd * der

    control_effort = p_term + i_term + d_term

    if self.min_limits is not None and self.max_limits is not None:
        clamped_effort = self.bk.clip(control_effort, self.min_limits, self.max_limits)
        # Branch-free anti-windup: an elementwise mask replaces the old
        # `if bk.any(saturated)` early-out (a Python branch on a traced
        # value, which would silently apply the back-calculation every
        # tick). Where a channel is unsaturated the mask is 0 and the
        # integral passes through unchanged — identical results.
        saturated = control_effort != clamped_effort
        self._integral = self.bk.where(
            saturated,
            self._integral - error * self.dt,
            self._integral,
        )
        control_effort = clamped_effort

    self._prev_error = self.bk.copy(error)
    self._has_run = self.bk.zeros_like(self.ki) + 1.0
    return control_effort

reset 

reset()

Reset the controller's internal state (integral and previous error).

Source code in src/shinro/controllers/pid.py 
139
140
141
142
143
def reset(self):
    """Reset the controller's internal state (integral and previous error)."""
    self._integral = self.bk.zeros_like(self.ki)
    self._prev_error = self.bk.zeros_like(self.kd)
    self._has_run = self.bk.zeros_like(self.ki)

from_config classmethod 

from_config(config, backend: ArrayBackend | None = None)

Create a PID controller from a TOML config dict or PIDConfig.

Config fields

kp: List of proportional gains (n,). Defaults to [1.0]. ki: List of integral gains (n,). Defaults to zeros. kd: List of derivative gains (n,). Defaults to zeros. dt: Time step. Required at runtime; injected from the plant in scenario builds. output_limits: Optional dict with min and max lists.

Parameters:

Name Type Description Default
config

TOML config dict or PIDConfig.

 required
backend ArrayBackend | None

Array backend. Defaults to NumpyBackend.

 None

Returns:

Type Description

PIDController instance.
Source code in src/shinro/controllers/pid.py 
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
@classmethod
def from_config(cls, config, backend: ArrayBackend | None = None):
    """Create a PID controller from a TOML config dict or :class:`PIDConfig`.

    Config fields:
        kp: List of proportional gains (n,). Defaults to [1.0].
        ki: List of integral gains (n,). Defaults to zeros.
        kd: List of derivative gains (n,). Defaults to zeros.
        dt: Time step. Required at runtime; injected from the plant in
            scenario builds.
        output_limits: Optional dict with ``min`` and ``max`` lists.

    Args:
        config: TOML config dict or PIDConfig.
        backend: Array backend. Defaults to NumpyBackend.

    Returns:
        PIDController instance.
    """
    bk = backend or NumpyBackend()
    cfg = cls.parse_config(config)
    if cfg.dt is None:
        raise ValueError(
            "PID: dt is required (runtime integration) — omit it only in scenario "
            "builds, where the plant's dt is injected"
        )
    n = len(cfg.kp) if cfg.kp is not None else 1
    limits = None
    if cfg.output_limits is not None:
        lim = strict_from_dict(LimitsConfig, cfg.output_limits, "PID.output_limits")
        limits = (bk.array(lim.min), bk.array(lim.max))
    return cls(
        kp=bk.array(cfg.kp if cfg.kp is not None else [1.0] * n),
        ki=bk.array(cfg.ki if cfg.ki is not None else [0.0] * n),
        kd=bk.array(cfg.kd if cfg.kd is not None else [0.0] * n),
        dt=cfg.dt,
        output_limits=limits,
        backend=bk,
    )

MPC_LTI_DeltaU

Bases: MPC_LTI

MPC with \(\Delta u\) (control rate) regularization.

Augments the state to \([x; u_{\text{prev}}]\) so the control variable becomes \(\Delta u = u_k - u_{k-1}\). The cost penalizes \(\Delta u^T S \Delta u\), smoothing chatter.

The augmented dynamics are:

\[ z_{k+1} = \begin{bmatrix} A & B \\ 0 & I \end{bmatrix} z_k + \begin{bmatrix} B \\ I \end{bmatrix} \Delta u_k \]

where \(z = [x; u_{\text{prev}}]\).

Parameters:

Name Type Description Default
delta_u_penalty

S — cost matrix for \(\Delta u\) (n_u, n_u).

 required
**kwargs

Passed to MPC_LTI.init.

 {}
Source code in src/shinro/controllers/mpc_lti.py 
228
229
230
231
def __init__(self, delta_u_penalty, backend: ArrayBackend | None = None, **kwargs):
    self.bk = backend or NumpyBackend()
    self.S_delta = delta_u_penalty
    super().__init__(backend=self.bk, **kwargs)

from_config classmethod 

from_config(config, backend: ArrayBackend | None = None)

Create an MPC_DeltaU controller from a TOML config dict or MPCDeltaUConfig.

Config fields

delta_u_penalty: Diagonal S weights (n_u,) or full S matrix (n_u, n_u). horizon: Prediction horizon. state_cost: Diagonal Q weights (n_x,) or full Q matrix (n_x, n_x). control_cost: Diagonal R weights (n_u,) or full R matrix (n_u, n_u). dt: Time step (used for validation against the plant). A_dynamics: Full A matrix (n_x, n_x). Required unless a scenario with a plant derives it. B_dynamics: Full B matrix (n_x, n_u). Required unless a scenario with a plant derives it. constraints: Optional dict with upper and lower bound lists.

Parameters:

Name Type Description Default
config

TOML config dict or MPCDeltaUConfig.

 required
backend ArrayBackend | None

Array backend. Defaults to NumpyBackend.

 None

Returns:

Type Description

MPC_LTI_DeltaU instance.
Source code in src/shinro/controllers/mpc_lti.py 
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
@classmethod
def from_config(cls, config, backend: ArrayBackend | None = None):
    """Create an MPC_DeltaU controller from a TOML config dict or :class:`MPCDeltaUConfig`.

    Config fields:
        delta_u_penalty: Diagonal S weights (n_u,) or full S matrix (n_u, n_u).
        horizon: Prediction horizon.
        state_cost: Diagonal Q weights (n_x,) or full Q matrix (n_x, n_x).
        control_cost: Diagonal R weights (n_u,) or full R matrix (n_u, n_u).
        dt: Time step (used for validation against the plant).
        A_dynamics: Full A matrix (n_x, n_x). Required unless a scenario with a
            plant derives it.
        B_dynamics: Full B matrix (n_x, n_u). Required unless a scenario with a
            plant derives it.
        constraints: Optional dict with ``upper`` and ``lower`` bound lists.

    Args:
        config: TOML config dict or MPCDeltaUConfig.
        backend: Array backend. Defaults to NumpyBackend.

    Returns:
        MPC_LTI_DeltaU instance.
    """
    bk = backend or NumpyBackend()
    cfg = cls.parse_config(config)
    if cfg.A_dynamics is None or cfg.B_dynamics is None:
        raise ValueError(
            "MPC_DeltaU: A_dynamics and B_dynamics are required — declare them in the config, "
            "or build the controller through a scenario with a plant (which derives them)."
        )
    A = bk.array(cfg.A_dynamics)
    B = bk.array(cfg.B_dynamics)
    Q = parse_matrix(bk, cfg.state_cost)
    m = B.shape[1]
    ctrl = cls(
        delta_u_penalty=parse_matrix(bk, cfg.delta_u_penalty),
        horizon=cfg.horizon,
        control_cost_matrix=parse_matrix(bk, cfg.control_cost),
        state_cost_matrix=Q,
        A_dynamics=A,
        B_dynamics=B,
        terminal_cost=parse_matrix(bk, cfg.terminal_cost if cfg.terminal_cost is not None else cfg.state_cost),
        backend=bk,
    )
    if cfg.constraints is not None:
        cons = strict_from_dict(ConstraintsConfig, cfg.constraints, "MPC_DeltaU.constraints")
        # The QP variable is the control (n_u wide), so default F = [I; -I] is n_u x n_u.
        F = bk.array(cons.matrix) if cons.matrix is not None else bk.vstack([bk.eye(m), -bk.eye(m)])
        ctrl.constraints(F, cons.upper, cons.lower)
    return ctrl

compute 

compute(current_state, target_state: Any | None = None, u_prev: Any | None = None)

Solve MPC with \(\Delta u\) regularization.

Regulates the tracking error e = current_state - target_state to zero, augmenting the state with the previous control input. The QP variable is the control rate \(\Delta u\); this method returns the applied control \(u_k = u_{\text{prev}} + \Delta u_k\), which is what the host feeds back as u_prev next tick.

Parameters:

Name Type Description Default
current_state

Original (non-augmented) state vector (n_x,).

 required
target_state Any | None

Optional reference state (n_x,) to track.

 None
u_prev Any | None

Previous control input (n_u,). Defaults to zeros.

 None

Returns:

Type Description

Applied control \(u_k = u_{\text{prev}} + \Delta u_k\) (n_u,).
Source code in src/shinro/controllers/mpc_lti.py 
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
def compute(self, current_state, target_state: Any | None = None, u_prev: Any | None = None):
    """Solve MPC with :math:`\\Delta u` regularization.

    Regulates the tracking error ``e = current_state - target_state`` to
    zero, augmenting the state with the previous control input. The QP
    variable is the control rate :math:`\\Delta u`; this method returns the
    **applied** control :math:`u_k = u_{\\text{prev}} + \\Delta u_k`, which is
    what the host feeds back as ``u_prev`` next tick.

    Args:
        current_state: Original (non-augmented) state vector (n_x,).
        target_state: Optional reference state (n_x,) to track.
        u_prev: Previous control input (n_u,). Defaults to zeros.

    Returns:
        Applied control :math:`u_k = u_{\\text{prev}} + \\Delta u_k` (n_u,).
    """
    u_prev_val: Any = self.bk.zeros(self.m) if u_prev is None else u_prev
    x0 = current_state - target_state if target_state is not None else current_state
    x0_aug = self.bk.hstack([x0, u_prev_val])
    delta_u = super().compute(x0_aug)
    # The QP variable is the increment, but the framework treats a
    # controller's output as the applied control and feeds it back as
    # u_prev (runner + compose). Return the accumulated value so the
    # plant receives u = u_prev + du, not the bare increment.
    return u_prev_val + delta_u

MPPIController

Bases: Controller

Model Predictive Path Integral controller.

Each compute() call samples \(N\) Gaussian control perturbations, rolls out the dynamics over a horizon of \(K\) steps, weights the perturbations by the softmax of their total cost, and advances the receding-horizon nominal control sequence by one step. The returned action is the first element of the updated nominal sequence, clipped to the configured control bounds.

The dynamics and cost callables receive batched arrays of shape (N, D_x) / (N, D_u) and may be injected directly or produced by attach_plant from a plant's model.

Parameters:

Name Type Description Default
dynamics_fn Any | None

Callable dynamics_fn(x, u, dt) -> x_next stepping a batch of states forward one time step. Receives backend arrays of shape (N, D_x) and (N, D_u) and returns (N, D_x). May be None and injected later or via attach_plant.

 None
cost_fn Any | None

Callable cost_fn(x, u) -> c returning the per-sample stage cost as an array of shape (N,). May be None and injected later or via attach_plant.

 None
num_samples int

Number of sampled perturbations N.

 100
temperature float

Softmax temperature \(\lambda\) (> 0).

 1.0
dt float

Time step passed to the dynamics.

 0.01
horizon int

Prediction horizon K.

 10
noise_sigma

Standard deviation of the control perturbation per input channel (D_u,). If None, defaults to [0.5].

 None
u_min

Optional lower control bound (D_u,) or scalar.

 None
u_max

Optional upper control bound (D_u,) or scalar.

 None
seed int | None

Optional RNG seed for reproducible sampling.

 None
backend ArrayBackend | None

Array backend. Defaults to NumpyBackend. The rollout loop runs on this backend, so a TorchBackend executes batched tensor ops over the sample dimension.

 None
Source code in src/shinro/controllers/mppi.py 
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
def __init__(
    self,
    dynamics_fn: Any | None = None,
    cost_fn: Any | None = None,
    num_samples: int = 100,
    temperature: float = 1.0,
    dt: float = 0.01,
    horizon: int = 10,
    noise_sigma=None,
    u_min=None,
    u_max=None,
    seed: int | None = None,
    backend: ArrayBackend | None = None,
):
    self.bk = backend or NumpyBackend()
    self.dynamics_fn = dynamics_fn
    self.cost_fn = cost_fn

    if num_samples <= 0:
        raise ValueError(f"num_samples must be positive, got {num_samples}")
    if temperature <= 0:
        raise ValueError(f"temperature must be positive, got {temperature}")
    if horizon <= 0:
        raise ValueError(f"horizon must be positive, got {horizon}")
    if dt <= 0:
        raise ValueError(f"dt must be positive, got {dt}")

    self.N = num_samples
    self.K = horizon
    self.dt = _as_float(dt, "dt")
    self.lam = _as_float(temperature, "temperature")
    self.seed = seed
    self._rng = np.random.default_rng(seed)

    if noise_sigma is None:
        noise_sigma = [0.5]
    self.noise_sigma = np.atleast_1d(np.asarray(noise_sigma, dtype=np.float64))
    self.D_u = self.noise_sigma.shape[0]

    self.u_min = np.atleast_1d(np.asarray(u_min, dtype=np.float64)) if u_min is not None else None
    self.u_max = np.atleast_1d(np.asarray(u_max, dtype=np.float64)) if u_max is not None else None

    self.u = np.zeros((self.K, self.D_u))

    self._last_epsilon = None
    self._last_costs = None
    self._adapter = None
    self._Q = None
    self._R = None
    # Tracking reference for the attach_plant cost. Held in a 1-element
    # list rather than as an array-like attr: the tracer's attr-diff
    # promotes every reassigned ndarray/Tracer attr to a recurrent state
    # port, and this is a per-call input, not state.
    self._x_ref_holder = [None]

host_input_shapes 

host_input_shapes() -> dict[str, tuple[int, ...]]

Free C-ABI input ports the host fills each tick.

Sampling is host-side (see the module docstring), so in a lowered graph epsilon — the (N, K*D_u) sampled perturbations, sample-major — is a free input port rather than something the kernel generates. The compile pipeline (shinro.codegen.build.build_composed_graph) reads this to declare the port and pass it through compose(host_inputs=...); the eager path ignores it, because compute samples internally.

Returns:

Type Description
dict[str, tuple[int, ...]]

Maps the port name to its shape.
Source code in src/shinro/controllers/mppi.py 
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
def host_input_shapes(self) -> dict[str, tuple[int, ...]]:
    """Free C-ABI input ports the host fills each tick.

    Sampling is host-side (see the module docstring), so in a lowered graph
    ``epsilon`` — the ``(N, K*D_u)`` sampled perturbations, sample-major —
    is a free input port rather than something the kernel generates. The
    compile pipeline (:func:`shinro.codegen.build.build_composed_graph`)
    reads this to declare the port and pass it through
    ``compose(host_inputs=...)``; the eager path ignores it, because
    :meth:`compute` samples internally.

    Returns:
        Maps the port name to its shape.
    """
    return {"epsilon": (self.N, self.K * self.D_u)}

attach_plant 

attach_plant(plant, Q: Any | None = None, R: Any | None = None)

Wire a plant into the controller via a batched dynamics adapter.

Builds a BatchedDynamicsAdapter from the plant and sets dynamics_fn / cost_fn from it. The cost uses the quadratic stage cost \(x^T Q x + u^T R u\), optionally tracking the reference passed to compute — that reference is threaded through self._x_ref_holder, a plain list, so the tracer does not mistake it for recurrent state. Q and R may be diagonal (D,) or full (D, D) matrices; they default to identity.

Parameters:

Name Type Description Default
plant

A Plant exposing get_model() (and optionally dynamics() for nonlinear plants).

 required
Q Any | None

State cost matrix (D_x, D_x) or diagonal (D_x,).

 None
R Any | None

Control cost matrix (D_u, D_u) or diagonal (D_u,).

 None

Raises:

Type Description
ValueError

If the plant's control dimension disagrees with the controller's noise_sigma length.
Source code in src/shinro/controllers/mppi.py 
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
def attach_plant(self, plant, Q: Any | None = None, R: Any | None = None):
    """Wire a plant into the controller via a batched dynamics adapter.

    Builds a :class:`BatchedDynamicsAdapter` from the plant and sets
    ``dynamics_fn`` / ``cost_fn`` from it. The cost uses the quadratic
    stage cost :math:`x^T Q x + u^T R u`, optionally tracking the reference
    passed to :func:`compute` — that reference is threaded through
    ``self._x_ref_holder``, a plain list, so the tracer does not mistake it
    for recurrent state. ``Q`` and ``R`` may be diagonal ``(D,)`` or full
    ``(D, D)`` matrices; they default to identity.

    Args:
        plant: A :class:`Plant` exposing ``get_model()`` (and optionally
            ``dynamics()`` for nonlinear plants).
        Q: State cost matrix (D_x, D_x) or diagonal (D_x,).
        R: Control cost matrix (D_u, D_u) or diagonal (D_u,).

    Raises:
        ValueError: If the plant's control dimension disagrees with the
            controller's ``noise_sigma`` length.
    """
    from shinro.utils.batched_adapter import BatchedDynamicsAdapter

    adapter = BatchedDynamicsAdapter(plant)
    if adapter.control_dim != self.D_u:
        raise ValueError(
            f"Plant control dimension {adapter.control_dim} does not match "
            f"controller noise_sigma dimension {self.D_u}."
        )
    self._adapter = adapter
    self._Q = self._Q if Q is None else Q
    self._R = self._R if R is None else R
    self.D_x = adapter.state_dim
    self.D_u = adapter.control_dim

    # Route the controller's *current* backend into the batched path: a
    # trace swaps self.bk to a TraceBackend, while the adapter holds the
    # plant's (concrete) backend. Resolving self.bk at call time is what
    # lets a plant's batched_dynamics emit sin/mul/slice nodes instead of
    # evaluating eagerly against a concrete array.
    self.dynamics_fn = lambda x, u, dt: adapter.dynamics_fn(x, u, dt, bk=self.bk)
    self.cost_fn = lambda x, u: adapter.cost_fn(x, u, self._Q, self._R, x_ref=self._x_ref_holder[0])

compute 

compute(current_state, target_state: Any | None = None, epsilon: Any | None = None)

Compute the MPPI control action for a given initial state.

Draws \(N\) Gaussian perturbation sequences (host-side — see the module docstring), rolls out the dynamics over the horizon, computes the softmax-weighted update, and returns the first action of the updated nominal sequence (clipped to bounds if configured).

Everything after sampling runs through self.bk on batched tensors, so it evaluates identically on numpy, torch, and the tracing backend. Every reduction is written as a matmul identity — a sum over an axis is a contraction with a ones vector or with the softmax weights — except the softmax shift beta = min(costs), which is the min graph op.

Parameters:

Name Type Description Default
current_state

Initial state vector (D_x,). Accepts the backend's native array type (numpy or torch).

 required
target_state Any | None

Optional reference state (D_x,) to track. When given, the cost penalizes deviation (x - target_state); otherwise the controller regulates to the origin.

 None
epsilon Any | None

Optional pre-drawn perturbations, shape (N, K*D_u), sample-major (row i, column k*D_u + d). This is the trace hook: when lowering, the perturbations become a free graph input port the host fills each tick, so the traced call never samples. None (the eager path) draws them from self._rng.

 None

Returns:

Type Description

First control action (D_u,) in the backend's native type.

Raises:

Type Description
RuntimeError

If dynamics_fn or cost_fn has not been set.
Source code in src/shinro/controllers/mppi.py 
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
def compute(self, current_state, target_state: Any | None = None, epsilon: Any | None = None):
    """Compute the MPPI control action for a given initial state.

    Draws :math:`N` Gaussian perturbation sequences (host-side — see the
    module docstring), rolls out the dynamics over the horizon, computes
    the softmax-weighted update, and returns the first action of the
    updated nominal sequence (clipped to bounds if configured).

    Everything after sampling runs through ``self.bk`` on batched tensors,
    so it evaluates identically on numpy, torch, and the tracing backend.
    Every reduction is written as a matmul identity — a sum over an axis is
    a contraction with a ones vector or with the softmax weights — except
    the softmax shift ``beta = min(costs)``, which is the ``min`` graph op.

    Args:
        current_state: Initial state vector (D_x,). Accepts the backend's
            native array type (numpy or torch).
        target_state: Optional reference state (D_x,) to track. When given,
            the cost penalizes deviation ``(x - target_state)``; otherwise
            the controller regulates to the origin.
        epsilon: Optional pre-drawn perturbations, shape ``(N, K*D_u)``,
            sample-major (row ``i``, column ``k*D_u + d``). This is the
            trace hook: when lowering, the perturbations become a free
            graph input port the host fills each tick, so the traced call
            never samples. ``None`` (the eager path) draws them from
            ``self._rng``.

    Returns:
        First control action (D_u,) in the backend's native type.

    Raises:
        RuntimeError: If ``dynamics_fn`` or ``cost_fn`` has not been set.
    """
    if self.dynamics_fn is None or self.cost_fn is None:
        raise RuntimeError(
            "dynamics_fn and cost_fn must be set before calling compute(). "
            "Inject them directly, call attach_plant(plant, ...), or if the "
            "controller was built with from_config, set ctrl.dynamics_fn = ... "
            "and ctrl.cost_fn = ..."
        )
    dynamics_fn = self.dynamics_fn
    cost_fn = self.cost_fn

    # Sampling is host-side by design: the one step of the tick with no
    # dataflow representation. The traced path receives epsilon as a graph
    # input port, so this branch is a trace-time constant and the traced
    # call never touches the RNG.
    if epsilon is None:
        eps_sample = self._rng.normal(loc=0.0, scale=self.noise_sigma, size=(self.N, self.K, self.D_u))
        self._last_epsilon = eps_sample
        eps_in = self.bk.from_numpy(eps_sample.reshape(self.N, self.K * self.D_u))
    else:
        # Host-supplied perturbations (the trace path, or a caller wanting
        # reproducible noise). Bridged like the sampled path so a numpy
        # array works on a torch backend; a no-op under tracing, where
        # epsilon is already the graph input Tracer.
        eps_in = self.bk.from_numpy(epsilon)

    N, K, Du = self.N, self.K, self.D_u
    lam = self.lam

    # Batch the initial state to (N, D_x) with a same-rank broadcast; the
    # tracer rejects numpy's rank-differing tile.
    d_x = current_state.shape[0]
    x_current = self.bk.zeros((N, d_x)) + self.bk.reshape(current_state, (1, d_x))

    # Tracking reference for the attach_plant cost closure (see __init__),
    # stored as a (1, D_x) row: the closure subtracts it from the batched
    # states, and (N, D_x) - (1, D_x) stays same-rank for the tracer (numpy
    # would broadcast a (D_x,) there; the tracer rejects the rank gap).
    self._x_ref_holder[0] = self.bk.reshape(target_state, (1, d_x)) if target_state is not None else None

    # eps_in is (N, K*D_u) sample-major, so its transpose is (K*D_u, N) and
    # step k's perturbations are the row block [k*D_u:(k+1)*D_u] transposed
    # back to (N, D_u).
    eps_t = eps_in.T
    ones_du = self.bk.array(np.ones(Du))
    ones_n_col = self.bk.array(np.ones((N, 1)))
    sigma2 = self.bk.reshape(self.bk.array(self.noise_sigma**2), (1, Du))
    costs = self.bk.zeros(N)
    # Bridge the nominal sequence to the backend once. It is numpy-hosted
    # (see the assignment at the end), and `numpy + torch` raises — only
    # `torch + numpy` works — so every use goes through this backend-native
    # copy. Under tracing this is a no-op passthrough and u_nominal is the
    # recurrent state input Tracer.
    u_nominal = self.bk.from_numpy(self.u)

    for k in range(K):
        eps_k = self.bk.slice_(eps_t, k * Du, (k + 1) * Du).T
        u_k = self.bk.slice_(u_nominal, k, k + 1)  # nominal row -> (1, D_u)
        v_k = eps_k + u_k
        if self.u_min is not None or self.u_max is not None:
            v_k = self.bk.clip(v_k, self.u_min, self.u_max)
        costs = costs + cost_fn(x_current, v_k)
        # lam * sum_d (u_k,d / sigma_d^2) eps_i,k,d — a row sum, i.e. a
        # contraction with a ones vector.
        control_penalty = lam * ((u_k / sigma2 * eps_k) @ ones_du)
        costs = costs + control_penalty
        x_current = dynamics_fn(x_current, v_k, self.dt)

    costs = costs + cost_fn(x_current, self.bk.zeros((N, Du)))

    # Diagnostics: the traced backend publishes `costs` as an auxiliary
    # graph output port (SMC's `healthy` precedent); eager backends keep the
    # concrete array. Assigning a Tracer here would be mis-detected as
    # recurrent state.
    self.bk.emit_named_output("costs", costs)
    costs_np = self.bk.to_numpy(costs)
    if isinstance(costs_np, np.ndarray):
        self._last_costs = costs_np.copy()

    # Softmax weights. beta is the shift that keeps exp from overflowing.
    # The normalizer is a contraction with a ones vector, not a 1-D @ 1-D
    # product: (1,N) @ (N,1) with the weights on *both* sides would sum the
    # squares. The (1,1) result ravels to (1,), which broadcasts against (N,).
    beta = self.bk.min(costs)
    w = self.bk.exp(-(costs - beta) / lam)
    w_sum = self.bk.ravel(self.bk.reshape(w, (1, N)) @ ones_n_col)
    w = w / w_sum

    # Weighted average of the perturbations: sum_i w_i eps_i is the matmul
    # (1,N) @ (N, K*D_u). to_numpy keeps the nominal sequence numpy-hosted
    # on the eager backends; under tracing it is a no-op passthrough, so
    # self.u stays a Tracer and is emitted as the recurrent state output.
    w_row = self.bk.reshape(w, (1, N))
    weighted_eps = self.bk.reshape(w_row @ eps_in, (K, Du))
    u_updated = u_nominal + weighted_eps

    # The returned action is the first element *before* the receding-horizon
    # shift (self.u[:-1] = self.u[1:]; self.u[-1] = self.u[-2], i.e. rows
    # 1..K-1 followed by row K-1 again). The tracer has no in-place slice
    # assignment, so the shift is K row slices + a stack.
    u_0 = self.bk.slice_(u_updated, 0, 1)
    if K > 1:
        rows = [self.bk.ravel(self.bk.slice_(u_updated, j, j + 1)) for j in range(1, K)]
        rows.append(self.bk.ravel(self.bk.slice_(u_updated, K - 1, K)))
        shifted = self.bk.stack(rows)
    else:
        shifted = u_updated
    self.u = self.bk.to_numpy(shifted)

    u_0 = self.bk.ravel(u_0)
    if self.u_min is not None or self.u_max is not None:
        u_0 = self.bk.clip(u_0, self.u_min, self.u_max)

    return u_0

reset 

reset()

Reset the controller to its initial state.

Zeros the nominal control sequence and clears the last-sample bookkeeping attributes.

Source code in src/shinro/controllers/mppi.py 
407
408
409
410
411
412
413
414
415
def reset(self):
    """Reset the controller to its initial state.

    Zeros the nominal control sequence and clears the last-sample
    bookkeeping attributes.
    """
    self.u = np.zeros((self.K, self.D_u))
    self._last_epsilon = None
    self._last_costs = None

from_config classmethod 

from_config(config, backend: ArrayBackend | None = None)

Create an MPPI controller from a TOML config dict or MPPIConfig.

Config fields

num_samples: Number of sampled perturbations N. temperature: Softmax temperature. horizon: Prediction horizon K. noise_sigma: Per-channel perturbation std dev (D_u,). dt: Time step. Required at runtime; injected from the plant in scenario builds. u_min: Optional lower bound list (D_u,). u_max: Optional upper bound list (D_u,). seed: Optional RNG seed. state_cost: Optional diagonal Q weights (D_x,) or full Q matrix. control_cost: Optional diagonal R weights (D_u,) or full R matrix.

The dynamics_fn and cost_fn callables cannot be serialized to TOML. They are created as None and must be injected after construction, either by setting the attributes directly or by calling attach_plant(plant):

.. code-block:: python

ctrl = MPPIController.from_config(config)
ctrl.dynamics_fn = my_dynamics
ctrl.cost_fn = my_cost
# or:
ctrl.attach_plant(plant)

Parameters:

Name Type Description Default
config

TOML config dict or MPPIConfig.

 required
backend ArrayBackend | None

Array backend. Defaults to NumpyBackend.

 None

Returns:

Type Description

MPPIController instance.
Source code in src/shinro/controllers/mppi.py 
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
@classmethod
def from_config(cls, config, backend: ArrayBackend | None = None):
    """Create an MPPI controller from a TOML config dict or :class:`MPPIConfig`.

    Config fields:
        num_samples: Number of sampled perturbations N.
        temperature: Softmax temperature.
        horizon: Prediction horizon K.
        noise_sigma: Per-channel perturbation std dev (D_u,).
        dt: Time step. Required at runtime; injected from the plant in
            scenario builds.
        u_min: Optional lower bound list (D_u,).
        u_max: Optional upper bound list (D_u,).
        seed: Optional RNG seed.
        state_cost: Optional diagonal Q weights (D_x,) or full Q matrix.
        control_cost: Optional diagonal R weights (D_u,) or full R matrix.

    The ``dynamics_fn`` and ``cost_fn`` callables cannot be serialized to
    TOML. They are created as ``None`` and must be injected after
    construction, either by setting the attributes directly or by calling
    ``attach_plant(plant)``:

    .. code-block:: python

        ctrl = MPPIController.from_config(config)
        ctrl.dynamics_fn = my_dynamics
        ctrl.cost_fn = my_cost
        # or:
        ctrl.attach_plant(plant)

    Args:
        config: TOML config dict or MPPIConfig.
        backend: Array backend. Defaults to NumpyBackend.

    Returns:
        MPPIController instance.
    """
    bk = backend or NumpyBackend()
    cfg = cls.parse_config(config)
    if cfg.dt is None:
        raise ValueError(
            "MPPI: dt is required (rollout stepping) — omit it only in scenario "
            "builds, where the plant's dt is injected"
        )
    ctrl = cls(
        dynamics_fn=None,
        cost_fn=None,
        num_samples=cfg.num_samples,
        temperature=cfg.temperature,
        dt=cfg.dt,
        horizon=cfg.horizon,
        noise_sigma=cfg.noise_sigma,
        u_min=cfg.u_min,
        u_max=cfg.u_max,
        seed=cfg.seed,
        backend=bk,
    )
    ctrl._Q = bk.array(cfg.state_cost) if cfg.state_cost is not None else None
    ctrl._R = bk.array(cfg.control_cost) if cfg.control_cost is not None else None
    return ctrl

LeRobotDiffusionAdapter

Bases: Controller

Wrap a LeRobot diffusion policy as a Controller.

The policy is loaded from a Hugging Face checkpoint or local path. The compute() method converts the plant state into LeRobot's observation dict format, runs the policy, and returns the action.

Accepts both numpy arrays and torch tensors for state and camera frames, so it works with any ArrayBackend.

Parameters:

Name Type Description Default
policy

Loaded LeRobot policy instance.

 required
use_camera bool

Whether to expect camera observations.

 False
device str

Torch device for policy inference.

 'cpu'
Source code in src/shinro/controllers/lerobot_adapter.py 
37
38
39
40
41
def __init__(self, policy, use_camera: bool = False, device: str = "cpu"):
    self.policy = policy
    self.use_camera = use_camera
    self.device = device
    self._latest_camera_frame = None

update_camera 

update_camera(frame)

Feed a camera frame for the next policy step.

Parameters:

Name Type Description Default
frame

Camera image as numpy array (HWC) or torch tensor (CHW or HWC).

 required
Source code in src/shinro/controllers/lerobot_adapter.py 
43
44
45
46
47
48
49
def update_camera(self, frame):
    """Feed a camera frame for the next policy step.

    Args:
        frame: Camera image as numpy array (HWC) or torch tensor (CHW or HWC).
    """
    self._latest_camera_frame = frame

compute 

compute(state, target=None)

Run the LeRobot policy on the current state.

Builds a LeRobot observation dict from the state vector, runs the policy, and returns the action as a numpy array.

Parameters:

Name Type Description Default
state

Plant state vector (n_x,). Can be numpy or torch tensor.

 required
target

Ignored for learned policies — they generate actions from observation alone.

 None

Returns:

Type Description

Action vector (n_u,) as numpy array.
Source code in src/shinro/controllers/lerobot_adapter.py 
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
def compute(self, state, target=None):
    """Run the LeRobot policy on the current state.

    Builds a LeRobot observation dict from the state vector, runs the
    policy, and returns the action as a numpy array.

    Args:
        state: Plant state vector (n_x,). Can be numpy or torch tensor.
        target: Ignored for learned policies — they generate actions
            from observation alone.

    Returns:
        Action vector (n_u,) as numpy array.
    """
    import torch

    if isinstance(state, torch.Tensor):
        obs = {"observation.state": state.float().unsqueeze(0)}
    else:
        obs = {"observation.state": torch.from_numpy(state).float().unsqueeze(0)}

    if self.use_camera and self._latest_camera_frame is not None:
        frame = self._latest_camera_frame
        if isinstance(frame, torch.Tensor):
            frame_tensor = frame.float()
        else:
            frame_tensor = torch.from_numpy(frame).float()
        if frame_tensor.ndim == 3:
            frame_tensor = frame_tensor.permute(2, 0, 1)
        obs["observation.images.cam"] = frame_tensor.unsqueeze(0)
        self._latest_camera_frame = None

    obs = {k: v.to(self.device) for k, v in obs.items()}
    with torch.no_grad():
        action = self.policy.select_action(obs)

    return action.squeeze(0).cpu().numpy()

reset 

reset()

Reset the policy's internal state.

Source code in src/shinro/controllers/lerobot_adapter.py 
89
90
91
92
def reset(self):
    """Reset the policy's internal state."""
    self.policy.reset()
    self._latest_camera_frame = None

from_config classmethod 

from_config(config)

Load a LeRobot policy from a config dict.

Config fields

policy_type: Type of policy ("diffusion", "act", "pi0", etc.). checkpoint: Hugging Face repo ID or local path. use_camera: Whether to expect camera observations (default: false). device: Torch device (default: "cpu").

Parameters:

Name Type Description Default
config

TOML config dict.

 required

Returns:

Type Description

LeRobotDiffusionAdapter instance.
Source code in src/shinro/controllers/lerobot_adapter.py 
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
@classmethod
def from_config(cls, config):
    """Load a LeRobot policy from a config dict.

    Config fields:
        policy_type: Type of policy (``"diffusion"``, ``"act"``, ``"pi0"``, etc.).
        checkpoint: Hugging Face repo ID or local path.
        use_camera: Whether to expect camera observations (default: false).
        device: Torch device (default: ``"cpu"``).

    Args:
        config: TOML config dict.

    Returns:
        LeRobotDiffusionAdapter instance.
    """
    policy_type = config["policy_type"]
    checkpoint = config["checkpoint"]
    use_camera = config.get("use_camera", False)
    device = config.get("device", "cpu")

    from lerobot.policies import make_policy  # type: ignore
    policy = make_policy(policy_type, pretrained_path=checkpoint)
    policy.to(device)
    policy.eval()

    return cls(policy, use_camera=use_camera, device=device)

OnnxRLAdapter

Bases: Controller

Run an ONNX-exported RL policy as a Controller.

The policy is prepared once (imported + interpreted, or a compiled .so is loaded) and each compute call runs one tick on the plant state.

Parameters:

Name Type Description Default
policy _GraphPolicy | _CompiledPolicy

A loaded policy artifact — _GraphPolicy (eager) or _CompiledPolicy (compiled). Built by from_config.

 required
seed int

RNG seed for action sampling. Sampling action spaces (discrete non-deterministic, stochastic non-deterministic) draw their noise from a generator seeded here.

 0
backend ArrayBackend | None

Array backend for the state input and action output. The kernel itself always works on numpy f64; the adapter converts at the boundary, so a torch state yields a torch action.

 None
Source code in src/shinro/controllers/onnx_rl_adapter.py 
277
278
279
280
281
def __init__(self, policy: _GraphPolicy | _CompiledPolicy, *, seed: int = 0, backend: ArrayBackend | None = None) -> None:
    self.policy = policy
    self.seed = seed
    self.bk = backend or NumpyBackend()
    self._rng = np.random.default_rng(self.seed)

compute 

compute(state, target=None)

Run the policy on the current plant state.

Parameters:

Name Type Description Default
state

Plant state vector in the configured backend's native type (numpy array, torch tensor, or a sequence). Must have the compiled graph's state port length.

 required
target

Ignored — learned policies act on the observation alone.

 None

Returns:

Type Description

The action vector (n_u,) in the backend's native type.
Source code in src/shinro/controllers/onnx_rl_adapter.py 
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
def compute(self, state, target=None):
    """Run the policy on the current plant state.

    Args:
        state: Plant state vector in the configured backend's native type
            (numpy array, torch tensor, or a sequence). Must have the
            compiled graph's ``state`` port length.
        target: Ignored — learned policies act on the observation alone.

    Returns:
        The action vector (n_u,) in the backend's native type.
    """
    x = np.asarray(self.bk.to_numpy(state), dtype=np.float64).ravel()
    if x.size != self.policy.state_size:
        raise ValueError(f"onnx_rl: expected a state of {self.policy.state_size} elements, got {x.size}")
    feed: dict[str, np.ndarray] = {self.policy.state_port: x}
    if self.policy.noise_port is not None:
        feed[self.policy.noise_port] = self._draw_noise(self.policy.noise_size)
    for name, size in self.policy.noise_ports:
        feed[name] = self._rng.standard_normal(size)
    return self.bk.from_numpy(self.policy.step(feed))

reset 

reset()

Reseed the action-sampling RNG and clear any recurrent state.

Source code in src/shinro/controllers/onnx_rl_adapter.py 
319
320
321
322
def reset(self):
    """Reseed the action-sampling RNG and clear any recurrent state."""
    self._rng = np.random.default_rng(self.seed)
    self.policy.reset()

from_config classmethod 

from_config(config, backend: ArrayBackend | None = None)

Create an OnnxRLAdapter from a TOML config dict or OnnxRLConfig.

Config fields

model_path: Path to the .onnx model (eager mode). artifact_dir: A make compile --out directory (compiled mode). action_space: "continuous", "discrete", or "stochastic" (default: continuous). deterministic: Return argmax/mean instead of sampling (default true). action_scale / action_bias: Post-policy affine transform. action_clip_low / action_clip_high: Clip the final action; both are required together (a missing bound would be ±inf, which the lowerer cannot represent). seed: RNG seed for sampling action spaces. n_x: Plant state dimension, when the observation sub-table does not reach the last state entry. output_name: ONNX tensor to import as the action. observation: The importer's observation table (state_keys, normalize, obs_mean, obs_std, clip, ...).

Parameters:

Name Type Description Default
config

TOML config dict or OnnxRLConfig.

 required
backend ArrayBackend | None

Array backend for state input and action output.

 None

Returns:

Type Description

OnnxRLAdapter instance.

Raises:

Type Description
ValueError

On a missing/invalid mode or an invalid action config.
FileNotFoundError

If a compiled artifact is missing.
Source code in src/shinro/controllers/onnx_rl_adapter.py 
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
@classmethod
def from_config(cls, config, backend: ArrayBackend | None = None):
    """Create an OnnxRLAdapter from a TOML config dict or :class:`OnnxRLConfig`.

    Config fields:
        model_path: Path to the ``.onnx`` model (eager mode).
        artifact_dir: A ``make compile --out`` directory (compiled mode).
        action_space: ``"continuous"``, ``"discrete"``, or ``"stochastic"``
            (default: continuous).
        deterministic: Return argmax/mean instead of sampling (default true).
        action_scale / action_bias: Post-policy affine transform.
        action_clip_low / action_clip_high: Clip the final action; both are
            required together (a missing bound would be ``±inf``, which the
            lowerer cannot represent).
        seed: RNG seed for sampling action spaces.
        n_x: Plant state dimension, when the observation sub-table does not
            reach the last state entry.
        output_name: ONNX tensor to import as the action.
        observation: The importer's observation table (``state_keys``,
            ``normalize``, ``obs_mean``, ``obs_std``, ``clip``, ...).

    Args:
        config: TOML config dict or :class:`OnnxRLConfig`.
        backend: Array backend for state input and action output.

    Returns:
        OnnxRLAdapter instance.

    Raises:
        ValueError: On a missing/invalid mode or an invalid action config.
        FileNotFoundError: If a compiled artifact is missing.
    """
    cfg = cls.parse_config(config)
    bk = backend or NumpyBackend()

    if cfg.artifact_dir is not None:
        policy: _GraphPolicy | _CompiledPolicy = _CompiledPolicy(cfg.artifact_dir)
    elif cfg.model_path is not None:
        from shinro.codegen.onnx_import import import_onnx_policy

        action_cfg: dict[str, Any] = {
            "action_space": cfg.action_space,
            "deterministic": cfg.deterministic,
            "action_scale": cfg.action_scale,
            "action_bias": cfg.action_bias,
        }
        if cfg.action_clip_low is not None:
            action_cfg["action_clip_low"] = cfg.action_clip_low
        if cfg.action_clip_high is not None:
            action_cfg["action_clip_high"] = cfg.action_clip_high
        cg = import_onnx_policy(
            cfg.model_path,
            n_x=cfg.n_x,
            obs_cfg=cfg.observation,
            action_cfg=action_cfg,
            output_name=cfg.output_name,
        )
        policy = _GraphPolicy(cg)
    else:
        raise ValueError("onnx_rl: config needs model_path (eager) or artifact_dir (compiled)")

    return cls(policy, seed=cfg.seed, backend=bk)

SlidingModeController

Bases: Controller

Sliding Mode Controller for nonlinear systems.

Implements the equivalent control approach with a switching term and optional boundary-layer smoothing. The sliding surface coefficients c must form a Hurwitz polynomial.

One surface per instance: c is a single surface vector, so the class is instantiated once per controlled axis (a bank of instances is the multi-axis pattern). Multi-surface SMC — a surface matrix solved jointly — is intentionally not implemented; see the module docstring's "Scope" note.

Parameters:

Name Type Description Default
c

Sliding surface coefficients (n,). The polynomial c[0] + c[1] p + ... + c[n-1] p^{n-1} must be Hurwitz.

 required
k1 float

Discontinuous (switching) gain — drives the state to the surface.

 required
phi float

Boundary layer thickness for chattering suppression. If 0, uses sign (pure switching).

 0.0
k2 float

Linear (proportional) gain on the sliding variable.

 0.0
smoother str

Boundary-layer smoothing function. One of "sat", "tanh", or "sigmoid".

 'sat'
alpha float

Fractional power exponent for the switching term \(|s|^\alpha\). 0 gives sign-only; 1 gives linear.

 0.0
controllability_eps float

Near-zero threshold on the controllability denominator: \(|c^T g(x)|\) for a scalar input, and its multi-input generalization \(\|c^T g(x)\|\) (the two agree when n_u == 1). A deployment design parameter, not a numerical constant — set it above the smallest value the plant can legitimately produce (the law amplifies \(1/c^T g\), so command saturation and chattering arrive long before the arithmetic floor). The 1e-12 default only protects the division itself, for plants whose g is well-conditioned everywhere. Live numpy calls raise RuntimeError below it; the lowered (compiled) graph instead emits a fail-safe zero command and reports a healthy flag, since a straight-line kernel cannot raise.

 1e-12
backend ArrayBackend | None

Array backend. Defaults to NumpyBackend.

 None
Source code in src/shinro/controllers/smc.py 
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
def __init__(
    self,
    c,
    k1: float,
    phi: float = 0.0,
    k2: float = 0.0,
    smoother: str = "sat",
    alpha: float = 0.0,
    controllability_eps: float = 1e-12,
    backend: ArrayBackend | None = None,
):
    self.bk = backend or NumpyBackend()
    self.c = self.bk.array(c).flatten()
    self.k1 = _as_float(k1, "k1")
    self.k2 = _as_float(k2, "k2")
    self.phi = _as_float(phi, "phi")
    self.alpha = _as_float(alpha, "alpha")
    self.controllability_eps = _as_float(controllability_eps, "controllability_eps")

    SMOOTHERS = self._dict_boundaries()

    if smoother not in SMOOTHERS:
        raise ValueError(f"Unknown smoother '{smoother}'. Options: {list(SMOOTHERS)}")

    if not self._is_hurwitz():
        raise ValueError(
            "Sliding surface coefficients are not Hurwitz. "
            "The polynomial c1 + c2 p + ... + cn p^{n-1} must have "
            "all roots with negative real parts."
        )

    self._smoother = SMOOTHERS[smoother]
    self._smoother_name = smoother

n property 

n: int

Number of sliding surface coefficients (state dimension).

compute 

compute(x, f_x, g_x)

Compute the sliding mode control action.

Evaluates \(u = (c^T g)^{-1} ( -c^T f - k_1 |s|^\alpha \, \text{smooth}(s) - k_2 s )\).

For scalar input (c^T g is scalar), uses direct division. For vector input the equivalent-control equation (c^T g) u = num is underdetermined (one surface, n_u unknowns), so the law takes its minimum-norm solution u = cg^T (cg cg^T)^{-1} num — the same point np.linalg.lstsq returned, but written as matmul/transpose/ divide so it traces and lowers.

The computation is written so it traces as well as it evaluates: the u/n_u branch keys off g_x's shape (known at trace time), the dot products use column-vector form (the tracer has no 1D @ 1D contraction), and the near-zero c^T g guard is an exception only when g_x is concrete — a graph cannot raise, so the traced path emits a fail-safe instead (see _scalar_control).

Parameters:

Name Type Description Default
x

Current state vector (n,).

 required
f_x

Drift dynamics \(f(x)\) evaluated at x (n,).

 required
g_x

Control matrix \(g(x)\) evaluated at x (n, n_u).

 required

Returns:

Type Description

Control input vector (n_u,).

Raises:

Type Description
RuntimeError

If \(c^T g(x)\) is near-zero and g_x is concrete (eager backends only; the traced path cannot raise).
Source code in src/shinro/controllers/smc.py 
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
def compute(self, x, f_x, g_x):
    """Compute the sliding mode control action.

    Evaluates :math:`u = (c^T g)^{-1} ( -c^T f - k_1 |s|^\\alpha \\, \\text{smooth}(s) - k_2 s )`.

    For scalar input (``c^T g`` is scalar), uses direct division. For
    vector input the equivalent-control equation ``(c^T g) u = num`` is
    underdetermined (one surface, ``n_u`` unknowns), so the law takes its
    minimum-norm solution ``u = cg^T (cg cg^T)^{-1} num`` — the same
    point ``np.linalg.lstsq`` returned, but written as matmul/transpose/
    divide so it traces and lowers.

    The computation is written so it traces as well as it evaluates: the
    ``u``/``n_u`` branch keys off ``g_x``'s *shape* (known at trace time),
    the dot products use column-vector form (the tracer has no 1D @ 1D
    contraction), and the near-zero ``c^T g`` guard is an exception only
    when ``g_x`` is concrete — a graph cannot raise, so the traced path
    emits a fail-safe instead (see :meth:`_scalar_control`).

    Args:
        x: Current state vector (n,).
        f_x: Drift dynamics :math:`f(x)` evaluated at x (n,).
        g_x: Control matrix :math:`g(x)` evaluated at x (n, n_u).

    Returns:
        Control input vector (n_u,).

    Raises:
        RuntimeError: If :math:`c^T g(x)` is near-zero and ``g_x`` is
            concrete (eager backends only; the traced path cannot raise).
    """
    x = self.bk.ravel(self.bk.array(x))
    f_x = self.bk.ravel(self.bk.array(f_x))
    g_x = self.bk.array(g_x)
    if len(g_x.shape) == 1:
        # 1-D g_x is a single control column; normalize to (n, 1).
        g_x = self.bk.reshape(g_x, (g_x.shape[0], 1))

    n = self.n
    # Column-vector contractions: (n,) @ (n, 1) -> (1,) and
    # (n,) @ (n, n_u) -> (n_u,). numpy would collapse 1D @ 1D to a 0-d
    # scalar, which the tracer rejects (no 0-d propagation) — same numbers,
    # a uniform (1,)/(n_u,) shape on every backend.
    s = self.c @ self.bk.reshape(x, (n, 1))
    cf = self.c @ self.bk.reshape(f_x, (n, 1))
    cg = self.bk.ravel(self.c @ g_x)

    if self.phi > 0:
        smooth_s = self._smoother(s)
    else:
        smooth_s = self.bk.sign(s)

    s_dot_desired = -self.k1 * self.bk.abs(s) ** self.alpha * smooth_s - self.k2 * s
    num = s_dot_desired - cf

    if len(g_x.shape) >= 2 and g_x.shape[-1] > 1:
        return self._vector_control(num, cg)
    return self._scalar_control(num, cg)

reset 

reset()

No internal state to reset for SMC.

Source code in src/shinro/controllers/smc.py 
351
352
def reset(self):
    """No internal state to reset for SMC."""

from_config classmethod 

from_config(config, backend: ArrayBackend | None = None)

Create an SMC controller from a TOML config dict or SMCConfig.

Config fields

c: List of sliding surface coefficients (n,). k1: Discontinuous (switching) gain. phi: Boundary layer thickness (default 0.0). k2: Linear gain on sliding variable (default 0.0). smoother: Smoothing function — "sat", "tanh", or "sigmoid" (default "sat"). alpha: Fractional power exponent (default 0.0). controllability_eps: Near-zero \(|c^T g|\) threshold, plant-scaled (default 1e-12 = arithmetic-only guard); see SlidingModeController for how to choose it.

Parameters:

Name Type Description Default
config

TOML config dict or SMCConfig.

 required
backend ArrayBackend | None

Array backend. Defaults to NumpyBackend.

 None

Returns:

Type Description

SlidingModeController instance.
Source code in src/shinro/controllers/smc.py 
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
@classmethod
def from_config(cls, config, backend: ArrayBackend | None = None):
    """Create an SMC controller from a TOML config dict or :class:`SMCConfig`.

    Config fields:
        c: List of sliding surface coefficients (n,).
        k1: Discontinuous (switching) gain.
        phi: Boundary layer thickness (default 0.0).
        k2: Linear gain on sliding variable (default 0.0).
        smoother: Smoothing function — ``"sat"``, ``"tanh"``, or
            ``"sigmoid"`` (default ``"sat"``).
        alpha: Fractional power exponent (default 0.0).
        controllability_eps: Near-zero :math:`|c^T g|` threshold,
            plant-scaled (default 1e-12 = arithmetic-only guard); see
            :class:`SlidingModeController` for how to choose it.

    Args:
        config: TOML config dict or SMCConfig.
        backend: Array backend. Defaults to NumpyBackend.

    Returns:
        SlidingModeController instance.
    """
    bk = backend or NumpyBackend()
    cfg = cls.parse_config(config)
    return cls(
        c=bk.array(cfg.c),
        k1=cfg.k1,
        phi=cfg.phi,
        k2=cfg.k2,
        smoother=cfg.smoother,
        alpha=cfg.alpha,
        controllability_eps=cfg.controllability_eps,
        backend=bk,
    )