Ingestion

This module contains functions for loading telemetry data from various sources.

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

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

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

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

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

Load telemetry from a CCSDS file or stream.

class telemetry_anomdet.ingest.dataset.TelemetryDataset(_df: DataFrame)

Container for telemetry data.

classmethod synthetic() → Self

Generate a small synthetic telemetry dataset for development / testing.

Parameters: N/A for now.

Returns: TelemetryDataset

to_pandas() → DataFrame

Return a copy of the Dataframe to prevent accidental mutation.

property data: DataFrame

Direct access to the underlying DataFrame.

head(n: int = 5) → DataFrame

Head of the underlying DataFrame (copy).