pycsamt.iot.monitoring#

Functions

`assess_telemetry`(packets, *[, config, now])	Convenience wrapper around `TelemetryMonitor`.
`monitoring_status_table`(statuses, *[, api])	Return monitoring statuses as a pyCSAMT table.
`packet_table`(packets, *[, api])	Return telemetry packets as a pyCSAMT table.
`telemetry_summary`(packets, *[, api])	Summarise telemetry packet counts by device and topic.

Classes

`EMMethod`(*values)	Electromagnetic survey methods recognised by IoT monitoring.
`MonitoringConfig`([method, ...])	Thresholds used to monitor AMT/MT/CSAMT telemetry streams.
`MonitoringLevel`(*values)	Overall status level for a monitored telemetry stream.
`MonitoringStatus`(level, n_packet, ...[, ...])	Status returned by `TelemetryMonitor`.
`TelemetryMonitor`([config])	Monitor field telemetry from AMT, MT, CSAMT, and related surveys.

class pycsamt.iot.monitoring.EMMethod(*values)[source]#

Bases: str, Enum

Electromagnetic survey methods recognised by IoT monitoring.

AMT = 'amt'#

MT = 'mt'#

CSAMT = 'csamt'#

CSEM = 'csem'#

TDEM = 'tdem'#

TEM = 'tem'#

UNKNOWN = 'unknown'#

class pycsamt.iot.monitoring.MonitoringConfig(method=EMMethod.UNKNOWN, expected_interval_s=None, max_latency_s=30.0, max_gap_s=None, min_packet_success_rate=0.95, min_edge_acceptance_rate=0.85, min_battery_v=None, max_clock_offset_ms=None, frequency_band_hz=None, required_channels=<factory>, metadata=<factory>)[source]#

Bases: PyCSAMTObject, MetadataMixin

Thresholds used to monitor AMT/MT/CSAMT telemetry streams.

Parameters:

method (EMMethod | str)
expected_interval_s (float | None)
max_latency_s (float | None)
max_gap_s (float | None)
min_packet_success_rate (float)
min_edge_acceptance_rate (float)
min_battery_v (float | None)
max_clock_offset_ms (float | None)
frequency_band_hz (tuple[float, float] | None)
required_channels (list[str])
metadata (dict[str, Any])

method: EMMethod | str = 'unknown'#

expected_interval_s: float | None = None#

max_latency_s: float | None = 30.0#

max_gap_s: float | None = None#

min_packet_success_rate: float = 0.95#

min_edge_acceptance_rate: float = 0.85#

min_battery_v: float | None = None#

max_clock_offset_ms: float | None = None#

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

required_channels: list[str]#

metadata: dict[str, Any]#

classmethod for_method(method, **overrides)[source]#

Build a config seeded with a method’s canonical defaults.

The expected frequency band and required channels are taken from the method’s MethodProfile, so the method-mismatch, missing-channel, and out-of-band checks in TelemetryMonitor become method-aware without any manual threshold tuning. Any keyword overrides win over the defaults.

Example

>>> cfg = MonitoringConfig.for_method("csamt", min_battery_v=11.5)
>>> cfg.required_channels
['ex', 'ey', 'hx', 'hy']

Parameters:

method (EMMethod | str)
overrides (Any)

Return type:

MonitoringConfig

validate()[source]#

Validate and normalise monitoring thresholds.

Return type:: None

class pycsamt.iot.monitoring.MonitoringLevel(*values)[source]#

Bases: str, Enum

Overall status level for a monitored telemetry stream.

OK = 'ok'#

WARNING = 'warning'#

CRITICAL = 'critical'#

NO_DATA = 'no_data'#

class pycsamt.iot.monitoring.MonitoringStatus(level, n_packet, packet_success_rate, edge_acceptance_rate, mean_latency_s, max_gap_s, battery_min_v, clock_offset_max_ms, methods=<factory>, stations=<factory>, channels=<factory>, frequency_min_hz=nan, frequency_max_hz=nan, issues=<factory>)[source]#