Estimators¶

shinro.estimators

State estimation algorithms for reconstructing system state from measurements.

Provides discrete-time state estimators that combine dynamics models with sensor measurements. All estimators implement the StateEstimator ABC.

Available estimators: KalmanFilter — Optimal stochastic filter (predict-update cycle) ExtendedKalmanFilter — Nonlinear EKF (local linearization of f, h) UnscentedKF — Nonlinear UKF (deterministic sigma-point propagation) ComplementaryFilter — Rate + absolute-angle fusion (no model, no noise stats) LuenbergerObserver — Deterministic observer with fixed gain

`KalmanFilter`¶

Bases: StateEstimator

Discrete-time linear Kalman filter for optimal state estimation.

Implements the predict-update cycle for a system of the form:

\[ x_{k+1} &= A x_k + B u_k + w_k, \quad w_k \sim \mathcal{N}(0, Q) \\ y_k &= C x_k + D u_k + v_k, \quad v_k \sim \mathcal{N}(0, R) \]

Tracks the posterior state estimate \(\hat{x}\) and error covariance \(P\) through the standard Kalman filter equations.

Uses flat vectors \((n,)\) throughout (not column \((n, 1)\)), matching the rest of the control stack.

Parameters:

Name	Type	Description	Default
`A`		State transition matrix (n_x, n_x).	required
`B`		Control input matrix (n_x, n_u).	required
`Q`		Process noise covariance (n_x, n_x).	required
`R`		Measurement noise covariance (n_y, n_y).	required
`C`	`Any \| None`	Observation matrix (n_y, n_x). Defaults to identity.	`None`
`D`	`Any \| None`	Feedthrough matrix (n_y, n_u). Defaults to zeros.	`None`
`x0`	`Any \| None`	Initial state estimate (n_x,). Defaults to zeros.	`None`
`backend`	`ArrayBackend \| None`	Array backend. Defaults to NumpyBackend.	`None`

Source code in src/shinro/estimators/kalman_filter.py

def __init__(
    self,
    A,
    B,
    Q,
    R,
    C: Any | None = None,
    D: Any | None = None,
    x0: Any | None = None,
    backend: ArrayBackend | None = None,
):
    self.bk = backend or NumpyBackend()
    self.A = A
    self.B = B
    self.Q = Q
    self.R = R

    self.C = self.bk.eye(A.shape[0]) if C is None else C
    self.D = self.bk.zeros((self.C.shape[0], B.shape[1])) if D is None else D

    self.x_hat = self.bk.zeros(A.shape[0]) if x0 is None else self.bk.ravel(self.bk.copy(x0))
    self.P = self.bk.eye(A.shape[0]) * 0.1

estimate ¶

estimate(measurement, control_input)

Run one predict-update cycle and return the posterior state estimate.

Implements the standard Kalman filter equations:

Predict: \(x_{\text{pred}} = A \hat{x} + B u\) \(P_{\text{pred}} = A P A^T + Q\)
Update: \(K = P_{\text{pred}} C^T (C P_{\text{pred}} C^T + R)^{-1}\) \(\hat{x} = x_{\text{pred}} + K (y - C x_{\text{pred}} - D u)\) \(P = (I - K C) P_{\text{pred}}\)

Parameters:

Name	Type	Description	Default
`measurement`		Observation vector (n_y,) from sensors.	required
`control_input`		Control vector (n_u,) applied at this step.	required

Returns:

Type	Description
	Posterior state estimate \(\hat{x}\) (n_x,).

Source code in src/shinro/estimators/kalman_filter.py

def estimate(self, measurement, control_input):
    """Run one predict-update cycle and return the posterior state estimate.

    Implements the standard Kalman filter equations:

    1. Predict:
       :math:`x_{\\text{pred}} = A \\hat{x} + B u`
       :math:`P_{\\text{pred}} = A P A^T + Q`

    2. Update:
       :math:`K = P_{\\text{pred}} C^T (C P_{\\text{pred}} C^T + R)^{-1}`
       :math:`\\hat{x} = x_{\\text{pred}} + K (y - C x_{\\text{pred}} - D u)`
       :math:`P = (I - K C) P_{\\text{pred}}`

    Args:
        measurement: Observation vector (n_y,) from sensors.
        control_input: Control vector (n_u,) applied at this step.

    Returns:
        Posterior state estimate :math:`\\hat{x}` (n_x,).
    """
    x_pred = self.A @ self.x_hat + self.B @ control_input
    self.P = self.A @ self.P @ self.A.T + self.Q

    S = self.C @ self.P @ self.C.T + self.R
    K_gain = self.P @ self.C.T @ self.bk.inv(S)

    y_pred = self.C @ x_pred + self.D @ control_input
    innovations = measurement - y_pred

    self.x_hat = x_pred + K_gain @ innovations
    self.P = (self.bk.eye(self.A.shape[0]) - K_gain @ self.C) @ self.P

    return self.x_hat

measure ¶

measure(state)

Expected measurement C x for a true state (feedthrough D is zero).

Source code in src/shinro/estimators/kalman_filter.py

def measure(self, state):
    """Expected measurement ``C x`` for a true ``state`` (feedthrough ``D`` is zero)."""
    return self.C @ self.bk.ravel(state)

reset ¶

reset(x0: Any | None = None)

Reset the filter to its initial state.

Parameters:

Name	Type	Description	Default
`x0`	`Any \| None`	Initial state estimate (n_x,). Defaults to zeros.	`None`

Source code in src/shinro/estimators/kalman_filter.py

def reset(self, x0: Any | None = None):
    """Reset the filter to its initial state.

    Args:
        x0: Initial state estimate (n_x,). Defaults to zeros.
    """
    self.x_hat = self.bk.zeros(self.A.shape[0]) if x0 is None else self.bk.ravel(self.bk.copy(x0))
    self.P = self.bk.eye(self.A.shape[0]) * 0.1

from_config `classmethod` ¶

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

Create a Kalman filter from a TOML config dict or KalmanFilterConfig.

Config fields

process_noise: Diagonal Q weights (n_x,) or full Q matrix (n_x, n_x). measurement_noise: Diagonal R weights (n_y,) or full R matrix (n_y, n_y). dt: Time step — used to set B = dt * I unless B_dynamics is given. A_dynamics: Optional full A matrix (n_x, n_x). Defaults to I. B_dynamics: Optional full B matrix (n_x, n_u). Defaults to dt * I. C: Optional full C matrix (n_y, n_x). Defaults to I. D: Optional full D matrix (n_y, n_u). Defaults to zeros.

Parameters:

Name	Type	Description	Default
`config`		TOML config dict or KalmanFilterConfig.	required
`backend`	`ArrayBackend \| None`	Array backend. Defaults to NumpyBackend.	`None`

Returns:

Type	Description
	KalmanFilter instance.

Source code in src/shinro/estimators/kalman_filter.py

@classmethod
def from_config(cls, config, backend: ArrayBackend | None = None):
    """Create a Kalman filter from a TOML config dict or :class:`KalmanFilterConfig`.

    Config fields:
        process_noise: Diagonal Q weights (n_x,) or full Q matrix (n_x, n_x).
        measurement_noise: Diagonal R weights (n_y,) or full R matrix (n_y, n_y).
        dt: Time step — used to set B = dt * I unless B_dynamics is given.
        A_dynamics: Optional full A matrix (n_x, n_x). Defaults to I.
        B_dynamics: Optional full B matrix (n_x, n_u). Defaults to dt * I.
        C: Optional full C matrix (n_y, n_x). Defaults to I.
        D: Optional full D matrix (n_y, n_u). Defaults to zeros.

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

    Returns:
        KalmanFilter instance.
    """
    bk = backend or NumpyBackend()
    cfg = cls.parse_config(config)
    Q = parse_matrix(bk, cfg.process_noise)
    n = Q.shape[0]
    R = parse_matrix(bk, cfg.measurement_noise)
    n_y = R.shape[0]
    A = bk.array(cfg.A_dynamics) if cfg.A_dynamics is not None else bk.eye(n)
    if cfg.B_dynamics is not None:
        B = bk.array(cfg.B_dynamics)
    elif cfg.dt is not None:
        B = cfg.dt * bk.eye(n)
    else:
        raise ValueError("KalmanFilter: no B_dynamics and no dt — standalone use requires one of them")
    return cls(
        A=A,
        B=B,
        Q=Q,
        R=R,
        C=bk.array(cfg.C) if cfg.C is not None else bk.eye(n),
        D=bk.array(cfg.D) if cfg.D is not None else bk.zeros((n_y, B.shape[1])),
        x0=bk.zeros(n),
        backend=bk,
    )

`ExtendedKalmanFilter`¶

Bases: StateEstimator

Extended Kalman filter for nonlinear state estimation.

Estimates the state of a continuous-time nonlinear system

\[ \dot{x} = f(x, u), \qquad y = h(x) \]

by linearizing \(f\) and \(h\) about the current estimate and running the standard Kalman predict-update cycle on the local model.

Conventions. Everything is a flat vector — state \((n_x,)\), control \((n_u,)\), measurement \((n_y,)\) — matching the rest of the control stack. dynamics_fn(x, u) -> dx/dt is continuous-time: the filter Euler-discretizes it at dt, so the discrete process Jacobian is \(F = I + dt \cdot \partial f/\partial x\). The control Jacobian \(\partial f/\partial u\) is never needed — the EKF assumes the applied input is known exactly, so it does not enter the covariance.

Predict (local linearization about \(\hat{x}\)):

\[ x^- &= \hat{x} + dt \cdot f(\hat{x}, u) \\ F &= I + dt \cdot \frac{\partial f}{\partial x}\bigg|_{\hat{x}} \\ P^- &= F P F^T + Q \]

Update (linearization about \(x^-\)):

\[ H &= \frac{\partial h}{\partial x}\bigg|_{x^-} \\ S &= H P^- H^T + R \\ K &= P^- H^T S^{-1} \\ \hat{x} &= x^- + K \,(y - h(x^-)) \\ P &= (I - K H) P^- \]

The two Jacobians are computed by the backend's numerical Jacobian (shinro.utils.array_backend.ArrayBackend.jacobian — central finite differences on numpy, autograd on torch; on the tracing backend it lowers as graph nodes via shinro.codegen.trace_backend.TraceBackend.jacobian), so dynamics_fn and measurement_fn are the only model inputs the filter needs. A linear measurement model may be given as measurement_matrix instead, which yields the exact H = C with no differentiation.

A model callable may accept an optional bk (as shinro.components.Plant.dynamics does). The filter passes its own backend when the signature accepts one, which is what lets a traced filter route the plant's ops through its TraceBackend; plain two-argument callables keep working.

Parameters:

Name	Type	Description	Default
`dynamics_fn`	`Callable[..., Any] \| None`	Continuous-time process model `f(x, u) -> dx/dt`. Receives and returns flat backend arrays. May be `None` only when built via `from_config`, to be injected later.	required
`measurement_fn`	`Callable[..., Any] \| None`	Measurement model `h(x) -> y`. Receives and returns flat backend arrays. May be `None` when `measurement_matrix` supplies a linear model instead, or (via `from_config`) to be injected later.	required
`dt`	`float`	Integration / sample time step in seconds.	required
`Q`	`Any`	Process noise covariance (n_x, n_x).	required
`R`	`Any`	Measurement noise covariance (n_y, n_y).	required
`x0`	`Any \| None`	Initial state estimate (n_x,). Defaults to zeros.	`None`
`backend`	`ArrayBackend \| None`	Array backend. Defaults to NumpyBackend.	`None`
`measurement_matrix`	`Any \| None`	Optional linear measurement matrix `C` (n_y, n_x), making `h(x) = C x` and the measurement Jacobian exactly `C` — no callable and no finite differences needed.	`None`

Source code in src/shinro/estimators/extended_kf.py

def __init__(
    self,
    dynamics_fn: Callable[..., Any] | None,
    measurement_fn: Callable[..., Any] | None,
    dt: float,
    Q: Any,
    R: Any,
    x0: Any | None = None,
    backend: ArrayBackend | None = None,
    measurement_matrix: Any | None = None,
):
    self.bk = backend or NumpyBackend()
    self.dynamics_fn = dynamics_fn
    self.measurement_fn = measurement_fn
    self.dt = dt
    self.Q = Q
    self.R = R
    self.n_x = Q.shape[0]
    self.n_y = R.shape[0]
    self.C = None if measurement_matrix is None else self.bk.array(measurement_matrix)
    if self.C is not None and tuple(self.C.shape) != (self.n_y, self.n_x):
        raise ValueError(
            f"measurement_matrix shape {tuple(self.C.shape)} does not match R's {self.n_y} "
            f"measurements x Q's {self.n_x} states"
        )
    self.x_hat = self.bk.zeros(self.n_x) if x0 is None else self._flat(self.bk.copy(x0), self.n_x)
    self.P = self.bk.eye(self.n_x) * 0.1

measure ¶

measure(state)

Expected measurement h(x) for a true state (the linear C or the callable/plant model).

Source code in src/shinro/estimators/extended_kf.py

def measure(self, state):
    """Expected measurement ``h(x)`` for a true ``state`` (the linear ``C`` or the callable/plant model)."""
    return self._measurement(self._flat(state, self.n_x))

estimate ¶

estimate(measurement, control_input)

Run one predict-update cycle and return the posterior state estimate.

See the class docstring for the equations. The Jacobians are the local linearization; the covariance update uses the same simple \((I - KH)P^-\) form as shinro.estimators.kalman_filter.KalmanFilter (not the Joseph form).

Parameters:

Name	Type	Description	Default
`measurement`		Observation vector \(y\) (n_y,).	required
`control_input`		Control vector \(u\) applied at this step (n_u,).	required

Returns:

Type	Description
	Posterior state estimate \(\hat{x}\) (n_x,).

Raises:

Type	Description
`RuntimeError`	If the model is unset — `dynamics_fn` missing, or neither `measurement_fn` nor `measurement_matrix` supplied (a filter built with `from_config` starts with both callables `None` and they must be injected first).

Source code in src/shinro/estimators/extended_kf.py

def estimate(self, measurement, control_input):
    """Run one predict-update cycle and return the posterior state estimate.

    See the class docstring for the equations. The Jacobians are the local
    linearization; the covariance update uses the same simple
    :math:`(I - KH)P^-` form as
    :class:`~shinro.estimators.kalman_filter.KalmanFilter` (not the Joseph form).

    Args:
        measurement: Observation vector :math:`y` (n_y,).
        control_input: Control vector :math:`u` applied at this step (n_u,).

    Returns:
        Posterior state estimate :math:`\\hat{x}` (n_x,).

    Raises:
        RuntimeError: If the model is unset — ``dynamics_fn`` missing, or
            neither ``measurement_fn`` nor ``measurement_matrix`` supplied
            (a filter built with :meth:`from_config` starts with both
            callables ``None`` and they must be injected first).
    """
    dynamics_fn = self._dynamics_fn()  # validate up front, before any partial work

    x = self._flat(self.x_hat, self.n_x)
    u = self._flat(control_input, control_input.shape[0])
    z = self._flat(measurement, self.n_y)

    # --- predict: Euler-discretize the continuous-time process model ---
    f = self._flat(call_model(dynamics_fn, x, u, bk=self.bk), self.n_x)
    x_pred = x + self.dt * f
    F = self.bk.eye(self.n_x) + self.dt * self._dynamics_jacobian(x, u)
    P_pred = F @ self.P @ F.T + self.Q

    # --- update: linearize the measurement model about the prediction ---
    H = self._measurement_jacobian(x_pred)
    z_pred = self._measurement(x_pred)
    S = H @ P_pred @ H.T + self.R
    K_gain = P_pred @ H.T @ self.bk.inv(S)

    self.x_hat = x_pred + K_gain @ (z - z_pred)
    self.P = (self.bk.eye(self.n_x) - K_gain @ H) @ P_pred
    return self.x_hat

reset ¶

reset(x0: Any | None = None)

Reset the filter to its initial state.

Parameters:

Name	Type	Description	Default
`x0`	`Any \| None`	Initial state estimate (n_x,). Defaults to zeros.	`None`

Source code in src/shinro/estimators/extended_kf.py

def reset(self, x0: Any | None = None):
    """Reset the filter to its initial state.

    Args:
        x0: Initial state estimate (n_x,). Defaults to zeros.
    """
    self.x_hat = self.bk.zeros(self.n_x) if x0 is None else self._flat(self.bk.copy(x0), self.n_x)
    self.P = self.bk.eye(self.n_x) * 0.1

attach_plant ¶

attach_plant(plant: Any) -> None

Inject the plant's process model, and its sensor if it exposes one.

The compiled/served path hands the EKF the plant so its dynamics_fn is the plant's model — written with the rank helpers and an optional bk, so it traces and lowers — instead of a hand-wired lambda. When the plant also overrides shinro.components.Plant.measurement, and the filter has neither a measurement_fn nor a linear C, that sensor becomes h too, derived from the same plant so the live and lowered measurement models cannot drift.

Each side is filled only when it is unset, so an explicit injection (or a measurement_matrix) always wins. A plant with no sensor model leaves h to the config's measurement_matrix.

Parameters:

Name	Type	Description	Default
`plant`	`Any`	The plant whose `shinro.components.Plant.dynamics` (and optional `measurement`) to use.	required

Raises:

Type	Description
`ValueError`	If the plant exposes no dynamics and none was injected.

Source code in src/shinro/estimators/extended_kf.py

def attach_plant(self, plant: Any) -> None:
    """Inject the plant's process model, and its sensor if it exposes one.

    The compiled/served path hands the EKF the plant so its ``dynamics_fn``
    is the plant's model — written with the rank helpers and an optional
    ``bk``, so it traces and lowers — instead of a hand-wired lambda. When
    the plant also overrides :meth:`~shinro.components.Plant.measurement`,
    and the filter has neither a ``measurement_fn`` nor a linear ``C``, that
    sensor becomes ``h`` too, derived from the same plant so the live and
    lowered measurement models cannot drift.

    Each side is filled only when it is **unset**, so an explicit injection
    (or a ``measurement_matrix``) always wins. A plant with no sensor model
    leaves ``h`` to the config's ``measurement_matrix``.

    Args:
        plant: The plant whose :meth:`~shinro.components.Plant.dynamics`
            (and optional ``measurement``) to use.

    Raises:
        ValueError: If the plant exposes no dynamics and none was injected.
    """
    if self.dynamics_fn is None:
        dynamics = getattr(plant, "dynamics", None)
        if dynamics is None or not callable(dynamics):
            raise ValueError(
                f"attach_plant: {type(plant).__name__} exposes no dynamics(), and the filter "
                f"has no dynamics_fn — inject one explicitly (ekf.dynamics_fn = ...)."
            )
        self.dynamics_fn = dynamics
    measurement = getattr(type(plant), "measurement", None)
    if (
        self.measurement_fn is None
        and self.C is None
        and measurement is not None
        and measurement is not Plant.measurement
    ):
        self.measurement_fn = plant.measurement

from_config `classmethod` ¶

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

Create an extended Kalman filter from a TOML config dict or ExtendedKalmanFilterConfig.

Config fields

process_noise: Diagonal Q weights (n_x,) or full Q matrix (n_x, n_x). measurement_noise: Diagonal R weights (n_y,) or full R matrix (n_y, n_y). dt: Time step. Required at runtime; injected from the plant in scenario builds. measurement_matrix: Optional linear measurement matrix C (n_y, n_x); when set, h(x) = C x and H = C exactly, with no callable. initial_state: Optional initial state estimate (n_x,). Defaults to zeros.

The dynamics_fn / measurement_fn callables cannot be serialized to TOML. They are created as None and must be injected after construction:

.. code-block:: python

ekf = ExtendedKalmanFilter.from_config(config)
ekf.dynamics_fn = my_f
ekf.measurement_fn = my_h

A config may instead carry a linear measurement_matrix (C), in which case measurement_fn is never needed — the compiled path takes this route, since a plant has no measurement callable to inject.

Parameters:

Name	Type	Description	Default
`config`		TOML config dict or ExtendedKalmanFilterConfig.	required
`backend`	`ArrayBackend \| None`	Array backend. Defaults to NumpyBackend.	`None`

Returns:

Type	Description
	ExtendedKalmanFilter instance.

Raises:

Type	Description
`ValueError`	If `dt` is None (standalone use requires it).

Source code in src/shinro/estimators/extended_kf.py

@classmethod
def from_config(cls, config, backend: ArrayBackend | None = None):
    """Create an extended Kalman filter from a TOML config dict or :class:`ExtendedKalmanFilterConfig`.

    Config fields:
        process_noise: Diagonal Q weights (n_x,) or full Q matrix (n_x, n_x).
        measurement_noise: Diagonal R weights (n_y,) or full R matrix (n_y, n_y).
        dt: Time step. Required at runtime; injected from the plant in
            scenario builds.
        measurement_matrix: Optional linear measurement matrix ``C`` (n_y, n_x);
            when set, ``h(x) = C x`` and ``H = C`` exactly, with no callable.
        initial_state: Optional initial state estimate (n_x,). Defaults to zeros.

    The ``dynamics_fn`` / ``measurement_fn`` callables cannot be serialized
    to TOML. They are created as ``None`` and must be injected after
    construction:

    .. code-block:: python

        ekf = ExtendedKalmanFilter.from_config(config)
        ekf.dynamics_fn = my_f
        ekf.measurement_fn = my_h

    A config may instead carry a linear ``measurement_matrix`` (``C``), in which
    case ``measurement_fn`` is never needed — the compiled path takes this route,
    since a plant has no measurement callable to inject.

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

    Returns:
        ExtendedKalmanFilter instance.

    Raises:
        ValueError: If ``dt`` is None (standalone use requires it).
    """
    bk = backend or NumpyBackend()
    cfg = cls.parse_config(config)
    if cfg.dt is None:
        raise ValueError(
            "ExtendedKalmanFilter: dt is required — omit it only in scenario "
            "builds, where the plant's dt is injected"
        )
    Q = parse_matrix(bk, cfg.process_noise)
    R = parse_matrix(bk, cfg.measurement_noise)
    x0 = None if cfg.initial_state is None else bk.array(cfg.initial_state)
    return cls(
        dynamics_fn=None,
        measurement_fn=None,
        dt=cfg.dt,
        Q=Q,
        R=R,
        x0=x0,
        backend=bk,
        measurement_matrix=cfg.measurement_matrix,
    )

`UnscentedKF`¶

Bases: StateEstimator

Source code in src/shinro/estimators/unscented_kf.py

def __init__(
    self,
    dynamics_fn: Callable[...,Any]| None,
    measurement_fn: Callable[...,Any]| None,
    dt: float,
    alpha: float,
    beta:float,
    kappa:float,
    Q: Any,
    R:Any,
    x0: Any | None= None,
    backend: ArrayBackend | None= None,
    measurement_matrix: Any | None = None,
    batched: bool | None = None,
    ):
        self.bk= backend or NumpyBackend()
        self.dynamics_fn= dynamics_fn
        self.measurement_fn=measurement_fn
        self.batched= batched
        self.dt= dt
        self.alpha=alpha
        self.beta= beta
        self.kappa=kappa
        self.Q=Q
        self.R=R

        self.nx=Q.shape[0] #process noise
        self.ny= R.shape[0] #measurement noise

        #optional linear measurement model: h(x) = C x, no callable needed
        self.C= None if measurement_matrix is None else self.bk.array(measurement_matrix)
        if self.C is not None and tuple(self.C.shape) != (self.ny, self.nx):
            raise ValueError(
                f"measurement_matrix shape {tuple(self.C.shape)} does not match R's {self.ny} "
                f"measurements x Q's {self.nx} states"
            )

        # initializing the kalman algo values
        self.x_hat= self.bk.zeros(self.nx) if x0 is None else self.bk.ravel(self.bk.copy(x0))
        self.P=self.bk.eye(self.nx)*0.1

        #calculate the weights in the start
        self._weights_sigmapoints()

measure ¶

measure(state)

Expected measurement h(x) for a true state (the linear C or the callable/plant model).

Source code in src/shinro/estimators/unscented_kf.py

def measure(self, state):
    """Expected measurement ``h(x)`` for a true ``state`` (the linear ``C`` or the callable/plant model)."""
    return self._measurement(self.bk.ravel(state))

reset ¶

reset(x0: Any | None = None)

Reset the filter to its initial state.

Clears the posterior estimate and covariance back to their construction values (\(\hat{x}=0\), \(P=0.1\,I\)) so the filter can be reused on a new run. Mirrors ExtendedKalmanFilter.reset.

Parameters:

Name	Type	Description	Default
`x0`	`Any \| None`	Initial state estimate (n_x,). Defaults to zeros.	`None`

Source code in src/shinro/estimators/unscented_kf.py

def reset(self, x0: Any | None = None):
    """Reset the filter to its initial state.

    Clears the posterior estimate and covariance back to their construction
    values (:math:`\\hat{x}=0`, :math:`P=0.1\\,I`) so the filter can be reused
    on a new run. Mirrors :meth:`ExtendedKalmanFilter.reset`.

    Args:
        x0: Initial state estimate (n_x,). Defaults to zeros.
    """
    self.x_hat= self.bk.zeros(self.nx) if x0 is None else self.bk.ravel(self.bk.copy(x0))
    self.P=self.bk.eye(self.nx)*0.1

attach_plant ¶

attach_plant(plant: Any) -> None

Inject the plant's process model, and its sensor if it exposes one.

The compiled/served path hands the UKF the plant so its dynamics_fn is the plant's model — written with the rank helpers and an optional bk, so it traces and lowers — instead of a hand-wired lambda. When the plant also overrides shinro.components.Plant.measurement, and the filter has neither a measurement_fn nor a linear C, that sensor becomes h too, derived from the same plant so the live and lowered measurement models cannot drift.

Each side is filled only when it is unset, so an explicit injection (or a measurement_matrix) always wins. A plant with no sensor model leaves h to the config's measurement_matrix.

Parameters:

Name	Type	Description	Default
`plant`	`Any`	The plant whose `shinro.components.Plant.dynamics` (and optional `measurement`) to use.	required

Raises:

Type	Description
`ValueError`	If the plant exposes no dynamics and none was injected.

Source code in src/shinro/estimators/unscented_kf.py

def attach_plant(self, plant: Any) -> None:
    """Inject the plant's process model, and its sensor if it exposes one.

    The compiled/served path hands the UKF the plant so its ``dynamics_fn``
    is the plant's model — written with the rank helpers and an optional
    ``bk``, so it traces and lowers — instead of a hand-wired lambda. When
    the plant also overrides :meth:`~shinro.components.Plant.measurement`,
    and the filter has neither a ``measurement_fn`` nor a linear ``C``, that
    sensor becomes ``h`` too, derived from the same plant so the live and
    lowered measurement models cannot drift.

    Each side is filled only when it is **unset**, so an explicit injection
    (or a ``measurement_matrix``) always wins. A plant with no sensor model
    leaves ``h`` to the config's ``measurement_matrix``.

    Args:
        plant: The plant whose :meth:`~shinro.components.Plant.dynamics`
            (and optional ``measurement``) to use.

    Raises:
        ValueError: If the plant exposes no dynamics and none was injected.
    """
    if self.dynamics_fn is None:
        dynamics = getattr(plant, "dynamics", None)
        if dynamics is None or not callable(dynamics):
            raise ValueError(
                f"attach_plant: {type(plant).__name__} exposes no dynamics(), and the filter "
                f"has no dynamics_fn — inject one explicitly (ukf.dynamics_fn = ...)."
            )
        self.dynamics_fn = dynamics
    measurement = getattr(type(plant), "measurement", None)
    if (
        self.measurement_fn is None
        and self.C is None
        and measurement is not None
        and measurement is not Plant.measurement
    ):
        self.measurement_fn = plant.measurement

from_config `classmethod` ¶

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

Create an unscented Kalman filter from a TOML config dict or UnscentedKFConfig.

The dynamics_fn / measurement_fn callables cannot be serialized to TOML: they are created as None and must be injected after construction (ukf.dynamics_fn = ...), or filled from a plant. A config may instead carry a linear measurement_matrix (C), in which case measurement_fn is never needed.

Parameters:

Name	Type	Description	Default
`config`		TOML config dict or UnscentedKFConfig.	required
`backend`	`ArrayBackend \| None`	Array backend. Defaults to NumpyBackend.	`None`

Returns:

Type	Description
	UnscentedKF instance.

Raises:

Type	Description
`ValueError`	If `dt` is None (standalone use requires it).

Source code in src/shinro/estimators/unscented_kf.py

@classmethod
def from_config(cls, config, backend: ArrayBackend | None = None):
    """Create an unscented Kalman filter from a TOML config dict or :class:`UnscentedKFConfig`.

    The ``dynamics_fn`` / ``measurement_fn`` callables cannot be serialized
    to TOML: they are created as ``None`` and must be injected after
    construction (``ukf.dynamics_fn = ...``), or filled from a plant. A
    config may instead carry a linear ``measurement_matrix`` (``C``), in which
    case ``measurement_fn`` is never needed.

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

    Returns:
        UnscentedKF instance.

    Raises:
        ValueError: If ``dt`` is None (standalone use requires it).
    """
    bk = backend or NumpyBackend()
    cfg = cls.parse_config(config)
    if cfg.dt is None:
        raise ValueError(
            "UnscentedKF: dt is required — omit it only in scenario builds, "
            "where the plant's dt is injected"
        )
    Q = parse_matrix(bk, cfg.process_noise)
    R = parse_matrix(bk, cfg.measurement_noise)
    x0 = None if cfg.initial_state is None else bk.array(cfg.initial_state)
    return cls(
        dynamics_fn=None,
        measurement_fn=None,
        dt=cfg.dt,
        alpha=cfg.alpha,
        beta=cfg.beta,
        kappa=cfg.kappa,
        Q=Q,
        R=R,
        x0=x0,
        backend=bk,
        measurement_matrix=cfg.measurement_matrix,
    )

`ComplementaryFilter`¶

Bases: StateEstimator

Complementary filter for rate + absolute-angle sensor fusion.

Estimates an angle \(\theta\) from a rate \(\omega\) (good at high frequency) and an absolute measurement \(\theta_a\) (good at low frequency). The discrete form is

\[ \hat{\theta}_k = \alpha \,(\hat{\theta}_{k-1} + \omega_{k-1} \Delta t) + (1 - \alpha)\, \theta_{a,k} \]

where \(\alpha \in [0, 1]\) is the blend weight on the integrated rate. It is the discrete equivalent of the continuous blend \(\hat{\theta} = \frac{1}{1+\tau s}\theta_a + \frac{\tau s}{1+\tau s}\frac{\omega}{s}\) with \(\alpha = \tau / (\tau + \Delta t)\): the rate is high-passed (its drift is rejected), the absolute is low-passed (its noise is rejected).

Measurement layout. The framework passes one measurement vector; it is the concatenation of the per-axis rate followed by the per-axis absolute, [\omega_0 .. \omega_{N-1}, \theta_{a,0} .. \theta_{a,N-1}] (length 2 * channels). Everything is flat (N,), matching the EKF/UKF.

State. The estimate \(\hat{\theta}\) (channels,) is the only recurrent state; self.x_hat carries it across ticks (the composed graph threads it as state_x_hat).

Parameters:

Name	Type	Description	Default
`alpha`	`float`	Blend weight on the integrated rate, in `[0, 1]`. Higher means more trust in the rate (faster response, slower drift rejection); `1` is pure integration, `0` is pure measurement.	required
`channels`	`int`	Number of independent angles fused (the state dimension n_x).	required
`dt`	`float`	Sample time in seconds.	required
`x0`	`Any \| None`	Initial angle estimate (channels,). Defaults to zeros.	`None`
`backend`	`ArrayBackend \| None`	Array backend. Defaults to NumpyBackend.	`None`

Source code in src/shinro/estimators/complementary_filter.py

def __init__(
    self,
    alpha: float,
    channels: int,
    dt: float,
    x0: Any | None = None,
    backend: ArrayBackend | None = None,
):
    if not 0.0 <= float(alpha) <= 1.0:
        raise ValueError(f"ComplementaryFilter: alpha must be in [0, 1]; got {alpha}")
    if int(channels) < 1:
        raise ValueError(f"ComplementaryFilter: channels must be >= 1; got {channels}")
    self.bk = backend or NumpyBackend()
    self.alpha = float(alpha)
    self.dt = dt
    self.n_x = int(channels)
    self.n_y = 2 * self.n_x
    self.x_hat = self.bk.zeros(self.n_x) if x0 is None else self.bk.ravel(self.bk.copy(x0))

estimate ¶

estimate(measurement, control_input)

Run one fuse step and return the posterior angle estimate.

Splits the measurement into its rate and absolute halves, integrates the rate over dt, and blends:

\[ \hat{\theta}_k = \alpha (\hat{\theta}_{k-1} + \omega \Delta t) + (1 - \alpha) \theta_a \]

Parameters:

Name	Type	Description	Default
`measurement`		Concatenated `[rates, absolutes]` (2 * n_x,).	required
`control_input`		Unused — a complementary filter has no control input, but the estimator contract requires the argument.	required

Returns:

Type	Description
	Posterior angle estimate \(\hat{\theta}\) (n_x,).

Source code in src/shinro/estimators/complementary_filter.py

def estimate(self, measurement, control_input):
    """Run one fuse step and return the posterior angle estimate.

    Splits the measurement into its rate and absolute halves, integrates the
    rate over ``dt``, and blends:

    .. math::

        \\hat{\\theta}_k = \\alpha (\\hat{\\theta}_{k-1} + \\omega \\Delta t)
            + (1 - \\alpha) \\theta_a

    Args:
        measurement: Concatenated ``[rates, absolutes]`` (2 * n_x,).
        control_input: Unused — a complementary filter has no control input,
            but the estimator contract requires the argument.

    Returns:
        Posterior angle estimate :math:`\\hat{\\theta}` (n_x,).
    """
    m = self.bk.ravel(measurement)
    n = self.n_x
    rates = self.bk.slice_(m, 0, n)
    absolutes = self.bk.slice_(m, n, 2 * n)
    prior = self.x_hat + self.dt * rates
    self.x_hat = self.alpha * prior + (1.0 - self.alpha) * absolutes
    return self.x_hat

reset ¶

reset(x0: Any | None = None)

Reset the filter to its initial state.

Parameters:

Name	Type	Description	Default
`x0`	`Any \| None`	Initial angle estimate (n_x,). Defaults to zeros.	`None`

Source code in src/shinro/estimators/complementary_filter.py

def reset(self, x0: Any | None = None):
    """Reset the filter to its initial state.

    Args:
        x0: Initial angle estimate (n_x,). Defaults to zeros.
    """
    self.x_hat = self.bk.zeros(self.n_x) if x0 is None else self.bk.ravel(self.bk.copy(x0))

from_config `classmethod` ¶

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

Create a complementary filter from a TOML config dict or ComplementaryFilterConfig.

Config fields

alpha: Blend weight on the integrated rate, in [0, 1]. channels: Number of fused axes (n_x); the measurement is 2 * channels. dt: Time step — required at runtime; injected from the plant in scenario builds. initial_state: Optional initial angle estimate (n_x,). Defaults to zeros.

Parameters:

Name	Type	Description	Default
`config`		TOML config dict or ComplementaryFilterConfig.	required
`backend`	`ArrayBackend \| None`	Array backend. Defaults to NumpyBackend.	`None`

Returns:

Type	Description
	ComplementaryFilter instance.

Raises:

Type	Description
`ValueError`	If `dt` is None (standalone use requires it) or `alpha` is outside `[0, 1]`.

Source code in src/shinro/estimators/complementary_filter.py

@classmethod
def from_config(cls, config, backend: ArrayBackend | None = None):
    """Create a complementary filter from a TOML config dict or :class:`ComplementaryFilterConfig`.

    Config fields:
        alpha: Blend weight on the integrated rate, in ``[0, 1]``.
        channels: Number of fused axes (n_x); the measurement is ``2 * channels``.
        dt: Time step — required at runtime; injected from the plant in
            scenario builds.
        initial_state: Optional initial angle estimate (n_x,). Defaults to zeros.

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

    Returns:
        ComplementaryFilter instance.

    Raises:
        ValueError: If ``dt`` is None (standalone use requires it) or ``alpha``
            is outside ``[0, 1]``.
    """
    bk = backend or NumpyBackend()
    cfg = cls.parse_config(config)
    if cfg.dt is None:
        raise ValueError(
            "ComplementaryFilter: dt is required — omit it only in scenario builds, "
            "where the plant's dt is injected"
        )
    x0 = None if cfg.initial_state is None else bk.array(cfg.initial_state)
    return cls(alpha=cfg.alpha, channels=cfg.channels, dt=cfg.dt, x0=x0, backend=bk)

`LuenbergerObserver`¶

Bases: StateEstimator

Luenberger observer for deterministic linear state estimation.

Implements the discrete-time observer dynamics:

\[ \hat{x}_{k+1} = A \hat{x}_k + B u_k + L (y_k - C \hat{x}_k - D u_k) \]

where L is the observer gain chosen to place the eigenvalues of \((A - LC)\) inside the unit circle for stable estimation.

Unlike the Kalman filter, the Luenberger observer uses a fixed gain and does not assume noise statistics. No matrix inverses are needed at runtime — just three matrix-vector multiplies.

Uses flat vectors \((n,)\) throughout (not column \((n, 1)\)), matching the rest of the control stack.

Parameters:

Name	Type	Description	Default
`A`		State transition matrix (n_x, n_x).	required
`B`		Control input matrix (n_x, n_u).	required
`observer_gain`		Observer gain matrix L (n_x, n_y). Must place eigenvalues of (A - LC) inside the unit circle.	required
`C`	`Any \| None`	Output matrix (n_y, n_x). Defaults to identity.	`None`
`D`	`Any \| None`	Feedthrough matrix (n_y, n_u). Defaults to zeros.	`None`
`x0`	`Any \| None`	Initial state estimate (n_x,). Defaults to zeros.	`None`
`backend`	`ArrayBackend \| None`	Array backend. Defaults to NumpyBackend.	`None`

Source code in src/shinro/estimators/luenberger_observer.py

def __init__(
    self,
    A,
    B,
    observer_gain,
    C: Any | None = None,
    D: Any | None = None,
    x0: Any | None = None,
    backend: ArrayBackend | None = None,
):
    self.bk = backend or NumpyBackend()
    self.A = A
    self.B = B
    self.C = self.bk.eye(A.shape[0]) if C is None else C
    self.D = self.bk.zeros((self.C.shape[0], B.shape[1])) if D is None else D
    self.L = observer_gain
    self.x_hat = self.bk.zeros(A.shape[0]) if x0 is None else self.bk.ravel(self.bk.copy(x0))

estimate ¶

estimate(measurement, control_input)

Perform one step of state estimation.

Computes the predicted state from the dynamics, calculates the innovation (measurement residual), and corrects the prediction using the observer gain:

\[ \hat{x}_{k+1} = A \hat{x}_k + B u_k + L (y_k - C (A \hat{x}_k + B u_k) - D u_k) \]

Parameters:

Name	Type	Description	Default
`measurement`		Output measurement \(y_k\) (n_y,).	required
`control_input`		Control input \(u_k\) (n_u,).	required

Returns:

Type	Description
	Updated state estimate \(\hat{x}_{k+1}\) (n_x,).

Source code in src/shinro/estimators/luenberger_observer.py

def estimate(self, measurement, control_input):
    """Perform one step of state estimation.

    Computes the predicted state from the dynamics, calculates the
    innovation (measurement residual), and corrects the prediction
    using the observer gain:

    .. math::

        \\hat{x}_{k+1} = A \\hat{x}_k + B u_k + L (y_k - C (A \\hat{x}_k + B u_k) - D u_k)

    Args:
        measurement: Output measurement :math:`y_k` (n_y,).
        control_input: Control input :math:`u_k` (n_u,).

    Returns:
        Updated state estimate :math:`\\hat{x}_{k+1}` (n_x,).
    """
    x_pred = self.A @ self.x_hat + self.B @ control_input
    innovations = measurement - (self.C @ x_pred + self.D @ control_input)
    self.x_hat = x_pred + self.L @ innovations
    return self.x_hat

measure ¶

measure(state)

Expected measurement C x for a true state (feedthrough D is zero).

Source code in src/shinro/estimators/luenberger_observer.py

def measure(self, state):
    """Expected measurement ``C x`` for a true ``state`` (feedthrough ``D`` is zero)."""
    return self.C @ self.bk.ravel(state)

reset ¶

reset(x0: Any | None = None)

Reset the observer to its initial state.

Parameters:

Name	Type	Description	Default
`x0`	`Any \| None`	Initial state estimate (n_x,). Defaults to zeros.	`None`

Source code in src/shinro/estimators/luenberger_observer.py

def reset(self, x0: Any | None = None):
    """Reset the observer to its initial state.

    Args:
        x0: Initial state estimate (n_x,). Defaults to zeros.
    """
    self.x_hat = self.bk.zeros(self.A.shape[0]) if x0 is None else self.bk.ravel(self.bk.copy(x0))

from_config `classmethod` ¶

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

Create a Luenberger observer from a TOML config dict or LuenbergerObserverConfig.

Config fields

observer_gain: Diagonal gain weights (n_x,) or full gain matrix (n_x, n_y). dt: Time step — used to set B = dt * I unless B_dynamics is given. A_dynamics: Optional full A matrix (n_x, n_x). Defaults to I. B_dynamics: Optional full B matrix (n_x, n_u). Defaults to dt * I. C: Optional full C matrix (n_y, n_x). Defaults to I. D: Optional full D matrix (n_y, n_u). Defaults to zeros.

Parameters:

Name	Type	Description	Default
`config`		TOML config dict or LuenbergerObserverConfig.	required
`backend`	`ArrayBackend \| None`	Array backend. Defaults to NumpyBackend.	`None`

Returns:

Type	Description
	LuenbergerObserver instance.

Source code in src/shinro/estimators/luenberger_observer.py

@classmethod
def from_config(cls, config, backend: ArrayBackend | None = None):
    """Create a Luenberger observer from a TOML config dict or :class:`LuenbergerObserverConfig`.

    Config fields:
        observer_gain: Diagonal gain weights (n_x,) or full gain matrix (n_x, n_y).
        dt: Time step — used to set B = dt * I unless B_dynamics is given.
        A_dynamics: Optional full A matrix (n_x, n_x). Defaults to I.
        B_dynamics: Optional full B matrix (n_x, n_u). Defaults to dt * I.
        C: Optional full C matrix (n_y, n_x). Defaults to I.
        D: Optional full D matrix (n_y, n_u). Defaults to zeros.

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

    Returns:
        LuenbergerObserver instance.
    """
    bk = backend or NumpyBackend()
    cfg = cls.parse_config(config)
    gain = parse_matrix(bk, cfg.observer_gain)
    n = gain.shape[0]
    A = bk.array(cfg.A_dynamics) if cfg.A_dynamics is not None else bk.eye(n)
    if cfg.B_dynamics is not None:
        B = bk.array(cfg.B_dynamics)
    elif cfg.dt is not None:
        B = cfg.dt * bk.eye(n)
    else:
        raise ValueError("LuenbergerObserver: no B_dynamics and no dt — standalone use requires one of them")
    C = bk.array(cfg.C) if cfg.C is not None else bk.eye(n)
    return cls(
        A=A,
        B=B,
        observer_gain=gain,
        C=C,
        D=bk.array(cfg.D) if cfg.D is not None else bk.zeros((C.shape[0], B.shape[1])),
        x0=bk.zeros(n),
        backend=bk,
    )

Estimators¶

KalmanFilter¶

estimate ¶

measure ¶

reset ¶

from_config classmethod ¶

ExtendedKalmanFilter¶

measure ¶

estimate ¶

reset ¶

attach_plant ¶

from_config classmethod ¶

UnscentedKF¶

measure ¶

reset ¶

attach_plant ¶

from_config classmethod ¶

ComplementaryFilter¶

estimate ¶

reset ¶

from_config classmethod ¶

LuenbergerObserver¶

estimate ¶

measure ¶

reset ¶

from_config classmethod ¶

`KalmanFilter`¶

from_config `classmethod` ¶

`ExtendedKalmanFilter`¶

from_config `classmethod` ¶

`UnscentedKF`¶

from_config `classmethod` ¶

`ComplementaryFilter`¶

from_config `classmethod` ¶

`LuenbergerObserver`¶

from_config `classmethod` ¶