Utilities¶

shinro.utils

`parse_matrix`¶

Convert a TOML config value to a matrix.

Flat list (e.g. [1.0, 2.0, 3.0]) → diagonal matrix. Nested list (e.g. [[1, 0], [0, 2]]) → full matrix.

Parameters:

Name	Type	Description	Default
`bk`		ArrayBackend instance.	required
`value`		List or list-of-lists from a TOML config.	required

Returns:

Type	Description
	Array in the backend's native type.

Source code in src/shinro/utils/array_backend.py

def parse_matrix(bk, value):
    """Convert a TOML config value to a matrix.

    Flat list (e.g. ``[1.0, 2.0, 3.0]``) → diagonal matrix.
    Nested list (e.g. ``[[1, 0], [0, 2]]``) → full matrix.

    Args:
        bk: ArrayBackend instance.
        value: List or list-of-lists from a TOML config.

    Returns:
        Array in the backend's native type.
    """
    if isinstance(value, list) and len(value) > 0 and isinstance(value[0], list):
        return bk.array(value)
    return bk.diag(value)

`ArrayBackend`¶

Bases: ABC

Abstract interface for array operations used by all components.

Every component takes an optional backend parameter. If None, NumpyBackend is used. This lets the entire control stack switch between numpy and torch by changing one object.

The @ operator (matmul) is intentionally not wrapped — it works identically in numpy and torch for 2D arrays.

emit_named_output ¶

emit_named_output(name: str, value: Any) -> None

Publish an auxiliary named signal from a compute/estimate call.

A concrete no-op for eager backends (numpy/torch): the signal is already available to the caller, so nothing to publish. The tracing backend overrides this to record a named graph output port, which is how a component exposes a diagnostic alongside its primary return value without changing its return contract (e.g. SMC's healthy controllability flag).

Parameters:

Name	Type	Description	Default
`name`	`str`	The output port name.	required
`value`	`Any`	The signal (any array-like the backend produced).	required

Source code in src/shinro/utils/array_backend.py

def emit_named_output(self, name: str, value: Any) -> None:
    """Publish an auxiliary named signal from a ``compute``/``estimate`` call.

    A concrete no-op for eager backends (numpy/torch): the signal is
    already available to the caller, so nothing to publish. The tracing
    backend overrides this to record a named graph ``output`` port, which
    is how a component exposes a diagnostic alongside its primary return
    value without changing its return contract (e.g. SMC's ``healthy``
    controllability flag).

    Args:
        name: The output port name.
        value: The signal (any array-like the backend produced).
    """
    return None

solve_qp `abstractmethod` ¶

solve_qp(q, H, A, lb, ub) -> Any

Solve a convex QP min ½ uᵀ H u + qᵀ u s.t. lb ≤ A u ≤ ub.

The solver is C-based (OSQP) and always runs on numpy, so this is the per-step bridge for MPC: q is the state-dependent linear cost, H/lb/ub are backend arrays, and A is a scipy sparse constraint matrix (from MPC_LTI.constraints). Returns the full solution u* (length = q length); MPC slices out the first m controls.

Returns:

Type	Description
`Any`	The full QP solution vector, in the backend's native type.

Source code in src/shinro/utils/array_backend.py

@abstractmethod
def solve_qp(self, q, H, A, lb, ub) -> Any:
    """Solve a convex QP ``min ½ uᵀ H u + qᵀ u`` s.t. ``lb ≤ A u ≤ ub``.

    The solver is C-based (OSQP) and always runs on numpy, so this is the
    per-step bridge for MPC: ``q`` is the state-dependent linear cost,
    ``H``/``lb``/``ub`` are backend arrays, and ``A`` is a scipy sparse
    constraint matrix (from ``MPC_LTI.constraints``). Returns the full
    solution ``u*`` (length = ``q`` length); MPC slices out the first
    ``m`` controls.

    Returns:
        The full QP solution vector, in the backend's native type.
    """
    ...

jacobian `abstractmethod` ¶

jacobian(f, x, eps=1e-06) -> Any

Compute the Jacobian of f at x using finite differences or autograd.

Parameters:

Name	Description	Default
`f`	Callable `f(x) -> y` where x is shape (n,) and y is shape (m,).	required
`x`	Point at which to evaluate the Jacobian, shape (n,).	required
`eps`	Step size for finite differences (ignored by autograd backends).	`1e-06`

Returns:

Type	Description
`Any`	Jacobian matrix of shape (m, n).

Source code in src/shinro/utils/array_backend.py

@abstractmethod
def jacobian(self, f, x, eps=1e-6) -> Any:
    """Compute the Jacobian of f at x using finite differences or autograd.

    Args:
        f: Callable ``f(x) -> y`` where x is shape (n,) and y is shape (m,).
        x: Point at which to evaluate the Jacobian, shape (n,).
        eps: Step size for finite differences (ignored by autograd backends).

    Returns:
        Jacobian matrix of shape (m, n).
    """
    ...

`NumpyBackend`¶

Bases: ArrayBackend

ArrayBackend implementation using numpy.

All operations delegate directly to np.xxx. The to_numpy and from_numpy methods are no-ops since the data is already numpy.

solve_qp ¶

solve_qp(q, H, A, lb, ub)

Solve the MPC QP with OSQP (numpy-native at the C boundary).

Uses eps=1e-6 to match the codegen-baked static solver the Zig VM drives (scripts/gen_emosqp_test.py), so the live numpy path agrees with the shipped .so. On a non-solved status, returns zeros (preserving MPC's original fallback).

Source code in src/shinro/utils/array_backend.py

def solve_qp(self, q, H, A, lb, ub):
    """Solve the MPC QP with OSQP (numpy-native at the C boundary).

    Uses ``eps=1e-6`` to match the codegen-baked static solver the Zig VM
    drives (scripts/gen_emosqp_test.py), so the live numpy path agrees
    with the shipped ``.so``. On a non-solved status, returns zeros
    (preserving MPC's original fallback).
    """
    import osqp
    from scipy import sparse

    q_np = np.asarray(q).ravel()
    l_np = np.asarray(lb).ravel()
    u_np = np.asarray(ub).ravel()
    prob = osqp.OSQP()
    prob.setup(
        sparse.csc_matrix(np.asarray(H, dtype=np.float64)),
        q_np,
        A,
        l_np,
        u_np,
        warm_starting=True,
        verbose=False,
    )
    prob.update_settings(eps_abs=1e-6, eps_rel=1e-6)
    res = prob.solve()
    if res.info.status != "solved":
        return np.zeros(q_np.size)
    return res.x

jacobian ¶

jacobian(f, x, eps=1e-06)

Central finite-difference Jacobian.

Parameters:

Name	Description	Default
`f`	Callable `f(x) -> y` where x is shape (n,) and y is shape (m,).	required
`x`	Point at which to evaluate, shape (n,).	required
`eps`	Step size for finite differences.	`1e-06`

Returns:

Type	Description
	Jacobian matrix of shape (m, n).

Source code in src/shinro/utils/array_backend.py

def jacobian(self, f, x, eps=1e-6):
    """Central finite-difference Jacobian.

    Args:
        f: Callable ``f(x) -> y`` where x is shape (n,) and y is shape (m,).
        x: Point at which to evaluate, shape (n,).
        eps: Step size for finite differences.

    Returns:
        Jacobian matrix of shape (m, n).
    """
    x = np.asarray(x, dtype=np.float64)
    fx = np.atleast_1d(np.asarray(f(x), dtype=np.float64))
    m = fx.shape[0]
    n = x.shape[0]
    J = np.zeros((m, n), dtype=np.float64)
    for i in range(n):
        h = np.zeros(n, dtype=np.float64)
        h[i] = eps
        J[:, i] = (np.atleast_1d(np.asarray(f(x + h), dtype=np.float64)) - np.atleast_1d(np.asarray(f(x - h), dtype=np.float64))) / (
            2.0 * eps
        )
    return J

`TorchBackend`¶

Bases: ArrayBackend

ArrayBackend implementation using PyTorch.

All operations delegate to torch.xxx. Data lives on the device specified at construction time. The to_numpy and from_numpy methods handle device transfers.

Parameters:

Name	Type	Description	Default
`device`		Torch device string (e.g. `"cpu"`, `"cuda"`).	`'cpu'`

Source code in src/shinro/utils/array_backend.py

def __init__(self, device="cpu"):
    import torch

    self.torch = torch
    self.device = device

solve_qp ¶

solve_qp(q, H, A, lb, ub)

Solve the MPC QP with OSQP, converting across the tensor boundary.

q/H/lb/ub may be torch tensors (from the config-baked matrices); A is a scipy sparse matrix. Converts to numpy, solves with eps=1e-6 (matching the codegen static solver), and returns the solution as a tensor.

Source code in src/shinro/utils/array_backend.py

def solve_qp(self, q, H, A, lb, ub):
    """Solve the MPC QP with OSQP, converting across the tensor boundary.

    q/H/lb/ub may be torch tensors (from the config-baked matrices); A is a
    scipy sparse matrix. Converts to numpy, solves with eps=1e-6 (matching
    the codegen static solver), and returns the solution as a tensor.
    """
    import osqp
    from scipy import sparse

    def _np(x):
        if isinstance(x, self.torch.Tensor):
            return x.detach().cpu().numpy()
        return np.asarray(x)

    q_np = _np(q).ravel()
    l_np = _np(lb).ravel()
    u_np = _np(ub).ravel()
    prob = osqp.OSQP()
    prob.setup(
        sparse.csc_matrix(np.asarray(_np(H), dtype=np.float64)),
        q_np,
        A,
        l_np,
        u_np,
        warm_starting=True,
        verbose=False,
    )
    prob.update_settings(eps_abs=1e-6, eps_rel=1e-6)
    res = prob.solve()
    if res.info.status != "solved":
        return self.torch.zeros(q_np.size, device=self.device, dtype=self.torch.float64)
    return self.torch.tensor(res.x, device=self.device, dtype=self.torch.float64)

jacobian ¶

jacobian(f, x, eps=1e-06)

Jacobian via torch.autograd.functional.jacobian.

Parameters:

Name	Description	Default
`f`	Callable `f(x) -> y` where x is shape (n,) and y is shape (m,).	required
`x`	Point at which to evaluate, shape (n,).	required
`eps`	Ignored (used for API compatibility with NumpyBackend).	`1e-06`

Returns:

Type	Description
	Jacobian matrix of shape (m, n).

Source code in src/shinro/utils/array_backend.py

def jacobian(self, f, x, eps=1e-6):
    """Jacobian via torch.autograd.functional.jacobian.

    Args:
        f: Callable ``f(x) -> y`` where x is shape (n,) and y is shape (m,).
        x: Point at which to evaluate, shape (n,).
        eps: Ignored (used for API compatibility with NumpyBackend).

    Returns:
        Jacobian matrix of shape (m, n).
    """
    if not isinstance(x, self.torch.Tensor):
        x = self.torch.tensor(x, dtype=self.torch.float64)
    x_t = x.detach().clone().requires_grad_(True)
    y = f(x_t)
    if y.ndim == 0:
        y = y.unsqueeze(0)
    J = self.torch.autograd.functional.jacobian(f, x_t)
    return J.reshape(y.shape[0], x_t.shape[0])

`BatchedDynamicsAdapter`¶

Adapt a Plant's single-state dynamics/cost to batched (N, ...) arrays.

The adapter detects the dynamics path once at construction:

plant.dynamics returning non-None (the ABC default is None) — the nonlinear path, integrating the plant's batch-capable derivative with explicit Euler. Trace-safe.
Otherwise the LTI matmul path.

Arrays live in the plant's backend, or in the backend passed to dynamics_fn (tracing passes the component's TraceBackend).

Usage

adapter = BatchedDynamicsAdapter(plant) x_next = adapter.dynamics_fn(x_batch, u_batch, dt) cost = adapter.cost_fn(x_batch, u_batch, Q, R)

Source code in src/shinro/utils/batched_adapter.py

def __init__(self, plant: Plant):
    self.plant = plant
    self.bk: ArrayBackend = getattr(plant, "bk", None) or NumpyBackend()
    self.dt = getattr(plant, "dt", 0.01)

    A, B = plant.get_model()
    self._A = A
    self._B = B
    self.D_x = self._A.shape[0]
    self.D_u = self._B.shape[1]

    # Path dispatch: a plant that overrides ``dynamics`` (the ABC default
    # is None) rolls out nonlinearly; otherwise the linearized (A, B)
    # matmul is exact and cheaper. The probe doubles as the capability
    # check the ABC used to leave to ``dynamics() is not None``.
    probe = plant.dynamics(state=self.bk.zeros(self.D_x), control=self.bk.zeros(self.D_u))
    self._nonlinear = probe is not None
    self._path = "nonlinear" if self._nonlinear else "lti"

state_dim `property` ¶

state_dim: int

State dimension \(D_x\).

control_dim `property` ¶

control_dim: int

Control dimension \(D_u\).

dynamics_path `property` ¶

dynamics_path: str

Which dynamics path this adapter dispatches to.

"nonlinear" (the plant's own batch-capable derivative, integrated with explicit Euler) or "lti" (one batched matmul).

dynamics_fn ¶

dynamics_fn(x_batch, u_batch, dt: float, bk: Any | None = None)

Batched dynamics update.

Dispatches to the path chosen at construction. bk is the backend the plant's nonlinear dynamics evaluates with, defaulting to the plant's own: tracing passes the component's TraceBackend here, because trace_node swaps only the traced component's self.bk — a plant calling self.bk.sin on a tracer would otherwise reach a concrete backend and fail. The LTI path ignores bk (operator arithmetic already routes through the tracer).

Parameters:

Name	Type	Description	Default
`x_batch`		Batch of states (N, D_x).	required
`u_batch`		Batch of controls (N, D_u).	required
`dt`	`float`	Time step (s).	required
`bk`	`Any \| None`	Backend for the nonlinear path; defaults to the plant's backend.	`None`

Returns:

Type	Description
	Batch of next states (N, D_x).

Source code in src/shinro/utils/batched_adapter.py

def dynamics_fn(self, x_batch, u_batch, dt: float, bk: Any | None = None):
    """Batched dynamics update.

    Dispatches to the path chosen at construction. ``bk`` is the backend
    the plant's nonlinear ``dynamics`` evaluates with, defaulting to the
    plant's own: tracing passes the component's ``TraceBackend`` here,
    because ``trace_node`` swaps only the traced component's ``self.bk`` —
    a plant calling ``self.bk.sin`` on a tracer would otherwise reach a
    concrete backend and fail. The LTI path ignores ``bk`` (operator
    arithmetic already routes through the tracer).

    Args:
        x_batch: Batch of states (N, D_x).
        u_batch: Batch of controls (N, D_u).
        dt: Time step (s).
        bk: Backend for the nonlinear path; defaults to the plant's backend.

    Returns:
        Batch of next states (N, D_x).
    """
    if self._nonlinear:
        f = self.plant.dynamics(x_batch, u_batch, bk=self.bk if bk is None else bk)
        return x_batch + dt * f
    return x_batch @ self._A.T + u_batch @ self._B.T

cost_fn ¶

cost_fn(x_batch, u_batch, Q, R, x_ref: Any | None = None)

Batched quadratic stage cost.

Computes \(c(x, u) = (x - x_{ref})^T Q (x - x_{ref}) + u^T R u\) for each sample, returning a per-sample cost vector of shape (N,).

Parameters:

Name	Type	Description	Default
`x_batch`		Batch of states (N, D_x).	required
`u_batch`		Batch of controls (N, D_u).	required
`Q`		State cost matrix (D_x, D_x) or diagonal (D_x,).	required
`R`		Control cost matrix (D_u, D_u) or diagonal (D_u,).	required
`x_ref`	`Any \| None`	Optional reference state (D_x,) to track. If None, regulates to the origin.	`None`

Returns:

Type	Description
	Per-sample cost vector (N,).

Source code in src/shinro/utils/batched_adapter.py

def cost_fn(self, x_batch, u_batch, Q, R, x_ref: Any | None = None):
    """Batched quadratic stage cost.

    Computes :math:`c(x, u) = (x - x_{ref})^T Q (x - x_{ref}) + u^T R u`
    for each sample, returning a per-sample cost vector of shape ``(N,)``.

    Args:
        x_batch: Batch of states (N, D_x).
        u_batch: Batch of controls (N, D_u).
        Q: State cost matrix (D_x, D_x) or diagonal (D_x,).
        R: Control cost matrix (D_u, D_u) or diagonal (D_u,).
        x_ref: Optional reference state (D_x,) to track. If None,
            regulates to the origin.

    Returns:
        Per-sample cost vector (N,).
    """
    if x_ref is not None:
        x_err = x_batch - x_ref
    else:
        x_err = x_batch
    x_cost = self._quad_form(x_err, Q)
    u_cost = self._quad_form(u_batch, R)
    return x_cost + u_cost

`as_batch`¶

Promote a 1-D vector to a (1, n) row.

Parameters:

Name	Type	Description	Default
`bk`	`ArrayBackend`	Backend to reshape with.	required
`x`	`Any`	State vector `(n,)` or batch `(N, n)`.	required

Returns:

Type	Description
`Any`	`(x_2d, was_1d)` — the batch form and whether `x` was a single
`bool`	vector (so the derivative can be converted back).

Source code in src/shinro/utils/batching.py

def as_batch(bk: ArrayBackend, x: Any) -> tuple[Any, bool]:
    """Promote a 1-D vector to a ``(1, n)`` row.

    Args:
        bk: Backend to reshape with.
        x: State vector ``(n,)`` or batch ``(N, n)``.

    Returns:
        ``(x_2d, was_1d)`` — the batch form and whether ``x`` was a single
        vector (so the derivative can be converted back).
    """
    if getattr(x, "shape", None) is None:
        x = bk.array(x)
    if len(x.shape) == 1:
        return bk.reshape(x, (1, x.shape[0])), True
    return x, False

`as_vector`¶

Undo as_batch for a derivative.

Parameters:

Name	Type	Description	Default
`bk`	`ArrayBackend`	Backend to reshape with.	required
`f`	`Any`	Derivative batch `(N, n)`.	required
`was_1d`	`bool`	The flag returned by `as_batch`.	required

Returns:

Type	Description
`Any`	`(n,)` when the state was a single vector, else `f` unchanged.

Source code in src/shinro/utils/batching.py

def as_vector(bk: ArrayBackend, f: Any, was_1d: bool) -> Any:
    """Undo :func:`as_batch` for a derivative.

    Args:
        bk: Backend to reshape with.
        f: Derivative batch ``(N, n)``.
        was_1d: The flag returned by :func:`as_batch`.

    Returns:
        ``(n,)`` when the state was a single vector, else ``f`` unchanged.
    """
    return bk.ravel(f) if was_1d else f

`column`¶

Column j of a 2-D (N, D) batch as an (N, 1) column.

Parameters:

Name	Type	Description	Default
`bk`	`ArrayBackend`	Backend to slice with.	required
`x`	`Any`	2-D batch `(N, D)`.	required
`j`	`int`	Column index (compile-time constant).	required

Returns:

Type	Description
`Any`	Column `j`, shape `(N, 1)`.

Source code in src/shinro/utils/batching.py

def column(bk: ArrayBackend, x: Any, j: int) -> Any:
    """Column ``j`` of a 2-D ``(N, D)`` batch as an ``(N, 1)`` column.

    Args:
        bk: Backend to slice with.
        x: 2-D batch ``(N, D)``.
        j: Column index (compile-time constant).

    Returns:
        Column ``j``, shape ``(N, 1)``.
    """
    return bk.slice_(x.T, j, j + 1).T

`control_batch`¶

Normalize a control to a 2-D (N, n_u) (or (1, n_u)) row.

Accepts a scalar (n_u == 1, or the first input with the rest zero for n_u > 1 — the historical scalar-control convention), a (n_u,) vector, or an (N, n_u) batch.

Parameters:

Name	Type	Description	Default
`bk`	`ArrayBackend`	Backend to build with.	required
`control`	`Any`	Scalar, `(n_u,)`, or `(N, n_u)`.	required
`n_u`	`int`	Control dimension.	required

Returns:

Type	Description
`Any`	A 2-D control row/batch.

Source code in src/shinro/utils/batching.py

def control_batch(bk: ArrayBackend, control: Any, n_u: int) -> Any:
    """Normalize a control to a 2-D ``(N, n_u)`` (or ``(1, n_u)``) row.

    Accepts a scalar (``n_u == 1``, or the first input with the rest zero for
    ``n_u > 1`` — the historical scalar-control convention), a ``(n_u,)``
    vector, or an ``(N, n_u)`` batch.

    Args:
        bk: Backend to build with.
        control: Scalar, ``(n_u,)``, or ``(N, n_u)``.
        n_u: Control dimension.

    Returns:
        A 2-D control row/batch.
    """
    if getattr(control, "shape", None) is None:
        control = bk.array([control] + [0.0] * (n_u - 1))
    if len(control.shape) == 0:
        control = bk.reshape(bk.array([control] + [0.0] * (n_u - 1)), (n_u,))
    if len(control.shape) == 1:
        return bk.reshape(control, (1, control.shape[0]))
    return control

`resolve_config_path`¶

Resolve a TOML config path to an existing file.

Parameters:

Name	Type	Description	Default
`path`	`str`	Absolute path, or a path relative to the current working directory (e.g. `configs/controllers/lqr_base.toml` for a config in your own project).	required

Returns:

Type	Description
`Path`	A `pathlib.Path` pointing at an existing config file.

Raises:

Type	Description
`FileNotFoundError`	If the path does not exist. Resolution never falls back to a packaged shinro config — a missing or wrong-CWD path is a loud error, not a silent substitution.

Source code in src/shinro/utils/config_resolver.py

def resolve_config_path(path: str) -> Path:
    """Resolve a TOML config path to an existing file.

    Args:
        path: Absolute path, or a path relative to the current working
            directory (e.g. ``configs/controllers/lqr_base.toml`` for a
            config in your own project).

    Returns:
        A :class:`pathlib.Path` pointing at an existing config file.

    Raises:
        FileNotFoundError: If the path does not exist. Resolution never falls
            back to a packaged shinro config — a missing or wrong-CWD path is a
            loud error, not a silent substitution.
    """
    path = str(path)
    p = Path(path)
    if p.is_absolute():
        if not p.exists():
            raise FileNotFoundError(f"Config not found: {path}")
        return p
    if p.exists():
        return p
    raise FileNotFoundError(
        f"Config not found: {path} (resolved relative to CWD {Path.cwd()}; "
        f"shinro ships no configs — copy the sample you need from the repo's samples/)"
    )

`DataclassProtocol`¶

Bases: Protocol

Minimal structural type for config dataclasses.

`strict_from_dict`¶

Parse a raw TOML dict into a config dataclass, strictly.

Parameters:

Name	Type	Description	Default
`cls`	`type[T]`	The config dataclass to construct.	required
`raw`	`dict[str, Any]`	Raw TOML dict. A `type` key is validated against `type_name` and dropped; it may be absent (e.g. MCP inline params).	required
`type_name`	`str`	The registered component name, used in error messages and to validate the `type` key.	required

Returns:

Type	Description
`T`	A `cls` instance.

Raises:

Type	Description
`ValueError`	On unknown keys, a mismatched `type` key, or missing required fields.

Source code in src/shinro/utils/config_spec.py

def strict_from_dict[T: DataclassProtocol](cls: type[T], raw: dict[str, Any], type_name: str) -> T:
    """Parse a raw TOML dict into a config dataclass, strictly.

    Args:
        cls: The config dataclass to construct.
        raw: Raw TOML dict. A ``type`` key is validated against ``type_name``
            and dropped; it may be absent (e.g. MCP inline params).
        type_name: The registered component name, used in error messages and
            to validate the ``type`` key.

    Returns:
        A ``cls`` instance.

    Raises:
        ValueError: On unknown keys, a mismatched ``type`` key, or missing
            required fields.
    """
    known = set(cls.__dataclass_fields__)
    unknown = set(raw) - known - {"type"}
    if unknown:
        raise ValueError(
            f"{type_name} config: unknown key(s) {sorted(unknown)} — valid keys: {sorted(known)}"
        )
    if (t := raw.get("type")) is not None and t != type_name:
        raise ValueError(f"{type_name} config: type = {t!r} — wrong file?")
    try:
        return cls(**{k: v for k, v in raw.items() if k != "type"})  # type: ignore[call-arg]
    except TypeError as e:
        raise ValueError(f"{type_name} config: {e}") from e

`config_from_toml`¶

Load a TOML file and strictly parse it into a config dataclass.

Parameters:

Name	Type	Description	Default
`cls`	`type[T]`	The config dataclass to construct.	required
`path`	`str`	TOML path (resolved via `shinro.utils.config_resolver.resolve_config_path`).	required
`type_name`	`str`	The registered component name.	required

Returns:

Type	Description
`T`	A `cls` instance.

Source code in src/shinro/utils/config_spec.py

def config_from_toml[T: DataclassProtocol](cls: type[T], path: str, type_name: str) -> T:
    """Load a TOML file and strictly parse it into a config dataclass.

    Args:
        cls: The config dataclass to construct.
        path: TOML path (resolved via
            :func:`shinro.utils.config_resolver.resolve_config_path`).
        type_name: The registered component name.

    Returns:
        A ``cls`` instance.
    """
    with open(resolve_config_path(path), "rb") as f:
        return strict_from_dict(cls, tomllib.load(f), type_name)

`strict_from_list`¶

Strict-parse a list of TOML dicts into a list of config dataclasses.

Parameters:

Name	Type	Description	Default
`cls`	`type[T]`	The nested config dataclass to construct per entry.	required
`raw`	`list`	List of raw TOML dicts (e.g. `[[segments]]` entries).	required
`type_name`	`str`	The component name, used in error messages.	required

Returns:

Type	Description
`list[T]`	A list of `cls` instances.

Source code in src/shinro/utils/config_spec.py

def strict_from_list[T: DataclassProtocol](cls: type[T], raw: list, type_name: str) -> list[T]:
    """Strict-parse a list of TOML dicts into a list of config dataclasses.

    Args:
        cls: The nested config dataclass to construct per entry.
        raw: List of raw TOML dicts (e.g. ``[[segments]]`` entries).
        type_name: The component name, used in error messages.

    Returns:
        A list of ``cls`` instances.
    """
    return [strict_from_dict(cls, item, type_name) for item in raw]

`strip_runtime_keys`¶

Pop runtime-injected keys from a raw TOML config dict.

Sim-backed builds inject non-TOML values into plant config dicts (engine objects, joint_groups tables) before shinro.components.ConfigDriven.parse_config runs; strict parsing must not see them.

Parameters:

Name	Type	Description	Default
`raw`	`dict[str, Any]`	Raw TOML config dict.	required
`keys`	`tuple[str, ...]`	Runtime-injected key names to pop.	required

Returns:

Type	Description
`tuple[dict[str, Any], dict[str, Any]]`	Tuple of (clean dict to strict-parse, dict of popped runtime values).

Source code in src/shinro/utils/config_spec.py

def strip_runtime_keys(raw: dict[str, Any], keys: tuple[str, ...]) -> tuple[dict[str, Any], dict[str, Any]]:
    """Pop runtime-injected keys from a raw TOML config dict.

    Sim-backed builds inject non-TOML values into plant config dicts
    (``engine`` objects, ``joint_groups`` tables) before
    :meth:`~shinro.components.ConfigDriven.parse_config` runs; strict parsing
    must not see them.

    Args:
        raw: Raw TOML config dict.
        keys: Runtime-injected key names to pop.

    Returns:
        Tuple of (clean dict to strict-parse, dict of popped runtime values).
    """
    popped = {k: raw[k] for k in keys if k in raw}
    return {k: v for k, v in raw.items() if k not in keys}, popped

`BoundsConfig`¶

Nested [state_bounds] table shared by analytical plants.

`LTISystemsAnalyzer`¶

Analyze linear time-invariant state-space systems.

Provides controllability/observability checks, Gramian computations (continuous, discrete, finite-horizon), spectral diagnostics, Hankel singular values, balanced realization, and balanced truncation.

Parameters:

Name	Type	Description	Default
`A`	`MatrixLike`	State matrix (n, n).	required
`B`	`MatrixLike \| None`	Input matrix (n, m). Defaults to zeros.	`None`
`C`	`MatrixLike \| None`	Output matrix (p, n). Defaults to zeros.	`None`
`D`	`MatrixLike \| None`	Feedthrough matrix (p, m). Defaults to zeros.	`None`
`dt`	`float \| None`	Sampling time for discrete-time analysis. Defaults to None.	`None`
`backend`	`ArrayBackend \| None`	Array backend. Defaults to NumpyBackend.	`None`

Initialize the analyzer.

The state-space matrices may be given as backend arrays or as nested Python sequences; sequences are coerced to backend arrays.

Source code in src/shinro/utils/controllability_checker.py

def __init__(
    self,
    A: MatrixLike,
    B: MatrixLike | None = None,
    C: MatrixLike | None = None,
    D: MatrixLike | None = None,
    dt: float | None = None,
    backend: ArrayBackend | None = None,
) -> None:
    """Initialize the analyzer.

    The state-space matrices may be given as backend arrays or as nested
    Python sequences; sequences are coerced to backend arrays.
    """
    self.bk = backend if backend is not None else NumpyBackend()
    self.A: np.ndarray = self._coerce(A)
    self.B: np.ndarray = self._coerce(B) if B is not None else self.bk.zeros((self.A.shape[0], 0))
    self.C: np.ndarray = self._coerce(C) if C is not None else self.bk.zeros((0, self.A.shape[0]))
    self.D: np.ndarray = (
        self._coerce(D) if D is not None else self.bk.zeros((self.C.shape[0], self.B.shape[1]))
    )
    self.dt: float | None = dt
    self._cached_values = {}
    self._validate_dimensions()

controllabilty ¶

controllabilty()

Build the controllability matrix C = [B, AB, A²B, ..., A^{n-1}B].

Returns:

Type	Description
	Controllability matrix (n, n*m).

Source code in src/shinro/utils/controllability_checker.py

def controllabilty(self):
    """Build the controllability matrix C = [B, AB, A²B, ..., A^{n-1}B].

    Returns:
        Controllability matrix (n, n*m).
    """
    n = self.A.shape[0]
    cols = [self.B]
    for i in range(1, n):
        cols.append(self.A @ cols[-1])
    C = self.bk.hstack(cols)
    return C

observability ¶

observability()

Build the observability matrix O = [C; CA; CA²; ...; CA^{n-1}].

Returns:

Type	Description
	Observability matrix (n*p, n).

Source code in src/shinro/utils/controllability_checker.py

def observability(self):
    """Build the observability matrix O = [C; CA; CA²; ...; CA^{n-1}].

    Returns:
        Observability matrix (n*p, n).
    """
    n = self.A.shape[0]
    cols = [self.C]
    for i in range(1, n):
        cols.append(self.C @ self.bk.matrix_power(self.A, i))
    O_mat = self.bk.vstack(cols)
    return O_mat

is_controllable ¶

is_controllable()

Check whether the system is controllable via Kalman rank test.

Returns:

Type	Description
	True if the controllability matrix has full row rank.

Source code in src/shinro/utils/controllability_checker.py

def is_controllable(self):
    """Check whether the system is controllable via Kalman rank test.

    Returns:
        True if the controllability matrix has full row rank.
    """
    C = self.controllabilty()
    rank = self.bk.matrix_rank(C)
    return rank == self.A.shape[0]

is_observable ¶

is_observable()

Check whether the system is observable via Kalman rank test.

Returns:

Type	Description
	True if the observability matrix has full column rank.

Source code in src/shinro/utils/controllability_checker.py

def is_observable(self):
    """Check whether the system is observable via Kalman rank test.

    Returns:
        True if the observability matrix has full column rank.
    """
    O_mat = self.observability()
    rank = self.bk.matrix_rank(O_mat)
    return rank == self.A.shape[0]

controllability_gramian ¶

controllability_gramian() -> ndarray

Infinite-horizon controllability Gramian (continuous-time).

Returns cached result unless the system has changed.

Returns:

Type	Description
`ndarray`	Controllability Gramian Wc (n, n).

Raises:

Type	Description
`ValueError`	If A is not Hurwitz (eigenvalues with non-negative real part).

Source code in src/shinro/utils/controllability_checker.py

def controllability_gramian(self) -> np.ndarray:
    """Infinite-horizon controllability Gramian (continuous-time).

    Returns cached result unless the system has changed.

    Returns:
        Controllability Gramian Wc (n, n).

    Raises:
        ValueError: If A is not Hurwitz (eigenvalues with non-negative real part).
    """
    if "Wc" in self._cached_values:
        return self._cached_values["Wc"]

    if not self._is_hurwitz():
        raise ValueError(
            "A is not Hurwitz; infinite-horizon controllability Gramian does not exist. "
            "Use controllability_gramian_finite(T) instead."
        )
    Q = self.B @ self.B.T
    Wc = self._solve_continuous_lyap(Q)
    self._cached_values["Wc"] = Wc
    return Wc

observability_gramian ¶

observability_gramian() -> ndarray

Infinite-horizon observability Gramian (continuous-time).

Returns cached result unless the system has changed.

Returns:

Type	Description
`ndarray`	Observability Gramian Wo (n, n).

Raises:

Type	Description
`ValueError`	If A is not Hurwitz (eigenvalues with non-negative real part).

Source code in src/shinro/utils/controllability_checker.py

def observability_gramian(self) -> np.ndarray:
    """Infinite-horizon observability Gramian (continuous-time).

    Returns cached result unless the system has changed.

    Returns:
        Observability Gramian Wo (n, n).

    Raises:
        ValueError: If A is not Hurwitz (eigenvalues with non-negative real part).
    """
    if "Wo" in self._cached_values:
        return self._cached_values["Wo"]

    if not self._is_hurwitz():
        raise ValueError(
            "A is not Hurwitz; infinite-horizon observability Gramian does not exist. "
            "Use observability_gramian_finite(T) instead."
        )
    Q = self.C.T @ self.C
    # Observability Gramian: A^T Wo + Wo A + C^T C = 0 — solve the
    # controllability-form Lyapunov equation for the adjoint A^T.
    Wo = self._solve_continuous_lyap(Q, self.A.T)
    self._cached_values["Wo"] = Wo
    return Wo

discrete_controllability_gramian ¶

discrete_controllability_gramian() -> ndarray

Discrete-time infinite-horizon controllability Gramian.

Solves A Wc A^T - Wc + B B^T = 0.

Returns:

Type	Description
`ndarray`	Discrete controllability Gramian Wc (n, n).

Source code in src/shinro/utils/controllability_checker.py

def discrete_controllability_gramian(self) -> np.ndarray:
    """Discrete-time infinite-horizon controllability Gramian.

    Solves A Wc A^T - Wc + B B^T = 0.

    Returns:
        Discrete controllability Gramian Wc (n, n).
    """
    if "Wc_discrete" in self._cached_values:
        return self._cached_values["Wc_discrete"]

    M = self.B @ self.B.T
    Wc_discrete = self._solve_discrete_lyap(self.A, M)
    self._cached_values["Wc_discrete"] = Wc_discrete
    return Wc_discrete

discrete_observability_gramian ¶

discrete_observability_gramian() -> ndarray

Discrete-time infinite-horizon observability Gramian.

Solves A^T Wo A - Wo + C^T C = 0.

Returns:

Type	Description
`ndarray`	Discrete observability Gramian Wo (n, n).

Source code in src/shinro/utils/controllability_checker.py

def discrete_observability_gramian(self) -> np.ndarray:
    """Discrete-time infinite-horizon observability Gramian.

    Solves A^T Wo A - Wo + C^T C = 0.

    Returns:
        Discrete observability Gramian Wo (n, n).
    """
    if "Wo_discrete" in self._cached_values:
        return self._cached_values["Wo_discrete"]

    M = self.C.T @ self.C
    Wo_discrete = self._solve_discrete_lyap(self.A.T, M)
    self._cached_values["Wo_discrete"] = Wo_discrete
    return Wo_discrete

controllability_gramian_finite ¶

controllability_gramian_finite(T: float) -> ndarray

Finite-horizon controllability Gramian: Wc(T) = ∫₀ᵀ e^{Aτ} B B^T e^{A^Tτ} dτ.

Works for any A (stable or unstable).

Parameters:

Name	Type	Description	Default
`T`	`float`	Horizon length.	required

Returns:

Type	Description
`ndarray`	Finite-horizon controllability Gramian (n, n).

Raises:

Type	Description
`ValueError`	If T is not positive.

Source code in src/shinro/utils/controllability_checker.py

def controllability_gramian_finite(self, T: float) -> np.ndarray:
    """Finite-horizon controllability Gramian: Wc(T) = ∫₀ᵀ e^{Aτ} B B^T e^{A^Tτ} dτ.

    Works for any A (stable or unstable).

    Args:
        T: Horizon length.

    Returns:
        Finite-horizon controllability Gramian (n, n).

    Raises:
        ValueError: If T is not positive.
    """
    if T <= 0:
        raise ValueError("T (horizon) must be positive.")
    Q = self.B @ self.B.T
    return self._finite_horizon_gramian(Q, T)

observability_gramian_finite ¶

observability_gramian_finite(T: float) -> ndarray

Finite-horizon observability Gramian: Wo(T) = ∫₀ᵀ e^{A^Tτ} C^T C e^{Aτ} dτ.

Works for any A (stable or unstable).

Parameters:

Name	Type	Description	Default
`T`	`float`	Horizon length.	required

Returns:

Type	Description
`ndarray`	Finite-horizon observability Gramian (n, n).

Raises:

Type	Description
`ValueError`	If T is not positive.

Source code in src/shinro/utils/controllability_checker.py

def observability_gramian_finite(self, T: float) -> np.ndarray:
    """Finite-horizon observability Gramian: Wo(T) = ∫₀ᵀ e^{A^Tτ} C^T C e^{Aτ} dτ.

    Works for any A (stable or unstable).

    Args:
        T: Horizon length.

    Returns:
        Finite-horizon observability Gramian (n, n).

    Raises:
        ValueError: If T is not positive.
    """
    if T <= 0:
        raise ValueError("T (horizon) must be positive.")
    Q = self.C.T @ self.C
    # Wo(T) = ∫ e^{A^T τ} C^T C e^{A τ} dτ — integrate the adjoint A^T.
    return self._finite_horizon_gramian(Q, T, A_matrix=self.A.T)

gramian_spectrum ¶

gramian_spectrum(gramian: str = 'Wc') -> ndarray

Return eigenvalues of a chosen Gramian.

Parameters:

Name	Type	Description	Default
`gramian`	`str`	Which Gramian to inspect. One of "Wc", "Wo", "Wc_finite:", "Wo_finite:". Defaults to "Wc".	`'Wc'`

Returns:

Type	Description
`ndarray`	Real parts of Gramian eigenvalues (n,).

Source code in src/shinro/utils/controllability_checker.py

def gramian_spectrum(self, gramian: str = "Wc") -> np.ndarray:
    """Return eigenvalues of a chosen Gramian.

    Args:
        gramian: Which Gramian to inspect. One of "Wc", "Wo",
            "Wc_finite:<T>", "Wo_finite:<T>". Defaults to "Wc".

    Returns:
        Real parts of Gramian eigenvalues (n,).
    """
    if gramian == "Wc":
        G = self.controllability_gramian()
    elif gramian == "Wo":
        G = self.observability_gramian()
    elif gramian.startswith("Wc_"):
        _, horizon = gramian.split(":")
        G = self.controllability_gramian_finite(float(horizon))
    elif gramian.startswith("Wo_"):
        _, horizon = gramian.split(":")
        G = self.observability_gramian_finite(float(horizon))
    else:
        raise ValueError(f"Unknown gramian identifier: {gramian}")

    return self.bk.real(self.bk.eigvals(G))

gramian_condition ¶

gramian_condition(gramian: str = 'Wc') -> float

Return the 2-norm condition number of the selected Gramian.

Parameters:

Name	Type	Description	Default
`gramian`	`str`	Which Gramian to inspect. One of "Wc", "Wo". Defaults to "Wc".	`'Wc'`

Returns:

Type	Description
`float`	Condition number, or inf if the Gramian is singular.

Source code in src/shinro/utils/controllability_checker.py

def gramian_condition(self, gramian: str = "Wc") -> float:
    """Return the 2-norm condition number of the selected Gramian.

    Args:
        gramian: Which Gramian to inspect. One of "Wc", "Wo". Defaults to "Wc".

    Returns:
        Condition number, or inf if the Gramian is singular.
    """
    if gramian == "Wc":
        G = self.controllability_gramian()
    elif gramian == "Wo":
        G = self.observability_gramian()
    else:
        raise ValueError("Only infinite-horizon 'Wc' / 'Wo' supported for cond().")
    rank = self.bk.matrix_rank(G)
    if rank < G.shape[0]:
        return np.inf
    return self.bk.cond(G)

hankel_singular_values ¶

hankel_singular_values() -> ndarray

Continuous-time Hankel singular values σ_i = sqrt(λ_i(Wc Wo)).

Returns:

Type	Description
`ndarray`	Hankel singular values sorted in descending order (n,).

Source code in src/shinro/utils/controllability_checker.py

def hankel_singular_values(self) -> np.ndarray:
    """Continuous-time Hankel singular values σ_i = sqrt(λ_i(Wc Wo)).

    Returns:
        Hankel singular values sorted in descending order (n,).
    """
    return self._hankel_from_gramians(
        self.controllability_gramian(), self.observability_gramian()
    )

discrete_hankel_singular_values ¶

discrete_hankel_singular_values() -> ndarray

Discrete-time Hankel singular values from the discrete-time Gramians.

The continuous Gramians (and hence hankel_singular_values) do not exist for a Schur-stable discrete system, so this is the valid HSV path for a discrete system.

Returns:

Type	Description
`ndarray`	Hankel singular values sorted in descending order (n,).

Raises:

Type	Description
`ValueError`	If `dt` is None or \|eig(A)\| >= 1.

Source code in src/shinro/utils/controllability_checker.py

def discrete_hankel_singular_values(self) -> np.ndarray:
    """Discrete-time Hankel singular values from the discrete-time Gramians.

    The continuous Gramians (and hence :meth:`hankel_singular_values`) do
    not exist for a Schur-stable discrete system, so this is the valid HSV
    path for a discrete system.

    Returns:
        Hankel singular values sorted in descending order (n,).

    Raises:
        ValueError: If ``dt`` is None or |eig(A)| >= 1.
    """
    return self._hankel_from_gramians(
        self.discrete_controllability_gramian(),
        self.discrete_observability_gramian(),
    )

balanced_realization ¶

balanced_realization() -> tuple[ndarray, ndarray, ndarray]

Return the balanced state-space matrices (Abal, Bbal, Cbal).

The transformation T satisfies

T^{-1} A T = Abal, T^{-1} B = Bbal, C T = Cbal,

and the balanced Gramians are diag(σ_1, ..., σ_n).

Returns:

Type	Description
`tuple[ndarray, ndarray, ndarray]`	Tuple of (Abal, Bbal, Cbal) with shapes (n, n), (n, m), (p, n).

Source code in src/shinro/utils/controllability_checker.py

def balanced_realization(self) -> tuple[np.ndarray, np.ndarray, np.ndarray]:
    """Return the balanced state-space matrices (Abal, Bbal, Cbal).

    The transformation T satisfies:
        T^{-1} A T = Abal,
        T^{-1} B   = Bbal,
        C T        = Cbal,
    and the balanced Gramians are diag(σ_1, ..., σ_n).

    Returns:
        Tuple of (Abal, Bbal, Cbal) with shapes (n, n), (n, m), (p, n).
    """
    self.hankel_singular_values()

    Wc = self.controllability_gramian()
    Wo = self.observability_gramian()
    Lc = self.bk.cholesky(Wc)
    Lo = self.bk.cholesky(Wo)

    U, s, Vh = self.bk.svd(Lo.T @ Lc)
    T = Lc @ Vh.T @ self.bk.diag(1.0 / self.bk.sqrt(s))
    Tinv = self.bk.diag(1.0 / self.bk.sqrt(s)) @ U.T @ Lo.T

    Ab = Tinv @ self.A @ T
    Bb = Tinv @ self.B
    Cb = self.C @ T

    return Ab, Bb, Cb

balanced_truncate ¶

balanced_truncate(r: int) -> tuple[ndarray, ndarray, ndarray, ndarray]

Perform balanced truncation to obtain an order-r reduced model.

Parameters:

Name	Type	Description	Default
`r`	`int`	Desired reduced order (0 < r <= n).	required

Returns:

Type	Description
`tuple[ndarray, ndarray, ndarray, ndarray]`	Tuple of (Ar, Br, Cr, Dr) with shapes (r, r), (r, m), (p, r), (p, m).

Source code in src/shinro/utils/controllability_checker.py

def balanced_truncate(self, r: int) -> tuple[np.ndarray, np.ndarray, np.ndarray, np.ndarray]:
    """Perform balanced truncation to obtain an order-r reduced model.

    Args:
        r: Desired reduced order (0 < r <= n).

    Returns:
        Tuple of (Ar, Br, Cr, Dr) with shapes (r, r), (r, m), (p, r), (p, m).
    """
    n = self.A.shape[0]
    if not (0 < r <= n):
        raise ValueError("Reduced order r must satisfy 0 < r <= n.")

    Ab, Bb, Cb = self.balanced_realization()
    sigma = self.hankel_singular_values()

    Ar = Ab[:r, :r]
    Br = Bb[:r, :]
    Cr = Cb[:, :r]
    Dr = self.D

    error_bound = 2.0 * self.bk.sum(sigma[r:])
    self._cached_values["balanced_trunc_error_bound"] = error_bound

    return Ar, Br, Cr, Dr

reset_cache ¶

reset_cache() -> None

Clear all memoised results after manually changing A, B, or C.

Source code in src/shinro/utils/controllability_checker.py

def reset_cache(self) -> None:
    """Clear all memoised results after manually changing A, B, or C."""
    self._cached_values.clear()

rank_report ¶

rank_report() -> dict[str, tuple[int, float]]

Return rank and condition of controllability and observability matrices.

Returns:

Type	Description
`dict[str, tuple[int, float]]`	Dict with keys 'controllability' and 'observability', each a
`dict[str, tuple[int, float]]`	tuple of (rank, condition_number).

Source code in src/shinro/utils/controllability_checker.py

def rank_report(self) -> dict[str, tuple[int, float]]:
    """Return rank and condition of controllability and observability matrices.

    Returns:
        Dict with keys 'controllability' and 'observability', each a
        tuple of (rank, condition_number).
    """
    n = self.A.shape[0]
    C_mat = self.controllabilty()
    O_mat = self.observability()
    rank_c = self.bk.matrix_rank(C_mat)
    rank_o = self.bk.matrix_rank(O_mat)
    # Full rank means rank == n for both matrices (C_mat is n x N*m and
    # O_mat is N*p x n).  Gating on n also avoids cond() on an empty
    # matrix when B or C has a zero dimension.
    cond_c = self.bk.cond(C_mat) if rank_c == n else np.inf
    cond_o = self.bk.cond(O_mat) if rank_o == n else np.inf
    return {
        "controllability": (int(rank_c), float(cond_c)),
        "observability": (int(rank_o), float(cond_o)),
    }

summary ¶

summary() -> str

Produce a human-readable report of system properties.

Includes rank/condition, Gramian eigenvalues, Hankel singular values, and a balanced-truncation error bound if previously computed.

Returns:

Type	Description
`str`	Formatted summary string.

Source code in src/shinro/utils/controllability_checker.py

def summary(self) -> str:
    """Produce a human-readable report of system properties.

    Includes rank/condition, Gramian eigenvalues, Hankel singular values,
    and a balanced-truncation error bound if previously computed.

    Returns:
        Formatted summary string.
    """
    n = self.A.shape[0]
    rank_info = self.rank_report()
    stable = self._is_hurwitz()
    wc = self.controllability_gramian() if stable else None
    wo = self.observability_gramian() if stable else None

    lines = [
        f"System order n = {n}",
        f"Controllable?      {self.is_controllable()}",
        f"Observable?       {self.is_observable()}",
        "",
        "Kalman rank / condition:",
        f"  Controllability : rank = {rank_info['controllability'][0]},  cond = {rank_info['controllability'][1]:.2e}",
        f"  Observability   : rank = {rank_info['observability'][0]},    cond = {rank_info['observability'][1]:.2e}",
        "",
    ]

    if wc is not None:
        eig_wc = self.bk.real(self.bk.eigvals(wc))
        lines.append("Controllability Gramian eigenvalues (sorted):")
        lines.append("  " + ", ".join(f"{ev:.3e}" for ev in self.bk.to_numpy(self.bk.sort(eig_wc))[::-1]))
    else:
        lines.append("Controllability Gramian: *not defined* (unstable A).")

    if wo is not None:
        eig_wo = self.bk.real(self.bk.eigvals(wo))
        lines.append("Observability Gramian eigenvalues (sorted):")
        lines.append("  " + ", ".join(f"{ev:.3e}" for ev in self.bk.to_numpy(self.bk.sort(eig_wo))[::-1]))
    else:
        lines.append("Observability Gramian: *not defined* (unstable A).")

    if wc is not None and wo is not None:
        sigma = self.hankel_singular_values()
        lines.append("")
        lines.append("Hankel singular values (σ1 >= σ2 ...):")
        lines.append("  " + ", ".join(f"{sv:.3e}" for sv in self.bk.to_numpy(sigma)))
    else:
        lines.append("")
        lines.append("Hankel singular values: *not defined* (unstable A).")

    err = self._cached_values.get("balanced_trunc_error_bound")
    if err is not None:
        lines.append("")
        lines.append(f"Balanced-truncation error bound (2*sum(σ_{{r+1..n}})) = {err:.3e}")

    return "\n".join(lines)

`linearize`¶

First-order Taylor expansion of f(x, u) around (x0, u0).

Computes the Jacobians A = ∂f/∂x and B = ∂f/∂u at the operating point (x0, u0) using central finite differences. The user's dynamics function always receives and returns numpy arrays regardless of the backend.

Parameters:

Name	Type	Description	Default
`f`	`Callable[[Any, Any], Any]`	Continuous-time dynamics `f(x, u) -> dx/dt` where x is (n_x,) and u is (n_u,), returns (n_x,).	required
`x0`	`Any`	Operating point state, shape (n_x,).	required
`u0`	`Any`	Operating point input, shape (n_u,).	required
`backend`	`ArrayBackend \| None`	Array backend. Defaults to NumpyBackend.	`None`
`eps`	`float`	Step size for finite differences.	`1e-06`

Returns:

Type	Description
`Any`	Tuple (A, B) where A = ∂f/∂x has shape (n_x, n_x) and
`Any`	B = ∂f/∂u has shape (n_x, n_u), in the backend's native type.

Source code in src/shinro/utils/linearization.py

def linearize(
    f: Callable[[Any, Any], Any],
    x0: Any,
    u0: Any,
    backend: ArrayBackend | None = None,
    eps: float = 1e-6,
) -> tuple[Any, Any]:
    """First-order Taylor expansion of f(x, u) around (x0, u0).

    Computes the Jacobians A = ∂f/∂x and B = ∂f/∂u at the operating point
    (x0, u0) using central finite differences. The user's dynamics function
    always receives and returns numpy arrays regardless of the backend.

    Args:
        f: Continuous-time dynamics ``f(x, u) -> dx/dt`` where x is (n_x,)
            and u is (n_u,), returns (n_x,).
        x0: Operating point state, shape (n_x,).
        u0: Operating point input, shape (n_u,).
        backend: Array backend. Defaults to NumpyBackend.
        eps: Step size for finite differences.

    Returns:
        Tuple (A, B) where A = ∂f/∂x has shape (n_x, n_x) and
        B = ∂f/∂u has shape (n_x, n_u), in the backend's native type.
    """
    bk = backend or NumpyBackend()
    x0_np = np.asarray(bk.to_numpy(x0) if hasattr(bk, 'to_numpy') else x0, dtype=np.float64)
    u0_np = np.asarray(bk.to_numpy(u0) if hasattr(bk, 'to_numpy') else u0, dtype=np.float64)

    def f_x(x):
        return np.asarray(f(x, u0_np), dtype=np.float64)

    def f_u(u):
        return np.asarray(f(x0_np, u), dtype=np.float64)

    n = x0_np.shape[0]
    m = f_x(x0_np).shape[0]
    A = np.zeros((m, n), dtype=np.float64)
    for i in range(n):
        h = np.zeros(n, dtype=np.float64)
        h[i] = eps
        A[:, i] = (f_x(x0_np + h) - f_x(x0_np - h)) / (2.0 * eps)

    r = u0_np.shape[0]
    B = np.zeros((m, r), dtype=np.float64)
    for i in range(r):
        h = np.zeros(r, dtype=np.float64)
        h[i] = eps
        B[:, i] = (f_u(u0_np + h) - f_u(u0_np - h)) / (2.0 * eps)

    return bk.from_numpy(A), bk.from_numpy(B)

`as_numpy_f`¶

Wrap a backend-bound f(x, u) into a numpy-in/numpy-out callable.

linearize requires f to take and return numpy arrays regardless of the backend. This bridges a backend-native callable such as plant.dynamics(x_b, u_b) -> dx_b into that contract via backend.from_numpy / backend.to_numpy.

Parameters:

Name	Type	Description	Default
`f`		Backend-bound callable `f(x, u) -> y` operating on backend arrays.	required
`backend`		ArrayBackend whose `from_numpy`/`to_numpy` do the bridging.	required

Returns:

Type	Description
	`f_np(x_np, u_np) -> y_np` operating on float64 numpy arrays.

Source code in src/shinro/utils/linearization.py

def as_numpy_f(f, backend):
    """Wrap a backend-bound f(x, u) into a numpy-in/numpy-out callable.

    :func:`linearize` requires ``f`` to take and return numpy arrays
    regardless of the backend. This bridges a backend-native callable such
    as ``plant.dynamics(x_b, u_b) -> dx_b`` into that contract via
    ``backend.from_numpy`` / ``backend.to_numpy``.

    Args:
        f: Backend-bound callable ``f(x, u) -> y`` operating on backend arrays.
        backend: ArrayBackend whose ``from_numpy``/``to_numpy`` do the bridging.

    Returns:
        ``f_np(x_np, u_np) -> y_np`` operating on float64 numpy arrays.
    """
    def f_np(x, u):
        x_b = backend.from_numpy(x)
        u_b = backend.from_numpy(u)
        return np.asarray(backend.to_numpy(f(x_b, u_b)), dtype=np.float64)
    return f_np

`discretize_euler`¶

First-order Euler discretization of a continuous-time linear model.

Computes the discrete-time matrices \(A_d = I + dt \cdot A_c\) and \(B_d = dt \cdot B_c\), matching the semi-implicit Euler integration used by the analytical plants. At small dt the discretization error is negligible, and unlike scipy.linalg.expm it is backend-agnostic (the Kalman filter stays torch-capable).

Parameters:

Name	Type	Description	Default
`A_c`		Continuous-time state matrix (n_x, n_x).	required
`B_c`		Continuous-time input matrix (n_x, n_u).	required
`dt`		Time step in seconds.	required
`backend`	`ArrayBackend \| None`	Array backend. Defaults to NumpyBackend.	`None`

Returns:

Type	Description
	Tuple (A_d, B_d) in the backend's native type.

Source code in src/shinro/utils/linearization.py

def discretize_euler(A_c, B_c, dt, backend: ArrayBackend | None = None):
    """First-order Euler discretization of a continuous-time linear model.

    Computes the discrete-time matrices :math:`A_d = I + dt \\cdot A_c` and
    :math:`B_d = dt \\cdot B_c`, matching the semi-implicit Euler integration
    used by the analytical plants. At small ``dt`` the discretization error is
    negligible, and unlike ``scipy.linalg.expm`` it is backend-agnostic (the
    Kalman filter stays torch-capable).

    Args:
        A_c: Continuous-time state matrix (n_x, n_x).
        B_c: Continuous-time input matrix (n_x, n_u).
        dt: Time step in seconds.
        backend: Array backend. Defaults to NumpyBackend.

    Returns:
        Tuple (A_d, B_d) in the backend's native type.
    """
    bk = backend or NumpyBackend()
    n = A_c.shape[0]
    A_d = bk.eye(n) + dt * A_c
    B_d = dt * B_c
    return A_d, B_d

`linearize_plant`¶

Linearize a Plant's dynamics around an operating point.

Resolves (x0, u0) defaults, bridges plant.dynamics through as_numpy_f, and delegates to linearize.

State default: x0 = plant.bk.zeros(len(plant.get_state())). Control default: u0 = plant.bk.zeros(plant.input_dim) when the plant declares input_dim; otherwise raises ValueError unless u0 is passed explicitly, so multi-input plants cannot silently get a wrong-dim u0.

Parameters:

Name	Description	Default
`plant`	Plant with `.dynamics`, `.bk`, `.get_state()`, and optionally `.input_dim`.	required
`x0`	Operating point state. Defaults to zeros(state_dim).	`None`
`u0`	Operating point input. Defaults to zeros(plant.input_dim), or required when `plant.input_dim` is None.	`None`
`eps`	Finite-difference step size.	`1e-06`

Returns:

Type	Description
	Tuple (A, B) in the plant's backend native type.

Source code in src/shinro/utils/linearization.py

def linearize_plant(plant, x0=None, u0=None, eps=1e-6):
    """Linearize a Plant's dynamics around an operating point.

    Resolves ``(x0, u0)`` defaults, bridges ``plant.dynamics`` through
    :func:`as_numpy_f`, and delegates to :func:`linearize`.

    State default: ``x0 = plant.bk.zeros(len(plant.get_state()))``.
    Control default: ``u0 = plant.bk.zeros(plant.input_dim)`` when the plant
    declares ``input_dim``; otherwise raises ``ValueError`` unless ``u0`` is
    passed explicitly, so multi-input plants cannot silently get a wrong-dim
    ``u0``.

    Args:
        plant: Plant with ``.dynamics``, ``.bk``, ``.get_state()``, and
            optionally ``.input_dim``.
        x0: Operating point state. Defaults to zeros(state_dim).
        u0: Operating point input. Defaults to zeros(plant.input_dim), or
            required when ``plant.input_dim`` is None.
        eps: Finite-difference step size.

    Returns:
        Tuple (A, B) in the plant's backend native type.
    """
    if x0 is None:
        x0 = plant.bk.zeros(len(plant.get_state()))
    if u0 is None:
        if plant.input_dim is None:
            raise ValueError(
                "Cannot infer control dimension: plant.input_dim is not "
                "set and u0 was not passed. Set self.input_dim on the "
                "plant, or pass u0 explicitly."
            )
        u0 = plant.bk.zeros(plant.input_dim)
    return linearize(as_numpy_f(plant.dynamics, plant.bk), x0, u0, plant.bk, eps=eps)

`derive_model`¶

Return the plant's discrete-time (A_d, B_d) model.

Delegates to plant.get_model(), which every plant implements as the discrete-time model (analytical plants linearize + Euler-discretize at their dt; velocity-commanded plants return A = I, B = dt·I directly). This is the single source of the derived model shared by the simulation and compile paths.

Parameters:

Name	Type	Description	Default
`plant`		A plant exposing `get_model()`.	required

Returns:

Type	Description
	Tuple `(A_d, B_d)` in the plant's backend native type.

Source code in src/shinro/utils/linearization.py

def derive_model(plant):
    """Return the plant's discrete-time ``(A_d, B_d)`` model.

    Delegates to ``plant.get_model()``, which every plant implements as the
    discrete-time model (analytical plants linearize + Euler-discretize at
    their ``dt``; velocity-commanded plants return ``A = I, B = dt·I``
    directly). This is the single source of the derived model shared by the
    simulation and compile paths.

    Args:
        plant: A plant exposing ``get_model()``.

    Returns:
        Tuple ``(A_d, B_d)`` in the plant's backend native type.
    """
    return plant.get_model()

`inject_model`¶

Fill A_dynamics/B_dynamics from the plant when a config omits them.

cfg may be a config dict or a TOML path (resolved via shinro.utils.config_resolver.resolve_config_path); a dict is returned either way. A config that already declares either matrix is left untouched — explicit model wins over the derived one.

Parameters:

Name	Type	Description	Default
`cfg`		Controller/estimator config dict or TOML path.	required
`plant`		The plant to derive the model from.	required

Returns:

Type	Description
`dict`	The config dict with `A_dynamics`/`B_dynamics` filled in when they
`dict`	were absent.

Source code in src/shinro/utils/linearization.py

def inject_model(cfg, plant) -> dict:
    """Fill ``A_dynamics``/``B_dynamics`` from the plant when a config omits them.

    ``cfg`` may be a config dict or a TOML path (resolved via
    :func:`shinro.utils.config_resolver.resolve_config_path`); a dict is
    returned either way. A config that already declares either matrix is left
    untouched — explicit model wins over the derived one.

    Args:
        cfg: Controller/estimator config dict or TOML path.
        plant: The plant to derive the model from.

    Returns:
        The config dict with ``A_dynamics``/``B_dynamics`` filled in when they
        were absent.
    """
    if isinstance(cfg, str):
        with open(resolve_config_path(cfg), "rb") as f:
            cfg = tomllib.load(f)
    if "A_dynamics" not in cfg and "B_dynamics" not in cfg:
        A_d, B_d = derive_model(plant)
        cfg = {**cfg, "A_dynamics": A_d, "B_dynamics": B_d}
    return cfg

`inject_plant_derived`¶

Fill dt / A_dynamics / B_dynamics on a Config dataclass from the plant.

The plant is the single source of physics truth. Precedence: explicit wins, derived fills, disagreement is loud.

dt: filled from plant.dt when the config omits it; a declared dt that disagrees with the plant's is a loud error (a mismatched PID/MPPI dt silently mis-scales integration otherwise).
A_dynamics/B_dynamics: derived via derive_model and injected (as TOML-serializable lists) when the config declares neither — only when with_model is set, so sim-backed velocity-commanded plants keep their untouched A = I, B = dt·I defaults.

Parameters:

Name	Type	Description	Default
`cfg`		A component Config dataclass instance (from `shinro.components.ConfigDriven.parse_config`).	required
`plant`		The plant to derive from.	required
`with_model`	`bool`	Also derive the model when A/B are both absent.	`False`

Returns:

Type	Description
	The (possibly replaced) Config dataclass.

Raises:

Type	Description
`ValueError`	On a `dt` disagreement, or a `dt`-less config without `B_dynamics` in standalone use.

Source code in src/shinro/utils/linearization.py

def inject_plant_derived(cfg, plant, *, with_model: bool = False):
    """Fill ``dt`` / ``A_dynamics`` / ``B_dynamics`` on a Config dataclass from the plant.

    The plant is the single source of physics truth. Precedence: **explicit
    wins, derived fills, disagreement is loud**.

    - ``dt``: filled from ``plant.dt`` when the config omits it; a declared
      ``dt`` that disagrees with the plant's is a loud error (a mismatched
      PID/MPPI ``dt`` silently mis-scales integration otherwise).
    - ``A_dynamics``/``B_dynamics``: derived via :func:`derive_model` and
      injected (as TOML-serializable lists) when the config declares *neither*
      — only when ``with_model`` is set, so sim-backed velocity-commanded
      plants keep their untouched ``A = I, B = dt·I`` defaults.

    Args:
        cfg: A component Config dataclass instance (from
            :meth:`shinro.components.ConfigDriven.parse_config`).
        plant: The plant to derive from.
        with_model: Also derive the model when A/B are both absent.

    Returns:
        The (possibly replaced) Config dataclass.

    Raises:
        ValueError: On a ``dt`` disagreement, or a ``dt``-less config without
            ``B_dynamics`` in standalone use.
    """
    names = {f.name for f in fields(cfg)}
    if "dt" in names:
        if cfg.dt is None:
            cfg = replace(cfg, dt=float(plant.dt))
        elif abs(float(cfg.dt) - float(plant.dt)) > 1e-12:
            raise ValueError(
                f"{type(cfg).__name__}: config dt ({cfg.dt}) disagrees with plant dt "
                f"({plant.dt}) — the plant is the source of truth; omit dt to inherit it."
            )
    if with_model and "A_dynamics" in names and cfg.A_dynamics is None and cfg.B_dynamics is None:
        A_d, B_d = derive_model(plant)
        cfg = replace(cfg, A_dynamics=A_d.tolist(), B_dynamics=B_d.tolist())
    return cfg

`PluginImportError`¶

Bases: Exception

A module named by --import could not be imported.

`import_modules`¶

Import each dotted module name so its @register_* components register.

Parameters:

Name	Type	Description	Default
`names`	`list[str]`	Dotted module names, e.g. `"my_pkg.components"`.	required

Raises:

Type	Description
`PluginImportError`	If a module cannot be imported. The message names the module and the underlying cause, so a typo is a clean CLI error instead of a traceback.

Source code in src/shinro/utils/plugin_loader.py

def import_modules(names: list[str]) -> None:
    """Import each dotted module name so its ``@register_*`` components register.

    Args:
        names: Dotted module names, e.g. ``"my_pkg.components"``.

    Raises:
        PluginImportError: If a module cannot be imported. The message names the
            module and the underlying cause, so a typo is a clean CLI error
            instead of a traceback.
    """
    for name in names:
        try:
            importlib.import_module(name)
        except Exception as e:  # surface any import-time failure (not just ImportError)
            raise PluginImportError(f"cannot import {name!r}: {type(e).__name__}: {e}") from e

Utilities¶

parse_matrix¶

ArrayBackend¶

emit_named_output ¶

solve_qp abstractmethod ¶

jacobian abstractmethod ¶

NumpyBackend¶

solve_qp ¶

jacobian ¶

TorchBackend¶

solve_qp ¶

jacobian ¶

BatchedDynamicsAdapter¶

state_dim property ¶

control_dim property ¶

dynamics_path property ¶

dynamics_fn ¶

cost_fn ¶

as_batch¶

as_vector¶

column¶

control_batch¶

resolve_config_path¶

DataclassProtocol¶

strict_from_dict¶

config_from_toml¶

strict_from_list¶

strip_runtime_keys¶

BoundsConfig¶

LTISystemsAnalyzer¶

controllabilty ¶

observability ¶

is_controllable ¶

is_observable ¶

controllability_gramian ¶

observability_gramian ¶

discrete_controllability_gramian ¶

discrete_observability_gramian ¶

controllability_gramian_finite ¶

observability_gramian_finite ¶

gramian_spectrum ¶

gramian_condition ¶

hankel_singular_values ¶

discrete_hankel_singular_values ¶

balanced_realization ¶

balanced_truncate ¶

reset_cache ¶

rank_report ¶

summary ¶

linearize¶

as_numpy_f¶

discretize_euler¶

linearize_plant¶

derive_model¶

inject_model¶

inject_plant_derived¶

PluginImportError¶

import_modules¶

`parse_matrix`¶

`ArrayBackend`¶

solve_qp `abstractmethod` ¶

jacobian `abstractmethod` ¶

`NumpyBackend`¶

`TorchBackend`¶

`BatchedDynamicsAdapter`¶

state_dim `property` ¶

control_dim `property` ¶

dynamics_path `property` ¶

`as_batch`¶

`as_vector`¶

`column`¶

`control_batch`¶

`resolve_config_path`¶

`DataclassProtocol`¶

`strict_from_dict`¶

`config_from_toml`¶

`strict_from_list`¶

`strip_runtime_keys`¶

`BoundsConfig`¶

`LTISystemsAnalyzer`¶

`linearize`¶

`as_numpy_f`¶

`discretize_euler`¶

`linearize_plant`¶

`derive_model`¶

`inject_model`¶

`inject_plant_derived`¶

`PluginImportError`¶

`import_modules`¶