Ingestion#

This module contains functions for loading telemetry data from various sources. Every loader returns the same long-form TelemetryDataset, so downstream preprocessing and detection do not care where the data came from.

Loader for the NASA SMAP / MSL telemetry benchmark (telemanom format).

The benchmark ships one .npy array per channel under train/ and test/ directories, each of shape (timesteps, features) where column 0 is the telemetry value and the remaining columns are one-hot command context. Anomaly labels live in labeled_anomalies.csv and are used for evaluation only, never for training.

SMAP arrays carry no real timestamps, so this loader synthesizes a uniform time index (configurable cadence). Output is the canonical long form [timestamp, variable, value] wrapped in a TelemetryDataset.

telemetry_anomdet.ingest.smap.load_smap_labels(labels_csv: str | Path, *, spacecraft: str | None = 'SMAP') → DataFrame[source]#

Load and parse labeled_anomalies.csv.

Parameters:

labels_csv – Path to the telemanom labeled_anomalies.csv.
spacecraft – Keep only rows for this spacecraft (‘SMAP’ or ‘MSL’). None keeps all rows.

Returns:

The label rows with three added columns:: ’sequences’ (list of (start, end) index tuples), ‘anomaly_span’ (total number of anomalous timesteps), and ‘classes’ (one label per sequence, where the file provides them).

Return type:

pd.DataFrame

Notes

The ‘class’ column labels each sequence ‘point’ or ‘contextual’. A point anomaly is extreme in isolation; a contextual one is individually plausible but wrong given the surrounding state, and is the harder case for a forecaster. Reporting detection rates separately by class is more informative than a single recall, since the two behave differently.

telemetry_anomdet.ingest.smap.anomaly_point_mask(sequences: Sequence[tuple[int, int]], n_timesteps: int) → ndarray[source]#

Build a point-level boolean mask from anomaly index ranges.

Parameters:

sequences – Iterable of inclusive (start, end) index ranges.
n_timesteps – Length of the mask.

Returns:

Boolean array of length n_timesteps, True inside any range.

Return type:

np.ndarray

telemetry_anomdet.ingest.smap.load_smap_channel(npy_path: str | Path, chan_id: str | None = None, *, dims: str | Sequence[int] = 'nonzero', cadence: str = '1s', start: str = '2000-01-01', tz: str = 'UTC') → TelemetryDataset[source]#

Load a single SMAP channel .npy into a long-form TelemetryDataset.

Parameters:

npy_path – Path to the channel array of shape (timesteps, features).
chan_id – Channel name used to prefix variables. Defaults to the file stem.
dims – Which feature columns to keep. ‘nonzero’ drops all-zero columns (the default, matching the command one-hots that are inactive for a channel), ‘all’ keeps every column, ‘telemetry’ keeps only column 0, or pass an explicit list of column indices.
cadence – Synthetic sampling interval (pandas offset alias, e.g. ‘1s’).
start – Synthetic start timestamp for the first sample.
tz – Timezone for the synthesized index.

Returns:

Long form with variables named f"{chan_id}_dim{j}".

Return type:

TelemetryDataset

telemetry_anomdet.ingest.smap.load_smap(data_dir: str | Path, channels: Sequence[str], *, split: str = 'test', dims: str | Sequence[int] = 'nonzero', cadence: str = '1s', start: str = '2000-01-01', tz: str = 'UTC') → TelemetryDataset[source]#

Load several SMAP channels into one combined long-form TelemetryDataset.

Parameters:

data_dir – Directory containing the train/ and test/ subfolders.
channels – Channel ids to load (e.g. [‘A-1’, ‘D-2’]).
split – ‘train’ or ‘test’.
dims – Passed through to load_smap_channel().
cadence – Passed through to load_smap_channel().
start – Passed through to load_smap_channel().
tz – Passed through to load_smap_channel().

Returns:

Combined long form; each channel keeps its own: f"{chan_id}_dim{j}" variables so they never collide.

Return type:

TelemetryDataset

telemetry_anomdet.ingest.csv_loader.load_from_csv(path: str, *, time_col: str | None = None, value_cols: Sequence[str] | None = None) → TelemetryDataset[source]#

Load telemetry from a CSV file.

Notes: CSV must contain at least colums: timestamp, variable, value. timestamp will be converted to pandas datetime.

Arguments: path - Path to CSV file. time_col - Explicit time column, if ommitted, guessed from aliases value_cols - Explicit value columns

Returns: TelemetryDataset

telemetry_anomdet.ingest.csv_loader.is_long_form(cols: Sequence[str], aliases: Mapping[str, Iterable[str]]) → bool[source]#: Determine whether a CSV is already in “long” form by checking that all three semantic roles appear under some alias.

telemetry_anomdet.ingest.csv_loader.coerce_long(df: DataFrame) → DataFrame[source]#: Final cleanup for a long form DataFrame - sort by time, then variable - ensure variable is string - parse timestamp to utc

telemetry_anomdet.ingest.csv_loader.pick_time_column(cols: Sequence[str], *, time_col: str | None, aliases: Mapping[str, Iterable[str]]) → str[source]#: Choose which colun is the time axis for wide CSV. - if time_col is explicitly provided, verify it exists and return it - else, try and match from alias candidates (like: “timestamp”, “time”) - if nothing can be found, raise error listing options and actual columns

telemetry_anomdet.ingest.ccsds_loader.load_from_ccsds(file_or_stream) → TelemetryDataset[source]#: Load telemetry from a CCSDS file or stream.

class telemetry_anomdet.ingest.dataset.TelemetryDataset(_df: DataFrame)[source]#

Container for telemetry data.

classmethod synthetic() → Self[source]#

Generate a small synthetic telemetry dataset for development / testing.

Parameters: N/A for now.

Returns: TelemetryDataset

to_pandas() → DataFrame[source]#: Return a copy of the Dataframe to prevent accidental mutation.

property data: DataFrame#: Direct access to the underlying DataFrame.

head(n: int = 5) → DataFrame[source]#: Head of the underlying DataFrame (copy).