Deep Models#

Deep detectors inherit from BaseDetector and share the same fit / decision_function / predict / is_anomaly API as the classical models, so they can be stacked or swapped without glue.

These models require torch, which is an optional dependency:

uv add "telemetry-anomdet[deep]"

GDN#

Graph deviation network. Learns a sensor embedding, builds a top-k relational graph over channels, and forecasts each channel from its graph neighbours. The anomaly score is the deviation between forecast and observation, normalised per node by the training median and IQR.

score_channels restricts which channels may contribute to that deviation. Channels excluded from the score still feed the model as context. This matters on data that mixes continuous sensors with discrete mode and status flags, where a command switch is a state change rather than a fault.

class telemetry_anomdet.models.deep.gdn.GDN(embed_dim: int = 64, topk: int = 15, epochs: int = 30, batch_size: int = 64, lr: float = 0.001, scale: bool = True, device: str | None = None, random_state: int | None = None, percentile: float = 95.0, smoothing: float | None = None, score_channels: Sequence[int] | None = None)[source]#

Bases: BaseDetector

Graph Deviation Network detector.

Parameters:

  • embed_dim (int, default=64) – Dimensionality of the learned per-sensor embeddings and hidden features.

  • topk (int, default=15) – Number of graph neighbours retained per sensor. Clamped internally to n_features - 1.

  • epochs (int, default=30) – Number of training epochs.

  • batch_size (int, default=64) – Minibatch size for training.

  • lr (float, default=1e-3) – Adam learning rate.

  • scale (bool, default=True) – Standardise each channel (per-feature z-score) before training and scoring. Recommended: unlike the classical detectors, the network has no implicit scale invariance, so unscaled telemetry lets large-magnitude channels dominate the forecast. The scaler is fitted on the training windows only and reused at inference, so no test statistics leak in.

  • device (str or None, default=None) – Torch device string (e.g. “cuda”, “cpu”). If None, uses CUDA when available, otherwise CPU.

  • random_state (int or None, default=None) – Seed for torch and numpy RNGs, for reproducible training.

  • percentile (float, default=95.0) – Percentile of training deviation scores used to set threshold_.

  • score_channels (sequence of int, optional) –

    Which feature channels may raise an alarm. Every channel still feeds the graph and the forecast; this restricts only the deviation score. Defaults to all of them.

    The distinction matters whenever a record mixes continuous sensors with discrete status, mode or command channels. A discrete channel produces a large forecast error every time it switches, which is a state change rather than a fault, so letting it into the score raises an alarm on normal operation. Such channels are still worth feeding to the model, because they carry context that improves the forecast.

    Choosing them well can outweigh any thresholding decision; see the anomaly scoring page of the documentation for the measured effect.

  • fit) (Attributes (set after)

  • --------------------------

  • decision_scores (np.ndarray, shape (n_windows,)) – Graph deviation scores on the training data.

  • threshold (float) – Default anomaly cutoff derived from training scores at percentile.

  • labels (np.ndarray, shape (n_windows,)) – Binary anomaly labels on training data. 0 = normal, 1 = anomaly.

  • net (GDNNet) – The fitted torch forecasting network.

  • n_nodes (int) – Number of sensor channels seen during fit.

  • window (int) – Forecasting context length (window_size - 1).

  • scaler (sklearn.preprocessing.StandardScaler or None) – Fitted per-channel scaler when scale = True, otherwise None.

Notes

The internal torch network is kept deliberately separate from this wrapper (see _net.GDNNet) so it can be consumed on its own by later explainability or symbolic-distillation work without the detector plumbing.

static ewma(errors: ndarray, alpha: float) → ndarray[source]#

Exponentially weighted moving average down the window (time) axis.

s_t = alpha * e_t + (1 - alpha) * s_{t-1}, seeded with s_0 = e_0 and applied independently per node. Smaller alpha means heavier smoothing.

The recursion only looks backwards, so no future information reaches a window’s score and the training statistics stay leakage free. Smoothing suppresses single-window spikes, which are the dominant source of isolated false alarms in a forecasting detector.

Parameters:

  • errors (np.ndarray, shape (n_windows, n_nodes)) – Per-window, per-node errors, ordered in time.

  • alpha (float) – Smoothing factor in (0, 1]. A value of 1.0 is the identity.

Returns:

Smoothed errors, same shape as the input.

Return type:

np.ndarray

fit(X: ndarray, y: ndarray | None = None) → GDN[source]#

Fit the GDN forecasting network on nominal telemetry windows.

Parameters:

  • X (np.ndarray, shape (n_windows, window_size, n_features)) – Windowed telemetry tensor from windowify(). window_size must be at least 2 (one context step plus one target step).

  • y (ignored) – Present for API consistency.

Returns:

self

Return type:

GDN

decision_function(X: ndarray) → ndarray[source]#

Compute the graph deviation score for each window.

Parameters:

X (np.ndarray, shape (n_windows, window_size, n_features))

Returns:

scores – Graph deviation scores. Higher = more anomalous.

Return type:

np.ndarray, shape (n_windows,)

KANGDN#

GDN with the multilayer perceptrons replaced by Kolmogorov-Arnold layers, whose learned univariate splines are what make the distillation and C generation below possible.

class telemetry_anomdet.models.deep.kan_gdn.KANGDN(embed_dim: int = 64, topk: int = 15, epochs: int = 30, batch_size: int = 64, lr: float = 0.001, scale: bool = True, device: str | None = None, random_state: int | None = None, percentile: float = 95.0, grid_size: int = 5, spline_order: int = 3, smoothing: float | None = None, score_channels=None)[source]#

Bases: GDN

KAN-GAT Graph Deviation Network detector.

Inherits GDN’s full pipeline (input scaling, training loop, deviation scoring, thresholding) and only swaps the network architecture: the GATEncoder gets a KAN activation and the forecast head is a KAN layer.

Parameters:

  • embed_dim – Same as GDN.

  • topk – Same as GDN.

  • epochs – Same as GDN.

  • batch_size – Same as GDN.

  • lr – Same as GDN.

  • scale – Same as GDN.

  • device – Same as GDN.

  • random_state – Same as GDN.

  • percentile – Same as GDN.

  • grid_size (int, default=5) – Number of spline grid intervals in the KAN layers. Larger = more flexible edge functions (and more coefficients to distil).

  • spline_order (int, default=3) – B-spline order for the KAN layers (3 = cubic).

  • fit) (Attributes (set after)

  • --------------------------

  • GDN (Same as)

  • KANGDNNet. (plus net is a)

Distillation#

Extracts a fitted KANGDN into plain NumPy: spline coefficients, the frozen adjacency, and the per-node normalisation constants. The result evaluates without torch, which is the intermediate step toward the flight artifact.

Symbolic distillation of a fitted KANGDN into a torch-free evaluator.

A fitted KANLayer is already a closed-form function: every edge is phi_ij(x) = w_base * silu(x) + sum_k theta_ijk * B_k(x) (a SiLU term plus a B-spline, both exact). This module pulls those learned coefficients out of the torch modules into plain dicts and rebuilds the forward pass in pure NumPy, no torch at inference.

The NumPy evaluator is the portable, dependency-free form of the detector, and is the intended source for a C port onto a microcontroller. The equivalence tests (NumPy output == torch output) are the guarantee that the distilled artifact behaves exactly like the trained model.

Two levels are provided:

per layer

extract_kan_layer() / KANLayerNumpy, one KAN layer, plus KANLayerNumpy.edge_function() for the atomic 1-D edge phi_ij(x) that later feeds symbolic regression.

whole detector

extract_kan_gdn() / KANGDNNumpy, the entire scoring path of a fitted KANGDN: per-channel scaler, GAT encoder (frozen graph, learned attention), both KAN layers, the deviation normalisation and the threshold. This is the deployable artifact, containing everything needed to turn a window of telemetry into an anomaly flag.

Two things collapse at extraction time and never need recomputing on device:

  • The graph is frozen. topk_graph depends only on the learned embeddings, so the top-k adjacency is resolved once here and stored as a boolean matrix. No cosine similarities or top-k sorts at inference.

  • Attention factorises. The scores are a^T [g_i | g_j], and a splits into halves, so score_ij = a_src . g_i + a_dst . g_j is an outer sum of two length-n_nodes vectors. The torch code materialises an (n, n, 4 * embed_dim) tensor; the distilled form never does.

This module imports only NumPy; it reads fitted torch modules via their tensors but never imports torch, so the distilled path is torch-free.

telemetry_anomdet.models.deep.distill.extract_kan_layer(layer) → dict[source]#

Pull a fitted KANLayer’s learned parameters into a torch-free dict.

Parameters:

layer (KANLayer) – A fitted KAN layer (its grid, base_weight, spline_weight are read and detached to NumPy).

Returns:

in_features, out_features, spline_order (ints) and grid, base_weight, spline_weight (NumPy arrays). Fully describes the layer’s function; can be JSON/NPZ-serialised for deployment.

Return type:

dict

class telemetry_anomdet.models.deep.distill.KANLayerNumpy(extracted: dict)[source]#

Bases: object

Torch-free evaluator of a distilled KAN layer.

Reconstructed from extract_kan_layer(), this reproduces the torch KANLayer forward pass using only NumPy, giving the portable form suitable for a C port. forward accepts (..., in_features) and returns (..., out_features).

forward(x: ndarray) → ndarray[source]#
edge_function(out_idx: int, in_idx: int)[source]#

Return the single 1-D edge function phi_ij(x) as a NumPy callable.

This is the atomic unit of KAN distillation: each edge is a function of one input, ready to sample, plot, or hand to symbolic regression.

Parameters:

  • out_idx (int) – Output and input indices selecting the edge.

  • in_idx (int) – Output and input indices selecting the edge.

Returns:

phi(x) for scalar or array x, where the layer output y[out_idx] = sum_in phi(x[in_idx]).

Return type:

callable

telemetry_anomdet.models.deep.distill.extract_kan_gdn_net(net) → dict[source]#

Pull a fitted KANGDNNet’s learned parameters into a torch-free dict.

Resolves the learned top-k graph and splits the attention vector into its source and target halves, so the distilled form has no graph construction and no (n, n, 4 * embed_dim) intermediate left to compute.

Parameters:

net (KANGDNNet) – A fitted KAN-GAT forecasting network.

Returns:

n_nodes, window, embed_dim (ints); embedding, feat_weight, feat_bias, attn_src, attn_dst, adj (NumPy arrays); leaky_slope (float); and activation / out, the two extracted KAN layers.

Return type:

dict

class telemetry_anomdet.models.deep.distill.KANGDNNetNumpy(extracted: dict)[source]#

Bases: object

Torch-free evaluator of a distilled KANGDNNet.

Reconstructed from extract_kan_gdn_net(), this reproduces the whole forecasting network (graph attention encoder, KAN activation, embedding gate, KAN forecast head) using only NumPy.

forward accepts (batch, n_nodes, window) and returns the one-step forecast (batch, n_nodes), matching KANGDNNet.forward.

forward(x: ndarray) → ndarray[source]#
telemetry_anomdet.models.deep.distill.extract_kan_gdn(detector) → dict[source]#

Distil a fitted KANGDN into a torch-free specification of its complete scoring path.

Parameters:

detector (KANGDN) – A fitted detector (fit must have been called).

Returns:

net (see extract_kan_gdn_net()); scaler_mean / scaler_scale (NumPy arrays, or None when scale=False); err_median / err_iqr (per-node deviation normalisation); threshold (float). Every array is NumPy, so the whole dict is NPZ-serialisable and is what the C export will read.

Return type:

dict

Raises:

RuntimeError – If the detector has not been fitted.

class telemetry_anomdet.models.deep.distill.KANGDNNumpy(extracted: dict)[source]#

Bases: object

Torch-free evaluator of a distilled KANGDN detector.

The deployable artifact: takes raw telemetry windows and returns deviation scores and anomaly flags identical to the fitted torch detector, using only NumPy. Mirrors the detector’s decision_function / predict.

Parameters:

extracted (dict) – Output of extract_kan_gdn().

Notes

Input is (n_windows, window_size, n_features), the same 3-D window tensor the detector is fitted on. Scaling, the context/target split, the forecast, the per-node deviation normalisation and the threshold comparison all happen here, so nothing outside this class is needed at inference.

forecast_errors(X: ndarray) → ndarray[source]#

Per-window, per-node absolute forecast error, shape (n_windows, n_nodes).

Applies the detector’s EWMA smoothing when it was fitted with any, so the distilled scores match the torch detector’s. Note that smoothing makes the score path stateful: a window’s score depends on the windows before it, and scoring a batch is not the same as scoring its rows separately.

decision_function(X: ndarray) → ndarray[source]#

Graph deviation score per window (higher = more anomalous).

Each node’s forecast error is normalised by its training median and IQR; the score is the maximum across nodes, so the sensor deviating most drives the window.

predict(X: ndarray) → ndarray[source]#

Binary anomaly labels (1 = anomaly) from the distilled threshold.

C Code Generation#

Emits a distilled detector as Power of Ten conformant C with no dynamic allocation, alongside golden vectors for host-side conformance testing. See Onboard Deployment Architecture for the target hardware, the resource budget, and how responsibility is split between ground and flight.

C code generation for a distilled KANGDN.

Takes the torch-free specification produced by extract_kan_gdn() and emits a self-contained C translation unit: every learned parameter becomes a static const array and the scoring path becomes ordinary loops. The result has no dependencies beyond math.h, allocates nothing, and exposes two entry points, so it ports to flight hardware without modification.

The generated code mirrors KANGDNNumpy statement for statement, which is what makes it checkable: the same window scored by the torch detector, the NumPy evaluator and the compiled C must agree.

Three properties of the distilled form are exploited:

  • The graph is frozen, so neighbours are emitted as explicit index lists and the attention softmax runs over topk + 1 terms per node rather than over all nodes.

  • Attention is separable, so each node needs two dot products (a_src . g_i and a_dst . g_i) instead of a pairwise concatenation.

  • The spline knots are fixed, so every Cox-de Boor denominator is known at generation time. They are tabulated per recursion level and index, which turns each division in the inner loop into a multiply and keeps the result exact for any knot vector rather than only for evenly spaced ones.

Coding standard#

The output follows the NASA/JPL Power of Ten rules for safety critical code:

  1. No goto, setjmp, or recursion.

  2. Every loop bound is a compile-time constant.

  3. No allocation. Scratch buffers are function-scope static.

  4. Functions stay short; the scoring path is split into single-purpose steps.

  5. At least two assertions per function, routed through KANGDN_ASSERT.

  6. Data is declared at the smallest workable scope; no file-scope mutable state.

  7. Every function returns a status code and validates its parameters, and every call site checks the returned status.

  8. The preprocessor is limited to include guards, the interface dimensions, and the assertion hook. Knot vectors are static const data rather than function-like macros, and internal dimensions are enumerations.

  9. Pointer use is limited to a single dereference. No function pointers.

  10. The output is intended to compile clean under -Wall -Wextra -Werror -pedantic.

Integrators should define KANGDN_ASSERT to route failures into the mission’s fault handler. The default delegates to <assert.h>, which aborts, and is appropriate for ground testing rather than flight.

Numeric type is selectable. float matches the single-precision FPUs on the relevant targets and is the deployment mode; double reproduces the NumPy evaluator to near machine precision and exists to separate porting errors from precision loss.

telemetry_anomdet.models.deep.codegen.generate_c(extracted: dict, prefix: str = 'kangdn', dtype: str = 'float', test_windows=None) → dict[str, str][source]#

Generate a self-contained C implementation of a distilled KANGDN.

Parameters:

  • extracted (dict) – Output of extract_kan_gdn().

  • prefix (str, default="kangdn") – Identifier prefix for the emitted symbols, macros and file names.

  • dtype ({"float", "double"}, default="float") – C type for parameters and arithmetic. float is the deployment mode; double reproduces the NumPy evaluator to near machine precision.

  • test_windows (np.ndarray, optional) – Windows of shape (n, window_size, n_features). When given, a third file of golden vectors is emitted: each window flattened as the entry point expects, beside the reference evaluator’s score for it. A port is then validated on the target itself, by scoring each window and comparing, with no host tooling in the loop.

Returns:

{f"{prefix}.h": ..., f"{prefix}.c": ...}, source text ready to write, plus {prefix}_vectors.h when test_windows is given.

Return type:

dict

Raises:

  • NotImplementedError – If a KAN layer’s knot vector differs between input features, if neighbour counts vary between nodes, or if the model has more nodes than a byte index can address.

  • ValueError – If dtype is unrecognised or a parameter is not finite.

telemetry_anomdet.models.deep.codegen.write_c(extracted: dict, out_dir, prefix: str = 'kangdn', dtype: str = 'float', test_windows=None) → list[source]#

Generate the C sources and write them into out_dir.

Returns the list of written paths. See generate_c() for the parameters.