2.32.4.1. pycsamt.iot.core#

Functions

deployment_report(deployment, *[, api])

Return one row per device describing declared IoT capabilities.

Classes

DeploymentConfig(survey_id[, devices, ...])

Survey-level IoT deployment metadata.

DeviceConfig(device_id[, station, protocol, ...])

Configuration for one field IoT node or recorder gateway.

DeviceRole(*values)

Role of an IoT device in a field deployment.

IoTCapability(*values)

IoT capability flags used in deployment reports.

PacketKind(*values)

Telemetry packet categories used by pyCSAMT IoT workflows.

TelemetryPacket(device_id, timestamp, topic, ...)

A timestamped message emitted by an IoT-enabled field node.
class pycsamt.iot.core.IoTCapability(*values)[source]

Bases: str, Enum

IoT capability flags used in deployment reports.

TELEMETRY = 'telemetry'
EDGE_PROCESSING = 'edge_processing'
SYNCHRONIZATION = 'synchronization'
ENERGY_MANAGEMENT = 'energy_management'
HEALTH_MONITORING = 'health_monitoring'
class pycsamt.iot.core.DeviceRole(*values)[source]

Bases: str, Enum

Role of an IoT device in a field deployment.

SENSOR_NODE = 'sensor_node'
RECORDER = 'recorder'
GATEWAY = 'gateway'
BASE_STATION = 'base_station'
REMOTE_REFERENCE = 'remote_reference'
TRANSMITTER = 'transmitter'
class pycsamt.iot.core.PacketKind(*values)[source]

Bases: str, Enum

Telemetry packet categories used by pyCSAMT IoT workflows.

DATA = 'data'
QC = 'qc'
HEALTH = 'health'
SYNC = 'sync'
POWER = 'power'
EVENT = 'event'
SOURCE = 'source'
class pycsamt.iot.core.DeviceConfig(device_id, station=None, protocol='mqtt', sample_rate_hz=None, channels=<factory>, role=DeviceRole.SENSOR_NODE, enabled=True, metadata=<factory>)[source]

Bases: PyCSAMTObject, MetadataMixin

Configuration for one field IoT node or recorder gateway.

Parameters:
device_id: str
station: str | None = None
protocol: str = 'mqtt'
sample_rate_hz: float | None = None
channels: list[str]
role: DeviceRole | str = 'sensor_node'
enabled: bool = True
metadata: dict[str, Any]
validate()[source]

Validate and normalise device configuration.

Return type:

None

property n_channels: int[source]

Number of declared acquisition channels.

topic(kind, *, survey_id=None)[source]

Return a canonical telemetry topic for this device.

Parameters:
Return type:

str

as_dict()[source]

Return a serialisable representation.

Return type:

dict[str, Any]

classmethod from_mapping(data)[source]

Build a device config from a mapping.

Parameters:

data (Mapping[str, Any])

Return type:

DeviceConfig

class pycsamt.iot.core.TelemetryPacket(device_id, timestamp, topic, payload, qos=0, kind=PacketKind.DATA, retained=False)[source]

Bases: PyCSAMTObject

A timestamped message emitted by an IoT-enabled field node.

Parameters:
device_id: str
timestamp: float
topic: str
payload: dict[str, Any]
qos: int = 0
kind: PacketKind | str = 'data'
retained: bool = False
validate()[source]

Validate and normalise packet fields.

Return type:

None

as_dict()[source]

Return a flat packet dictionary suitable for logging.

Return type:

dict[str, Any]

classmethod from_device(device, *, timestamp, payload, kind=PacketKind.DATA, survey_id=None, qos=0, retained=False)[source]

Create a packet using a device’s canonical topic.

Parameters:
Return type:

TelemetryPacket

class pycsamt.iot.core.DeploymentConfig(survey_id, devices=<factory>, capabilities=<factory>, notes='', metadata=<factory>)[source]

Bases: PyCSAMTObject, MetadataMixin

Survey-level IoT deployment metadata.

Parameters:
survey_id: str
devices: list[DeviceConfig]
capabilities: list[IoTCapability]
notes: str = ''
metadata: dict[str, Any]
validate()[source]

Validate and normalise deployment fields.

Return type:

None

property n_devices: int[source]

Number of devices in the deployment.

add_device(device)[source]

Append a device and revalidate uniqueness.

Parameters:

device (DeviceConfig | Mapping[str, Any])

Return type:

None

has_capability(capability)[source]

Return whether the deployment declares capability.

Parameters:

capability (IoTCapability | str)

Return type:

bool

device(device_id)[source]

Return a device by id.

Parameters:

device_id (str)

Return type:

DeviceConfig

as_dict()[source]

Return a serialisable deployment dictionary.

Return type:

dict[str, Any]

pycsamt.iot.core.deployment_report(deployment, *, api=None)[source]

Return one row per device describing declared IoT capabilities.

Parameters:
Return type:

Any