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)

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)

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)

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>)

Bases: PyCSAMTObject, MetadataMixin

Configuration for one field IoT node or recorder gateway.

Parameters:

• device_id (str)

• station (str | None)

• protocol (str)

• sample_rate_hz (float | None)

• channels (list[str])

• role (DeviceRole | str)

• enabled (bool)

• metadata (dict[str, Any])

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()

Validate and normalise device configuration.

Return type:

None

property n_channels: int

Number of declared acquisition channels.

topic(kind, *, survey_id=None)

Return a canonical telemetry topic for this device.

Parameters:

• kind (PacketKind | str)

• survey_id (str | None)

Return type:

str

as_dict()

Return a serialisable representation.

Return type:

dict[str, Any]

classmethod from_mapping(data)

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)

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)

• kind (PacketKind | str)

• retained (bool)

device_id: str

timestamp: float

topic: str

payload: dict[str, Any]

qos: int = 0

kind: PacketKind | str = 'data'

retained: bool = False

validate()

Validate and normalise packet fields.

Return type:

None

as_dict()

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)

Create a packet using a device’s canonical topic.

Parameters:

• device (DeviceConfig)

• timestamp (float)

• payload (Mapping[str, Any])

• kind (PacketKind | str)

• survey_id (str | None)

• qos (int)

• retained (bool)

Return type:

TelemetryPacket

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

Bases: PyCSAMTObject, MetadataMixin

Survey-level IoT deployment metadata.

Parameters:

• survey_id (str)

• devices (list[DeviceConfig])

• capabilities (list[IoTCapability])

• notes (str)

• metadata (dict[str, Any])

survey_id: str

devices: list[DeviceConfig]

capabilities: list[IoTCapability]

notes: str = ''

metadata: dict[str, Any]

validate()

Validate and normalise deployment fields.

Return type:

None

property n_devices: int

Number of devices in the deployment.

add_device(device)

Append a device and revalidate uniqueness.

Parameters:

device (DeviceConfig | Mapping[str, Any])

Return type:

None

has_capability(capability)

Return whether the deployment declares capability.

Parameters:

capability (IoTCapability | str)

Return type:

bool

device(device_id)

Return a device by id.

Parameters:

device_id (str)

Return type:

DeviceConfig

as_dict()

Return a serialisable deployment dictionary.

Return type:

dict[str, Any]

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

Return one row per device describing declared IoT capabilities.

Parameters:

• deployment (DeploymentConfig)

• api (bool | None)

Return type:

Any