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) 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 column vectors \((n, 1)\) throughout (not flat \((n,)\)).

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, 1). 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], 1)) if x0 is None else 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, 1) from sensors.	required
`control_input`		Control vector (n_u, 1) applied at this step.	required

Returns:

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

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, 1) from sensors.
        control_input: Control vector (n_u, 1) applied at this step.

    Returns:
        Posterior state estimate :math:`\\hat{x}` (n_x, 1).
    """
    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

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, 1). 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, 1). Defaults to zeros.
    """
    self.x_hat = self.bk.zeros((self.A.shape[0], 1)) if x0 is None else 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, 1)),
        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 column vectors \((n, 1)\) throughout (not flat \((n,)\)).

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, 1). 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], 1)) if x0 is None else 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, 1).	required
`control_input`		Control input \(u_k\) (n_u, 1).	required

Returns:

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

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, 1).
        control_input: Control input :math:`u_k` (n_u, 1).

    Returns:
        Updated state estimate :math:`\\hat{x}_{k+1}` (n_x, 1).
    """
    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

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, 1). 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, 1). Defaults to zeros.
    """
    self.x_hat = self.bk.zeros((self.A.shape[0], 1)) if x0 is None else 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")
    return cls(
        A=A,
        B=B,
        observer_gain=gain,
        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, B.shape[1])),
        x0=bk.zeros((n, 1)),
        backend=bk,
    )

Estimators¶

KalmanFilter¶

estimate ¶

reset ¶

from_config classmethod ¶

LuenbergerObserver¶

estimate ¶

reset ¶

from_config classmethod ¶

`KalmanFilter`¶

from_config `classmethod` ¶

`LuenbergerObserver`¶

from_config `classmethod` ¶