pycsamt.iot.protocols#

Telemetry transports for IoT-enabled field acquisition.

This package supersedes the former single-module protocols.py. It keeps the original public names (IoTProtocol, TelemetryAck, TelemetryClient, build_telemetry_client()) and adds concrete clients:

Every client supports dry_run=True for offline recording.

Module Attributes

CLIENT_REGISTRY

Concrete client for each protocol with a real transport implementation.

Functions

build_telemetry_client(protocol, *[, ...])

Construct a telemetry client for protocol.
class pycsamt.iot.protocols.IoTProtocol(*values)#

Bases: str, Enum

Supported telemetry protocol identifiers.

MQTT = 'mqtt'#
HTTP = 'http'#
LORA = 'lora'#
SERIAL = 'serial'#
FILE = 'file'#
WEBSOCKET = 'websocket'#
class pycsamt.iot.protocols.TelemetryAck(ok, protocol, packet_id, detail='')#

Bases: object

Acknowledgement returned by a telemetry client.

Parameters:
ok: bool#
protocol: str#
packet_id: str#
detail: str = ''#
exception pycsamt.iot.protocols.TelemetryError#

Bases: RuntimeError

Raised when a telemetry transport operation fails.

class pycsamt.iot.protocols.BaseTelemetryClient(endpoint=None, *, protocol=None, dry_run=False, **options)#

Bases: object

Common behaviour for telemetry transports.

Parameters:

  • endpoint (str, optional) – Transport-specific destination (broker URL, file path, URL, serial port, …).

  • protocol (IoTProtocol) – The protocol this client implements.

  • dry_run (bool) – When True, packets are recorded to sent and no real transport is opened.

  • **options – Transport-specific options (credentials, timeouts, TLS, …).

protocol: IoTProtocol = 'file'#
options: dict[str, Any]#
sent: list[TelemetryPacket]#
subscriptions: list[str]#
connect()#

Open the transport (no-op in dry-run mode).

Return type:

None

disconnect()#

Close the transport (no-op in dry-run mode).

Return type:

None

send(packet)#

Send packet, recording it in dry-run mode.

Parameters:

packet (Any)

Return type:

TelemetryAck

receive(*, timeout=None)#

Receive one payload, or None in dry-run/unsupported mode.

Parameters:

timeout (float | None)

Return type:

dict[str, Any] | None

subscribe(topic)#

Register interest in topic (transport permitting).

Parameters:

topic (str)

Return type:

None

listen(callback, *, max_messages=None, timeout=None)#

Poll receive(), invoking callback per payload.

Returns the number of payloads dispatched. In dry-run mode this returns 0 immediately.

Parameters:
Return type:

int

flush()#

Flush any buffered packets; returns count flushed (default 0).

Return type:

int

healthcheck()#

Return whether the transport is reachable.

Return type:

bool

class pycsamt.iot.protocols.TelemetryClient(endpoint=None, *, protocol=IoTProtocol.FILE, dry_run=True, **options)#

Bases: BaseTelemetryClient

Generic recorder used for dry-run simulation and unmapped protocols.

Retains the historical dry-run behaviour: with dry_run=True (default) packets are recorded to sent. With dry_run=False it has no real transport and raises NotImplementedError on send, since a concrete client should be used instead.

Parameters:
class pycsamt.iot.protocols.FileTelemetryClient(endpoint=None, *, dry_run=False, append=True, **options)#

Bases: BaseTelemetryClient

Append/replay telemetry packets to a JSON-lines file.

Parameters:
protocol: IoTProtocol = 'file'#
read_all()#

Return every packet dictionary written to the file.

Return type:

list[dict[str, Any]]

class pycsamt.iot.protocols.HTTPTelemetryClient(endpoint=None, *, dry_run=False, timeout=10.0, token=None, headers=None, method='POST', **options)#

Bases: BaseTelemetryClient

POST telemetry packets to an HTTP(S) endpoint.

Parameters:
protocol: IoTProtocol = 'http'#
class pycsamt.iot.protocols.MQTTTelemetryClient(endpoint=None, *, dry_run=False, host=None, port=1883, username=None, password=None, tls=False, keepalive=60, client_id=None, **options)#

Bases: BaseTelemetryClient

Publish/subscribe telemetry over MQTT.

Parameters:

  • endpoint (str | None)

  • dry_run (bool)

  • host (str | None)

  • port (int)

  • username (str | None)

  • password (str | None)

  • tls (bool)

  • keepalive (int)

  • client_id (str | None)

  • options (Any)

protocol: IoTProtocol = 'mqtt'#
class pycsamt.iot.protocols.SerialTelemetryClient(endpoint=None, *, dry_run=False, baudrate=115200, timeout=1.0, **options)#

Bases: BaseTelemetryClient

Send/receive telemetry over a serial port.

Parameters:
protocol: IoTProtocol = 'serial'#
class pycsamt.iot.protocols.StoreAndForwardClient(client, *, spool_path=None, max_queue=None, base_backoff_s=1.0, max_backoff_s=300.0)#

Bases: object

Wrap a telemetry client with a persistent offline send buffer.

Parameters:

  • client (BaseTelemetryClient) – The underlying transport used for real sends.

  • spool_path (str, optional) – Path to a JSON-lines spool file. When given, the queue is persisted on every change and reloaded on construction, so a node keeps its backlog across restarts.

  • max_queue (int, optional) – Maximum number of buffered packets. When the buffer is full the oldest packet is dropped (the newest data is the most valuable for a live survey). None means unbounded.

  • base_backoff_s (float) – Base delay for the exponential backoff hint (see next_retry_delay_s).

  • max_backoff_s (float) – Ceiling for the backoff hint.

property pending: int#

Number of packets currently buffered.

property next_retry_delay_s: float#

Exponential-backoff hint for when to next call flush().

Returns 0.0 when the queue is empty. After consecutive flush failures the delay grows as base * 2**(failures-1), capped at max_backoff_s – a caller’s scheduling loop can honour this without this class ever sleeping itself.

send(packet)#

Send packet, or queue it when the transport is unavailable.

A direct send is attempted only when the buffer is empty, so a backlog is never overtaken by fresh packets. On transport failure the packet is queued and a queued acknowledgement is returned rather than raising.

Parameters:

packet (Any)

Return type:

TelemetryAck

flush()#

Drain the buffer in order; return the number of packets sent.

Sending stops at the first failure, leaving the remaining packets queued for a later attempt (at-least-once, order-preserving).

Return type:

int

class pycsamt.iot.protocols.WebSocketTelemetryClient(endpoint=None, *, dry_run=False, timeout=10.0, **options)#

Bases: BaseTelemetryClient

Send/receive telemetry over a WebSocket connection.

Parameters:
protocol: IoTProtocol = 'websocket'#
pycsamt.iot.protocols.CLIENT_REGISTRY: dict[IoTProtocol, type[BaseTelemetryClient]] = {IoTProtocol.FILE: <class 'pycsamt.iot.protocols.file.FileTelemetryClient'>, IoTProtocol.HTTP: <class 'pycsamt.iot.protocols.http.HTTPTelemetryClient'>, IoTProtocol.MQTT: <class 'pycsamt.iot.protocols.mqtt.MQTTTelemetryClient'>, IoTProtocol.SERIAL: <class 'pycsamt.iot.protocols.serial.SerialTelemetryClient'>, IoTProtocol.WEBSOCKET: <class 'pycsamt.iot.protocols.websocket.WebSocketTelemetryClient'>}#

Concrete client for each protocol with a real transport implementation.

pycsamt.iot.protocols.build_telemetry_client(protocol, *, endpoint=None, dry_run=True, **options)[source]#

Construct a telemetry client for protocol.

Parameters:

  • protocol (str or IoTProtocol) – Transport identifier ("mqtt", "http", "file", "serial", "websocket", or "lora").

  • endpoint (str, optional) – Transport destination (broker URL, file path, HTTP URL, port).

  • dry_run (bool) – When True (default), the returned client records packets without opening a real transport — safe for tests and demos.

  • **options – Transport-specific options forwarded to the client.

Returns:

A concrete client when the protocol has a real transport, else a generic TelemetryClient recorder (e.g. for lora).

Return type:

BaseTelemetryClient