pycsamt.iot.protocols.base#

Telemetry transport base classes and the dry-run recorder.

BaseTelemetryClient defines the transport interface shared by every concrete client (MQTT, HTTP, file, serial, websocket): connect, disconnect, send, receive, subscribe, listen, flush, and healthcheck. Subclasses implement the _transport_* hooks; the base handles validation, dry-run recording, auto-connect, and error wrapping.

Every client supports dry_run=True, in which packets are recorded to client.sent and no network/hardware access occurs. This keeps demos, tests, and documentation fully offline and makes optional dependencies (paho-mqtt, pyserial, websocket-client) truly optional — they are only imported when a real connection is opened.

Classes

BaseTelemetryClient([endpoint, protocol, ...])

Common behaviour for telemetry transports.

IoTProtocol(*values)

Supported telemetry protocol identifiers.

TelemetryAck(ok, protocol, packet_id[, detail])

Acknowledgement returned by a telemetry client.

TelemetryClient([endpoint, protocol, dry_run])

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

Exceptions

TelemetryError

Raised when a telemetry transport operation fails.
class pycsamt.iot.protocols.base.IoTProtocol(*values)[source]#

Bases: str, Enum

Supported telemetry protocol identifiers.

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

Bases: object

Acknowledgement returned by a telemetry client.

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

Bases: RuntimeError

Raised when a telemetry transport operation fails.

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

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()[source]#

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

Return type:

None

disconnect()[source]#

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

Return type:

None

send(packet)[source]#

Send packet, recording it in dry-run mode.

Parameters:

packet (Any)

Return type:

TelemetryAck

receive(*, timeout=None)[source]#

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

Parameters:

timeout (float | None)

Return type:

dict[str, Any] | None

subscribe(topic)[source]#

Register interest in topic (transport permitting).

Parameters:

topic (str)

Return type:

None

listen(callback, *, max_messages=None, timeout=None)[source]#

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()[source]#

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

Return type:

int

healthcheck()[source]#

Return whether the transport is reachable.

Return type:

bool

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

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: