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.

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:

If plant.dynamics returns None (the ABC default — LTI plants need not override it), the LTI matmul path is used.
Otherwise the nonlinear path is used, integrating plant.dynamics with semi-implicit Euler. On torch this is vectorized with torch.vmap; on numpy it loops over the batch.

All arrays live in the plant's backend, so torch inputs stay torch throughout and run as batched native ops.

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]

    self._has_dynamics = plant.dynamics(state=self.bk.zeros(self.D_x), control=self.bk.zeros(self.D_u)) is not None
    self._vmap = None
    torch = getattr(self.bk, "torch", None)
    if self._has_dynamics and torch is not None:
        # Vectorize the single-state nonlinear dynamics over the batch.
        self._vmap = torch.vmap(plant.dynamics, in_dims=(0, 0))

state_dim `property` ¶

state_dim: int

State dimension \(D_x\).

control_dim `property` ¶

control_dim: int

Control dimension \(D_u\).

dynamics_fn ¶

dynamics_fn(x_batch, u_batch, dt: float)

Batched dynamics update.

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

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):
    """Batched dynamics update.

    Args:
        x_batch: Batch of states (N, D_x).
        u_batch: Batch of controls (N, D_u).
        dt: Time step (s).

    Returns:
        Batch of next states (N, D_x).
    """
    if self._has_dynamics:
        return self._integrate(x_batch, u_batch, dt)
    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

`resolve_config_path`¶

Resolve a TOML config path to an existing file.

Parameters:

Name	Type	Description	Default
`path`	`str`	Absolute path, a `configs/...` package-relative name, or a bare filename relative to the CWD / packaged configs.	required

Returns:

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

Raises:

Type	Description
`FileNotFoundError`	If no candidate location exists.

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, a ``configs/...`` package-relative name, or a
            bare filename relative to the CWD / packaged configs.

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

    Raises:
        FileNotFoundError: If no candidate location exists.
    """
    path = str(path)
    p = Path(path)
    if p.is_absolute():
        if not p.exists():
            raise FileNotFoundError(f"Config not found: {path}")
        return p

    candidates: list[Path] = []
    if path.startswith("configs/"):
        packaged = _package_config_root().parent.joinpath(path)
        candidates.append(packaged)
        candidates.append(p)
    elif p.exists():
        candidates.append(p)
    else:
        packaged = _package_config_root().joinpath(p.name)
        candidates.append(packaged)
        candidates.append(p)

    for candidate in candidates:
        if candidate.exists():
            return candidate

    raise FileNotFoundError(f"Config not found: {path} (tried: {', '.join(str(c) for c in candidates)})")

`get_config_path`¶

Public helper: resolve a config name (without the configs/ prefix).

Parameters:

Name	Type	Description	Default
`name`	`str`	Package-relative config name, e.g. `controllers/lqr_base.toml`.	required

Returns:

Type	Description
`Path`	A `pathlib.Path` pointing at the packaged config file.

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

def get_config_path(name: str) -> Path:
    """Public helper: resolve a config name (without the ``configs/`` prefix).

    Args:
        name: Package-relative config name, e.g. ``controllers/lqr_base.toml``.

    Returns:
        A :class:`pathlib.Path` pointing at the packaged config file.
    """
    return resolve_config_path(f"configs/{name.lstrip('/')}")

`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`	`ndarray`	State matrix (n, n).	required
`B`	`ndarray \| None`	Input matrix (n, m). Defaults to zeros.	`None`
`C`	`ndarray \| None`	Output matrix (p, n). Defaults to zeros.	`None`
`D`	`ndarray \| 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`

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

def __init__(
    self,
    A: np.ndarray,
    B: np.ndarray | None = None,
    C: np.ndarray | None = None,
    D: np.ndarray | None = None,
    dt: float | None = None,
    backend: ArrayBackend | None = None,
) -> None:
    self.bk = backend if backend is not None else NumpyBackend()
    self.A: np.ndarray = A
    self.B: np.ndarray = B if B is not None else self.bk.zeros((A.shape[0], 0))
    self.C: np.ndarray = C if C is not None else self.bk.zeros((0, A.shape[0]))
    self.D: np.ndarray = 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"]

    eigA = self.bk.eigvals(self.A)
    if self.bk.any(self.bk.real(eigA) >= 0):
        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"]

    eigA = self.bk.eigvals(self.A)
    if self.bk.any(self.bk.real(eigA) >= 0):
        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
    Wo = self._solve_continuous_lyap(Q)
    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
    return self._finite_horizon_gramian(Q, 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

Compute the 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:
    """Compute the Hankel singular values σ_i = sqrt(λ_i(Wc Wo)).

    Returns:
        Hankel singular values sorted in descending order (n,).
    """
    Wc = self.controllability_gramian()
    Wo = self.observability_gramian()
    prod = Wc @ Wo
    eigs = self.bk.real(self.bk.eigvals(prod))
    eigs = self.bk.where(eigs < 0, self.bk.zeros_like(eigs), eigs)
    sigma = self.bk.sqrt(self.bk.sort(eigs)[::-1])
    return sigma

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).
    """
    C_mat = self.controllabilty()
    O_mat = self.observability()
    rank_c = self.bk.matrix_rank(C_mat)
    rank_o = self.bk.matrix_rank(O_mat)
    cond_c = self.bk.cond(C_mat) if rank_c == C_mat.shape[0] else np.inf
    cond_o = self.bk.cond(O_mat) if rank_o == O_mat.shape[0] 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()
    wc = self.controllability_gramian() if self.is_controllable() else None
    wo = self.observability_gramian() if self.is_observable() 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).")

    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)))

    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

Utilities¶

parse_matrix¶

ArrayBackend¶

solve_qp abstractmethod ¶

jacobian abstractmethod ¶

NumpyBackend¶

solve_qp ¶

jacobian ¶

TorchBackend¶

solve_qp ¶

jacobian ¶

BatchedDynamicsAdapter¶

state_dim property ¶

control_dim property ¶

dynamics_fn ¶

cost_fn ¶

resolve_config_path¶

get_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 ¶

balanced_realization ¶

balanced_truncate ¶

reset_cache ¶

rank_report ¶

summary ¶

linearize¶

as_numpy_f¶

discretize_euler¶

linearize_plant¶

derive_model¶

inject_model¶

inject_plant_derived¶

`parse_matrix`¶

`ArrayBackend`¶

solve_qp `abstractmethod` ¶

jacobian `abstractmethod` ¶

`NumpyBackend`¶

`TorchBackend`¶

`BatchedDynamicsAdapter`¶

state_dim `property` ¶

control_dim `property` ¶

`resolve_config_path`¶

`get_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`¶