Telemetry Transports#
Telemetry is the message layer between IoT field hardware and the rest of the pyCSAMT acquisition workflow. A telemetry packet carries a device id, timestamp, canonical topic, packet kind, QoS flag, retained flag, and a JSON-serialisable payload. The same packet can be sent to MQTT, HTTP, serial, WebSocket, a JSON-lines file, or a dry-run recorder.
Use dry-run clients while developing notebooks, dashboards, and examples: they validate packets and record what would have been sent without opening network or hardware connections. Use the file transport when you want an offline replay log that can be read back later.
Create Canonical Packets#
Start by declaring the field devices. A
pycsamt.iot.DeviceConfig knows the station, acquisition channels,
preferred protocol, and canonical topic layout. Use
pycsamt.iot.TelemetryPacket.from_device() so the topic is built
consistently.
1from pycsamt.iot import DeviceConfig, PacketKind, TelemetryPacket
2
3devices = [
4 DeviceConfig(
5 "l18-node-01",
6 station="L18-001",
7 protocol="mqtt",
8 sample_rate_hz=256.0,
9 channels=["ex", "ey", "hx", "hy"],
10 ),
11 DeviceConfig(
12 "l18-node-02",
13 station="L18-002",
14 protocol="mqtt",
15 sample_rate_hz=256.0,
16 channels=["ex", "ey", "hx", "hy"],
17 ),
18]
19
20packets = [
21 TelemetryPacket.from_device(
22 devices[0],
23 timestamp=1_700_000_000.0,
24 survey_id="WILLY-L18-TELEMETRY",
25 kind=PacketKind.HEALTH,
26 payload={
27 "station": "L18-001",
28 "battery_v": 12.62,
29 "temperature_c": 31.4,
30 "firmware": "iot-demo-0.4.0",
31 },
32 qos=1,
33 ),
34 TelemetryPacket.from_device(
35 devices[0],
36 timestamp=1_700_000_006.0,
37 survey_id="WILLY-L18-TELEMETRY",
38 kind=PacketKind.QC,
39 payload={
40 "station": "L18-001",
41 "method": "amt",
42 "channels": ["ex", "ey", "hx", "hy"],
43 "accepted": True,
44 "decision": "accept",
45 "finite_coverage": 0.994,
46 "frequency_band_hz": [2.0, 512.0],
47 },
48 qos=1,
49 ),
50 TelemetryPacket.from_device(
51 devices[1],
52 timestamp=1_700_000_012.0,
53 survey_id="WILLY-L18-TELEMETRY",
54 kind=PacketKind.HEALTH,
55 payload={
56 "station": "L18-002",
57 "battery_v": 11.38,
58 "temperature_c": 33.1,
59 "firmware": "iot-demo-0.4.0",
60 },
61 qos=1,
62 ),
63 TelemetryPacket.from_device(
64 devices[1],
65 timestamp=1_700_000_018.0,
66 survey_id="WILLY-L18-TELEMETRY",
67 kind=PacketKind.QC,
68 payload={
69 "station": "L18-002",
70 "method": "amt",
71 "channels": ["ex", "ey", "hx", "hy"],
72 "accepted": False,
73 "decision": "reject",
74 "finite_coverage": 0.861,
75 "frequency_band_hz": [2.0, 512.0],
76 "reasons": ["finite_coverage_below_threshold"],
77 },
78 qos=1,
79 ),
80 TelemetryPacket.from_device(
81 devices[1],
82 timestamp=1_700_000_024.0,
83 survey_id="WILLY-L18-TELEMETRY",
84 kind=PacketKind.SYNC,
85 payload={
86 "station": "L18-002",
87 "quality": "fair",
88 "clock_offset_ms": 4.8,
89 "drift_ppm": 2.1,
90 },
91 qos=0,
92 ),
93]
94
95first = packets[0]
96print(f"device_id: {first.device_id}")
97print(f"kind: {first.kind.value}")
98print(f"topic: {first.topic}")
99print(f"qos: {first.qos}")
100print(f"retained: {first.retained}")
101print(f"payload_keys: {', '.join(sorted(first.payload))}")
Output:
device_id: l18-node-01
kind: health
topic: pycsamt/WILLY-L18-TELEMETRY/L18-001/l18-node-01/health
qos: 1
retained: False
payload_keys: battery_v, firmware, station, temperature_c
Record A Dry-Run Transport#
Use pycsamt.iot.build_telemetry_client() for every transport. The
default dry_run=True is deliberate: examples and tests can use MQTT,
HTTP, serial, WebSocket, or LoRa-style names without needing a broker,
port, modem, or optional client library.
1from pycsamt.iot import build_telemetry_client
2
3client = build_telemetry_client(
4 "mqtt",
5 endpoint="mqtt://field-gateway.local:1883",
6 dry_run=True,
7)
8client.subscribe("pycsamt/WILLY-L18-TELEMETRY/+/+/qc")
9
10for packet in packets[:2]:
11 ack = client.send(packet)
12
13print(f"protocol: {ack.protocol}")
14print(f"ack_ok: {ack.ok}")
15print(f"ack_detail: {ack.detail}")
16print(f"recorded_packets: {len(client.sent)}")
17print(f"subscriptions: {', '.join(client.subscriptions)}")
18print(f"healthcheck: {client.healthcheck()}")
Output:
protocol: mqtt
ack_ok: True
ack_detail: dry-run packet recorded
recorded_packets: 2
subscriptions: pycsamt/WILLY-L18-TELEMETRY/+/+/qc
healthcheck: True
Write And Replay JSONL Telemetry#
The file transport is the most useful offline transport. It writes one packet dictionary per line, can be read back by another file client, and does not require any optional dependencies.
1from pathlib import Path
2
3from pycsamt.iot import build_telemetry_client
4
5result_dir = Path("results")
6result_dir.mkdir(exist_ok=True)
7jsonl_path = result_dir / "iot-telemetry-demo.jsonl"
8
9if jsonl_path.exists():
10 jsonl_path.unlink()
11
12with build_telemetry_client(
13 "file",
14 endpoint=str(jsonl_path),
15 dry_run=False,
16 append=False,
17) as file_client:
18 for packet in packets:
19 ack = file_client.send(packet)
20 print(f"protocol: {ack.protocol}")
21 print(f"ack_ok: {ack.ok}")
22 print(f"detail: {ack.detail}")
23 print(f"sent_packets: {len(file_client.sent)}")
24
25reader = build_telemetry_client(
26 "file",
27 endpoint=str(jsonl_path),
28 dry_run=False,
29)
30rows = reader.read_all()
31
32print(f"jsonl_rows: {len(rows)}")
33print(f"first_row_kind: {rows[0]['kind']}")
34print(f"first_row_station: {rows[0]['payload']['station']}")
Output:
protocol: file
ack_ok: True
detail: appended to results\iot-telemetry-demo.jsonl
sent_packets: 5
jsonl_rows: 5
first_row_kind: health
first_row_station: L18-001
Inspect Packet Tables#
Use pycsamt.iot.packet_table() for a flat packet inventory and
pycsamt.iot.telemetry_summary() for counts by device/topic. These
tables are useful before monitoring because they reveal missing packet
kinds, wrong topics, or devices that stopped reporting.
1from pycsamt.iot import packet_table, telemetry_summary
2
3packet_df = packet_table(packets)
4print(
5 packet_df[
6 ["device_id", "kind", "timestamp", "payload_keys"]
7 ].to_string(index=False)
8)
9
10print()
11
12summary = telemetry_summary(packets)
13print(summary[["device_id", "topic", "n_packet"]].to_string(index=False))
Output:
device_id kind timestamp payload_keys
l18-node-01 health 1700000000.0 battery_v;firmware;station;temperature_c
l18-node-01 qc 1700000006.0 accepted;channels;decision;finite_coverage;frequency_band_hz;method;station
l18-node-02 health 1700000012.0 battery_v;firmware;station;temperature_c
l18-node-02 qc 1700000018.0 accepted;channels;decision;finite_coverage;frequency_band_hz;method;reasons;station
l18-node-02 sync 1700000024.0 clock_offset_ms;drift_ppm;quality;station
device_id topic n_packet
l18-node-01 pycsamt/WILLY-L18-TELEMETRY/L18-001/l18-node-01/health 1
l18-node-01 pycsamt/WILLY-L18-TELEMETRY/L18-001/l18-node-01/qc 1
l18-node-02 pycsamt/WILLY-L18-TELEMETRY/L18-002/l18-node-02/health 1
l18-node-02 pycsamt/WILLY-L18-TELEMETRY/L18-002/l18-node-02/qc 1
l18-node-02 pycsamt/WILLY-L18-TELEMETRY/L18-002/l18-node-02/sync 1
Assess Telemetry In A Session#
Add packets to a pycsamt.iot.FieldSession when you want telemetry
to interact with station metadata, monitoring thresholds, and dashboards.
The example below is intentionally synthetic and deterministic: station
L18-002 rejects one QC packet, so the session is marked critical.
1from pycsamt.iot import (
2 FieldSession,
3 MonitoringConfig,
4 StationConfig,
5)
6
7stations = [
8 StationConfig(
9 "L18-001",
10 profile="L18",
11 position_m=0.0,
12 lat=7.501,
13 lon=-5.201,
14 channels=["ex", "ey", "hx", "hy"],
15 ),
16 StationConfig(
17 "L18-002",
18 profile="L18",
19 position_m=50.0,
20 lat=7.502,
21 lon=-5.198,
22 channels=["ex", "ey", "hx", "hy"],
23 ),
24]
25
26session = FieldSession(
27 "WILLY-L18-TELEMETRY",
28 devices=devices,
29 stations=stations,
30 monitoring_config=MonitoringConfig(
31 method="amt",
32 expected_interval_s=6.0,
33 max_gap_s=18.0,
34 min_edge_acceptance_rate=0.75,
35 min_battery_v=11.2,
36 required_channels=["ex", "ey", "hx", "hy"],
37 ),
38)
39session.add_packets(packets)
40status = session.assess(now=1_700_000_030.0)
41
42print(f"level: {status.level.value}")
43print(f"n_packet: {status.n_packet}")
44print(f"packet_success_rate: {status.packet_success_rate:.3f}")
45print(f"edge_acceptance_rate: {status.edge_acceptance_rate:.3f}")
46print(f"max_gap_s: {status.max_gap_s:.1f}")
47print(f"issues: {', '.join(status.issues) or '-'}")
Output:
level: critical
n_packet: 5
packet_success_rate: 1.000
edge_acceptance_rate: 0.500
max_gap_s: 6.0
issues: edge_acceptance_rate_below_threshold
Plot A Telemetry Dashboard#
Use pycsamt.iot.plot_field_dashboard() to visualise the same
session. This figure combines station status, edge-QC acceptance, power
or synchronisation status, and packet timing.
1from pathlib import Path
2
3from pycsamt.iot import plot_field_dashboard
4
5out_dir = Path("docs/source/images/user_guide/iot")
6out_dir.mkdir(parents=True, exist_ok=True)
7
8fig = plot_field_dashboard(
9 session,
10 now=1_700_000_030.0,
11 station_axis="profile",
12 title="Telemetry transport demo: WILLY-L18",
13)
14fig.savefig(
15 out_dir / "user-guide-iot-telemetry-01.png",
16 dpi=180,
17)
Transport Choices#
The transport name controls the client returned by
pycsamt.iot.build_telemetry_client().
Protocol |
Client |
Typical use |
|---|---|---|
|
Offline logging, replay, tests, documentation. |
|
|
Field broker publishing and topic subscriptions. |
|
|
Gateway POST requests to an API endpoint. |
|
|
UART links to local acquisition hardware. |
|
|
Live dashboard streams. |
|
|
Dry-run recorder unless a project-specific transport is added. |
MQTT, serial, and WebSocket use optional third-party libraries only when a
real connection is opened. Keep dry_run=True in notebooks, CI jobs,
and documentation examples unless you intentionally want to contact a
field endpoint.
Surviving an intermittent uplink#
Remote nodes lose their link. Wrap any client in a
StoreAndForwardClient so a failed send queues the
packet instead of dropping it; flush() later drains the backlog in
order, and an optional spool file carries it across a restart:
from pycsamt.iot import StoreAndForwardClient, build_telemetry_client
client = build_telemetry_client("http", endpoint="https://hub/ingest",
dry_run=False)
buffered = StoreAndForwardClient(client, spool_path="spool.jsonl",
max_queue=5000)
buffered.send(packet) # delivered, or queued if the link is down
# ... later, when connectivity returns ...
buffered.flush() # drains the queue in order (at-least-once)
Delivery is at-least-once and order-preserving, max_queue bounds the
buffer with a drop-oldest policy, and next_retry_delay_s provides an
exponential-backoff hint for a scheduling loop.
Field Interpretation#
Telemetry tables answer operational questions before geophysical
processing starts: which node reported, what kind of packet was sent,
which station it belongs to, and whether the payload contains health,
QC, sync, or power information. Monitoring then turns those packets into
a survey status. In this example the packet stream is complete, but the
edge acceptance rate is only 0.500, so the session is flagged
critical even though packet delivery itself is healthy.
For real surveys, keep topic names stable across the acquisition campaign and persist file logs or broker exports with the processing report. That makes later QC, provenance, and troubleshooting much easier to audit.