pycsamt.iot.schemas#

Formal telemetry payload schemas.

Raw IoT payloads are free-form dictionaries, which is convenient but fragile: the same quantity may arrive as battery_v, battery_voltage, or voltage. These schemas define a canonical field set per packet kind, absorb common key aliases, and validate ranges, so downstream monitoring and provenance code can rely on stable keys.

Each schema is a light dataclass exposing:

from_mapping(payload) — tolerant parser that pulls known aliases into canonical fields and preserves anything else under extra.
as_dict(drop_none=...) — flat, serialisable payload with canonical keys merged with extra.

Use validate_payload() / parse_payload() to route a payload through the schema registered for a given PacketKind.

Module Attributes

PAYLOAD_SCHEMAS

Map each packet kind to its canonical payload schema.

Functions

`parse_payload`(kind, payload)	Parse payload into the schema registered for kind.
`schema_for`(kind)	Return the payload schema registered for kind.
`validate_payload`(kind, payload, *[, drop_none])	Return a canonicalised payload dictionary for kind.

Classes

`AcquisitionPayload`([kind, station, extra, ...])	Metadata describing one raw acquisition record/window.
`EventPayload`([kind, station, extra, event, ...])	Discrete field event (state change, alarm, operator note).
`EventSeverity`(*values)	Severity levels for event telemetry.
`HealthPayload`([kind, station, extra, ...])	Device-health telemetry (battery, temperature, link quality).
`PowerPayload`([kind, station, extra, ...])	Energy-budget telemetry for a field node.
`QCPayload`([kind, station, extra, accepted, ...])	Edge quality-control telemetry for one acquisition window.
`SourcePayload`([kind, station, extra, ...])	Transmitter telemetry for a controlled-source (CSAMT/TDEM) survey.
`SyncPayload`([kind, station, extra, ...])	Clock-synchronisation telemetry for a field node.
`TelemetryPayload`([kind, station, extra])	Base class for canonical telemetry payloads.

class pycsamt.iot.schemas.EventSeverity(*values)[source]#

Bases: str, Enum

Severity levels for event telemetry.

INFO = 'info'#

WARNING = 'warning'#

ERROR = 'error'#

CRITICAL = 'critical'#

class pycsamt.iot.schemas.TelemetryPayload(kind=PacketKind.DATA, station=None, extra=<factory>)[source]#

Bases: PyCSAMTObject

Base class for canonical telemetry payloads.

Parameters:

kind (PacketKind)
station (str | None)
extra (dict[str, Any])

kind: PacketKind = 'data'#: Canonical packet kind the schema describes. Overridden per subclass.

station: str | None = None#

extra: dict[str, Any]#

as_dict(*, drop_none=False)[source]#

Return a flat payload dictionary with canonical keys.

Parameters:: drop_none (bool)
Return type:: dict[str, Any]

class pycsamt.iot.schemas.HealthPayload(kind=PacketKind.HEALTH, station=None, extra=<factory>, battery_v=None, temperature_c=None, uptime_s=None, free_storage_mb=None, rssi_dbm=None, firmware=None)[source]#

Bases: TelemetryPayload

Device-health telemetry (battery, temperature, link quality).

Parameters:

kind (PacketKind)
station (str | None)
extra (dict[str, Any])
battery_v (float | None)
temperature_c (float | None)
uptime_s (float | None)
free_storage_mb (float | None)
rssi_dbm (float | None)
firmware (str | None)

kind: PacketKind = 'health'#: Canonical packet kind the schema describes. Overridden per subclass.

battery_v: float | None = None#

temperature_c: float | None = None#

uptime_s: float | None = None#

free_storage_mb: float | None = None#

rssi_dbm: float | None = None#

firmware: str | None = None#

validate()[source]#

Validate object state.

Subclasses can override this hook. The base implementation intentionally accepts all states.

Return type:: None

classmethod from_mapping(payload)[source]#

Parameters:: payload (Mapping[str, Any])
Return type:: HealthPayload

as_dict(*, drop_none=False)[source]#

Return a flat payload dictionary with canonical keys.

Parameters:: drop_none (bool)
Return type:: dict[str, Any]

class pycsamt.iot.schemas.QCPayload(kind=PacketKind.QC, station=None, extra=<factory>, accepted=None, decision=None, snr_db=None, finite_coverage=None, spike_fraction=None, rms=None, method=None, channels=<factory>, frequency_band_hz=None, reasons=<factory>)[source]#

Bases: TelemetryPayload

Edge quality-control telemetry for one acquisition window.

Parameters:

kind (PacketKind)
station (str | None)
extra (dict[str, Any])
accepted (bool | None)
decision (str | None)
snr_db (float | None)
finite_coverage (float | None)
spike_fraction (float | None)
rms (float | None)
method (str | None)
channels (list[str])
frequency_band_hz (tuple[float, float] | None)
reasons (list[str])

kind: PacketKind = 'qc'#: Canonical packet kind the schema describes. Overridden per subclass.

accepted: bool | None = None#

decision: str | None = None#

snr_db: float | None = None#

finite_coverage: float | None = None#

spike_fraction: float | None = None#

rms: float | None = None#

method: str | None = None#

channels: list[str]#

frequency_band_hz: tuple[float, float] | None = None#

reasons: list[str]#

validate()[source]#

Validate object state.

Subclasses can override this hook. The base implementation intentionally accepts all states.

Return type:: None

classmethod from_mapping(payload)[source]#

Parameters:: payload (Mapping[str, Any])
Return type:: QCPayload

as_dict(*, drop_none=False)[source]#

Return a flat payload dictionary with canonical keys.

Parameters:: drop_none (bool)
Return type:: dict[str, Any]

class pycsamt.iot.schemas.PowerPayload(kind=PacketKind.POWER, station=None, extra=<factory>, battery_v=None, state=None, runtime_days=None, net_wh_per_day=None, solar_w=None, load_w=None)[source]#

Bases: TelemetryPayload

Energy-budget telemetry for a field node.

Parameters:

kind (PacketKind)
station (str | None)
extra (dict[str, Any])
battery_v (float | None)
state (str | None)
runtime_days (float | None)
net_wh_per_day (float | None)
solar_w (float | None)
load_w (float | None)

kind: PacketKind = 'power'#: Canonical packet kind the schema describes. Overridden per subclass.

battery_v: float | None = None#

state: str | None = None#

runtime_days: float | None = None#

net_wh_per_day: float | None = None#

solar_w: float | None = None#

load_w: float | None = None#

validate()[source]#

Validate object state.

Subclasses can override this hook. The base implementation intentionally accepts all states.

Return type:: None

classmethod from_mapping(payload)[source]#

Parameters:: payload (Mapping[str, Any])
Return type:: PowerPayload

as_dict(*, drop_none=False)[source]#

Return a flat payload dictionary with canonical keys.

Parameters:: drop_none (bool)
Return type:: dict[str, Any]

class pycsamt.iot.schemas.SourcePayload(kind=PacketKind.SOURCE, station=None, extra=<factory>, source_id=None, tx_current_a=None, tx_voltage_v=None, tx_frequency_hz=None, tx_power_w=None, dipole_length_m=None, duty_cycle=None, on=None, offset_m=None, azimuth_deg=None)[source]#

Bases: TelemetryPayload

Transmitter telemetry for a controlled-source (CSAMT/TDEM) survey.

Reported by a transmitter node so the receiver-side QC and provenance know the state of the source that produced each sounding: the injected current and voltage, the frequency currently being transmitted, the grounded-dipole geometry, and the keyed on/off state.

Parameters:

kind (PacketKind)
station (str | None)
extra (dict[str, Any])
source_id (str | None)
tx_current_a (float | None)
tx_voltage_v (float | None)
tx_frequency_hz (float | None)
tx_power_w (float | None)
dipole_length_m (float | None)
duty_cycle (float | None)
on (bool | None)
offset_m (float | None)
azimuth_deg (float | None)

kind: PacketKind = 'source'#: Canonical packet kind the schema describes. Overridden per subclass.

source_id: str | None = None#

tx_current_a: float | None = None#

tx_voltage_v: float | None = None#

tx_frequency_hz: float | None = None#

tx_power_w: float | None = None#

dipole_length_m: float | None = None#

duty_cycle: float | None = None#

on: bool | None = None#

offset_m: float | None = None#

azimuth_deg: float | None = None#

validate()[source]#

Validate object state.

Subclasses can override this hook. The base implementation intentionally accepts all states.

Return type:: None

classmethod from_mapping(payload)[source]#

Parameters:: payload (Mapping[str, Any])
Return type:: SourcePayload

as_dict(*, drop_none=False)[source]#

Return a flat payload dictionary with canonical keys.

Parameters:: drop_none (bool)
Return type:: dict[str, Any]

class pycsamt.iot.schemas.SyncPayload(kind=PacketKind.SYNC, station=None, extra=<factory>, offset_ms=None, drift_ppm=None, jitter_ms=None, gps_lock=None, n_reference_points=None, reference=None, tx_locked=None, tx_sync_offset_ms=None, tx_id=None)[source]#

Bases: TelemetryPayload

Clock-synchronisation telemetry for a field node.

Parameters:

kind (PacketKind)
station (str | None)
extra (dict[str, Any])
offset_ms (float | None)
drift_ppm (float | None)
jitter_ms (float | None)
gps_lock (bool | None)
n_reference_points (int | None)
reference (str | None)
tx_locked (bool | None)
tx_sync_offset_ms (float | None)
tx_id (str | None)

kind: PacketKind = 'sync'#: Canonical packet kind the schema describes. Overridden per subclass.

offset_ms: float | None = None#

drift_ppm: float | None = None#

jitter_ms: float | None = None#

gps_lock: bool | None = None#

n_reference_points: int | None = None#

reference: str | None = None#

tx_locked: bool | None = None#

tx_sync_offset_ms: float | None = None#

tx_id: str | None = None#

validate()[source]#

Validate object state.

Subclasses can override this hook. The base implementation intentionally accepts all states.

Return type:: None

classmethod from_mapping(payload)[source]#

Parameters:: payload (Mapping[str, Any])
Return type:: SyncPayload

as_dict(*, drop_none=False)[source]#

Return a flat payload dictionary with canonical keys.

Parameters:: drop_none (bool)
Return type:: dict[str, Any]

class pycsamt.iot.schemas.EventPayload(kind=PacketKind.EVENT, station=None, extra=<factory>, event=None, severity=EventSeverity.INFO, message=None, code=None)[source]#

Bases: TelemetryPayload

Discrete field event (state change, alarm, operator note).

Parameters:

kind (PacketKind)
station (str | None)
extra (dict[str, Any])
event (str | None)
severity (EventSeverity | str)
message (str | None)
code (str | None)

kind: PacketKind = 'event'#: Canonical packet kind the schema describes. Overridden per subclass.

event: str | None = None#

severity: EventSeverity | str = 'info'#

message: str | None = None#

code: str | None = None#

validate()[source]#

Validate object state.

Subclasses can override this hook. The base implementation intentionally accepts all states.

Return type:: None

classmethod from_mapping(payload)[source]#

Parameters:: payload (Mapping[str, Any])
Return type:: EventPayload

as_dict(*, drop_none=False)[source]#

Return a flat payload dictionary with canonical keys.

Parameters:: drop_none (bool)
Return type:: dict[str, Any]

class pycsamt.iot.schemas.AcquisitionPayload(kind=PacketKind.DATA, station=None, extra=<factory>, method=None, channels=<factory>, sample_rate_hz=None, frequency_hz=None, frequency_band_hz=None, n_samples=None, gain=None, duration_s=None)[source]#

Bases: TelemetryPayload

Metadata describing one raw acquisition record/window.

Parameters:

kind (PacketKind)
station (str | None)
extra (dict[str, Any])
method (str | None)
channels (list[str])
sample_rate_hz (float | None)
frequency_hz (float | None)
frequency_band_hz (tuple[float, float] | None)
n_samples (int | None)
gain (float | None)
duration_s (float | None)

kind: PacketKind = 'data'#: Canonical packet kind the schema describes. Overridden per subclass.

method: str | None = None#

channels: list[str]#

sample_rate_hz: float | None = None#

frequency_hz: float | None = None#

frequency_band_hz: tuple[float, float] | None = None#

n_samples: int | None = None#

gain: float | None = None#

duration_s: float | None = None#

validate()[source]#

Validate object state.

Subclasses can override this hook. The base implementation intentionally accepts all states.

Return type:: None

classmethod from_mapping(payload)[source]#

Parameters:: payload (Mapping[str, Any])
Return type:: AcquisitionPayload

as_dict(*, drop_none=False)[source]#

Return a flat payload dictionary with canonical keys.

Parameters:: drop_none (bool)
Return type:: dict[str, Any]

pycsamt.iot.schemas.PAYLOAD_SCHEMAS: dict[PacketKind, type[TelemetryPayload]] = {PacketKind.DATA: <class 'pycsamt.iot.schemas.AcquisitionPayload'>, PacketKind.EVENT: <class 'pycsamt.iot.schemas.EventPayload'>, PacketKind.HEALTH: <class 'pycsamt.iot.schemas.HealthPayload'>, PacketKind.POWER: <class 'pycsamt.iot.schemas.PowerPayload'>, PacketKind.QC: <class 'pycsamt.iot.schemas.QCPayload'>, PacketKind.SOURCE: <class 'pycsamt.iot.schemas.SourcePayload'>, PacketKind.SYNC: <class 'pycsamt.iot.schemas.SyncPayload'>}#: Map each packet kind to its canonical payload schema.

pycsamt.iot.schemas.schema_for(kind)[source]#

Return the payload schema registered for kind.

Parameters:: kind (PacketKind | str)
Return type:: type[TelemetryPayload]

pycsamt.iot.schemas.parse_payload(kind, payload)[source]#

Parse payload into the schema registered for kind.

Parameters:

kind (PacketKind | str)
payload (Mapping[str, Any])

Return type:

TelemetryPayload

pycsamt.iot.schemas.validate_payload(kind, payload, *, drop_none=True)[source]#

Return a canonicalised payload dictionary for kind.

Aliased keys are folded into canonical names and values are range checked. Unknown keys are preserved. Raises ValueError when a field fails validation.

Parameters:

kind (PacketKind | str)
payload (Mapping[str, Any])
drop_none (bool)

Return type:

dict[str, Any]