pycsamt.iot#

IoT field telemetry, edge processing, synchronisation, power budgeting, security, provenance, and simulation helpers for AMT, MT, CSAMT, and CSEM deployments.

The pycsamt.iot package is the public API for IoT-enabled field acquisition. It is intentionally separate from the EDI/impedance science API: IoT objects describe what happened in the field, what packets were sent, what edge decisions were made, how clocks and power behaved, and how the acquisition can be audited. Processed transfer functions still flow through the normal pycsamt.emtools, pycsamt.site, and pycsamt.z layers.

Implemented Capabilities#

The package implements concrete capabilities that can be reported in a methods section:

  • device and station metadata, including station coordinates, line position, electric dipole geometry, channels, and recorder IDs;

  • canonical telemetry packets and payload schemas for data, QC, health, power, synchronisation, event, and acquisition messages;

  • dry-run and real telemetry clients for file, HTTP(S), MQTT, serial, and WebSocket transports, with a generic recorder fallback for protocols such as LoRa, and a store-and-forward wrapper that buffers packets (with an optional on-disk spool) across an intermittent uplink;

  • generic edge QC: decimation, finite coverage, RMS/statistics, robust spike fraction, per-channel accept/reject decisions, and QC telemetry packet export;

  • AMT/CSAMT edge QC: powerline harmonic detection, channel SNR, saturation, contact-resistance proxy checks, frequency coverage, live spectra, impedance stability, and sensor dropout checks;

  • CSAMT/CSEM controlled-source edge QC: skin-depth near/transition/far field-zone classification, transmitter frequency-comb detection, source current/voltage stability, and CSEM magnitude/phase-versus-offset (MVO/PVO) analysis, plus a transmitter device role, a SourcePayload transmitter telemetry schema, and a transmitter timing-lock extension on the sync payload;

  • a galvanic static-shift indicator that separates a frequency-independent resistivity split (static shift) from anisotropy;

  • a data-model bridge that turns edge impedance estimates into a pycsamt.z.z.Z, writes a preliminary EDIFile, promotes a session into EDI-backed sites for the processing pipeline, routes those sites through pycsamt.emtools QC so field and post-processing QC agree, and seeds a re-occupation-ready FieldSession or DeploymentConfig from an existing EDI survey;

  • method-aware QC: per-method acquisition profiles (frequency band, required channels, controlled-source and powerline-sensitivity flags) that drive monitoring thresholds and gate powerline detection;

  • telemetry stream monitoring for AMT, MT, CSAMT, TEM, and TDEM, including packet success, edge acceptance, latency, packet gaps, clock offset, battery voltage, required channels, and frequency-band checks;

  • power-budget estimation with duty cycle, battery reserve, regulator and charge efficiency, solar harvesting, telemetry load, edge-processing load, auxiliary load, status classification, and power packet export;

  • clock synchronisation audit: offset, drift, jitter, GPS lock, quality grade, batch status tables, and GPS-dropout detection;

  • acquisition provenance: station audit records, QC decision logs, raw-file hashes, manifests, and reproducibility bundles, with optional HMAC manifest signing and a tamper-evident QC-decision hash chain;

  • transport security configuration with TLS options, credential schemes, and secret redaction; and

  • reproducible synthetic field-network simulation for documentation, tests, and demonstrations without hardware.

  • acquisition visualisation with a compact field dashboard for station health, edge-QC acceptance, power/synchronisation status, and telemetry packet timing, plus a dedicated edge-QC summary for channel coverage, spike fraction, decisions, rejection reasons, and a power-budget summary for daily load, harvest, runtime, and energy-management state, and a synchronisation-quality summary for offset, drift, jitter, GPS lock, and quality grades.

Minimal Example#

import numpy as np
from pycsamt.iot import (
    DeviceConfig,
    EdgeProcessingConfig,
    EdgeProcessor,
    FieldSession,
    MonitoringConfig,
)

device = DeviceConfig(
    "node-1",
    station="S01",
    protocol="mqtt",
    sample_rate_hz=256.0,
    channels=["ex", "hy"],
)

edge = EdgeProcessor(
    EdgeProcessingConfig(
        finite_threshold=0.95,
        channel_names=["ex", "hy"],
        spike_threshold=6.0,
    )
)
result = edge.process(np.random.randn(1024, 2))
packet = result.to_packet(device, timestamp=1_700_000_000.0,
                          survey_id="survey-a")

session = FieldSession(
    "survey-a",
    devices=[device],
    monitoring_config=MonitoringConfig(
        method="amt",
        required_channels=["ex", "hy"],
        min_edge_acceptance_rate=0.85,
    ),
)
session.add_packet(packet)
status = session.assess()

Packet and Table API#

Most functions return plain Python objects or pandas-like tables wrapped by the pyCSAMT API view layer when requested. Common reporting helpers include:

IoT helpers for pyCSAMT field telemetry and edge workflows.

This subpackage supports IoT-enabled AMT/CSAMT field acquisition:

pycsamt.iot.build_edifile(z, *, station, lat=None, lon=None, elevation=None, method=None, acqby='pycsamt.iot', metadata=None)#

Assemble an in-memory EDIFile from an impedance tensor.

The container carries the minimum valid SEG-EDI structure — a >HEAD block (station id and geometry), a >=MTSECT measurement section, and the impedance transfer function — so it round-trips through pycsamt.seg.edi.EDIFile.write().

Parameters:

  • z (pycsamt.z.z.Z) – Impedance tensor for one station.

  • station (str) – Station / DATAID label.

  • lat (float, optional) – Station location written to the EDI head.

  • lon (float, optional) – Station location written to the EDI head.

  • elevation (float, optional) – Station location written to the EDI head.

  • method (str, optional) – Acquisition method recorded as an EDI info note.

  • acqby (str) – ACQBY value; defaults to a pyCSAMT-IoT provenance tag.

  • metadata (mapping, optional) – Extra head attributes (e.g. county, prospect).

Return type:

pycsamt.seg.edi.EDIFile

pycsamt.iot.deployment_from_edis(sources, *, survey_id, protocol='mqtt', role='sensor_node', capabilities=None, device_suffix='-node')#

Seed a DeploymentConfig from an EDI survey.

A lighter-weight counterpart to field_session_from_edis() that returns just the deployment inventory (one device per station), useful for planning device provisioning from a prior survey’s station list.

Parameters:
Return type:

Any

pycsamt.iot.edi_survey_table(sources, *, api=None)#

Return an EDI-survey summary as a pyCSAMT table.

Parameters:
Return type:

Any

pycsamt.iot.emtools_qc(source, impedance=None, freq=None, *, method=None, flags=False, api=None, **kwargs)#

Route IoT-acquired impedance through pycsamt.emtools QC.

Running the same coherence/skew/SNR quality control that downstream processing uses keeps field-side and post-processing QC consistent, rather than having two independent notions of a “good” station.

Parameters:

  • source (FieldSession, site.Sites, or EDI source) – A FieldSession (with impedance), an already-built Sites collection, or any EDI source accepted by read_edi_survey() (directory, glob, file, EDIFile, …).

  • impedance (mapping, optional) – Per-station impedance, required when source is a session (see field_session_to_edifiles() for the accepted forms).

  • freq (array-like, optional) – Shared frequency vector for impedance arrays without their own.

  • method (str, optional) – Acquisition method recorded on the built EDIs.

  • flags (bool) – When True return pycsamt.emtools.qc.qc_flags() (per-station pass/flag verdicts); otherwise pycsamt.emtools.qc.build_qc_table() (the full QC metrics table).

  • api (bool, optional) – Forwarded to the emtools helper’s table wrapper.

  • **kwargs – Forwarded to the emtools helper (thresholds such as min_frac_ok, min_snr_med, max_skew_med).

Return type:

A pyCSAMT QC table (or flags table).

Notes

Requires the optional geospatial stack (pyproj), since emtools QC operates on EDI-backed sites.

pycsamt.iot.field_session_from_edis(sources, *, survey_id, protocol='mqtt', role='sensor_node', operator=None, method=None, device_suffix='-node')#

Seed a FieldSession from an EDI survey.

Every station in sources becomes a registered station (with its recorded geometry and channels) and a sensor node, ready to be re-occupied in a follow-up IoT-enabled campaign.

Parameters:

  • sources (str, path, EDIFile, or iterable thereof) – The EDI survey to import (see read_edi_survey()).

  • survey_id (str) – Identifier for the new session.

  • protocol (str) – Telemetry protocol assigned to each seeded device.

  • role (str) – Device role for each seeded node.

  • operator (str, optional) – Operator recorded on the session.

  • method (str, optional) – Acquisition method recorded on the session.

  • device_suffix (str) – Suffix appended to a station id to form its device id.

Return type:

pycsamt.iot.session.FieldSession

pycsamt.iot.field_session_to_edifiles(session, impedance, freq=None, *, savepath=None, method=None, write=False, acqby='pycsamt.iot')#

Build per-station EDIs from a session plus impedance estimates.

Station geometry (lat/lon/elevation) is taken from the session’s registered stations, so the resulting EDIs carry the field-recorded location without any manual bookkeeping.

Parameters:

  • session (pycsamt.iot.session.FieldSession) – Source of station geometry and the observed method.

  • impedance (mapping) – {station_id: value} where value is a Z, a (impedance_array, freq) pair, or an impedance array paired with the shared freq argument.

  • freq (array-like, optional) – Shared frequency vector used for array values that do not carry their own frequencies.

  • savepath (str, optional) – Directory for written files (used only when write=True).

  • method (str, optional) – Overrides the session method recorded in each EDI.

  • write (bool) – When True the EDIs are written and the returned mapping holds file paths; otherwise it holds in-memory EDIFile objects.

  • acqby (str) – ACQBY provenance tag.

Returns:

{station_id: EDIFile} or {station_id: path}.

Return type:

dict

pycsamt.iot.impedance_to_z(impedance, freq, *, impedance_err=None, station=None, method=None, component='offdiag', aggregate='mean')#

Build a Z from edge impedance estimates.

This is the natural continuation of the edge diagnostics: the field node already forms per-window impedance estimates (see assess_impedance_stability()); this turns them into the impedance-tensor container the processing and inversion flow consumes.

Parameters:

  • impedance (array-like of complex) –

    Accepted shapes, with n = len(freq):

    • (n,) — one scalar impedance per frequency, placed according to component (an off-diagonal antisymmetric tensor by default);

    • (n, 2, 2) — a full impedance tensor per frequency;

    • (n_windows, n) — per-window scalar estimates, aggregated across windows (error taken from the window spread);

    • (n_windows, n, 2, 2) — per-window tensors, aggregated;

    • (2, 2) — a single tensor (only when n == 1).

  • freq (array-like of float) – Strictly positive frequency vector in Hz.

  • impedance_err (array-like, optional) – Explicit absolute errors matching the aggregated impedance shape. Overrides any error derived from window spread.

  • station (str, optional) – Name attached to the returned Z.

  • method (str, optional) – Acquisition method ("amt", "mt", "csamt") recorded in the object metadata.

  • component ({'offdiag', 'xy', 'yx'}) – Placement for scalar inputs.

  • aggregate ({'mean', 'median'}) – Reduction across a window axis when one is present.

Return type:

pycsamt.z.z.Z

pycsamt.iot.read_edi_survey(sources)#

Read an EDI survey into per-station summary records.

Parameters:

sources (str, path, EDIFile, or iterable thereof) – A directory of .edi files, a glob pattern, a single file or EDIFile, or any iterable mixing these.

Returns:

One record per station with station, lat, lon, elevation, n_freq, f_min_hz, f_max_hz, and channels.

Return type:

list of dict

pycsamt.iot.z_to_edi(z, *, station, savepath=None, filename=None, lat=None, lon=None, elevation=None, method=None, datatype=None, acqby='pycsamt.iot', metadata=None)#

Write an impedance tensor to a .edi file and return its path.

See build_edifile() for the assembled structure. datatype is forwarded to pycsamt.seg.edi.EDIFile.write() and is auto-detected (MT) when left as None.

Parameters:
Return type:

str

pycsamt.iot.z_to_site(z, *, station, lat=None, lon=None, elevation=None, method=None)#

Return an EDI-backed Site for z.

Unlike FieldSession.to_sites(), which returns acquisition descriptors, this yields a real Site built on an EDIFile and therefore requires the optional geospatial stack (pyproj).

Parameters:
Return type:

Any

class pycsamt.iot.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]#
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]

class pycsamt.iot.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 = 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:
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.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.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.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.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 = 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:
Return type:

TelemetryPacket

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

Return one row per device describing declared IoT capabilities.

Parameters:
Return type:

Any

class pycsamt.iot.StationConfig(station_id, lat=None, lon=None, elevation=None, profile=None, position_m=None, channels=<factory>, dipole_length_m=None, ex_azimuth_deg=None, ey_azimuth_deg=None, device_ids=<factory>, operator=None, notes='', metadata=<factory>)#

Bases: PyCSAMTObject, MetadataMixin

Geospatial and acquisition metadata for one field station.

Parameters:

  • station_id (str) – Stable station identifier (e.g. "S01"). Used as the join key across telemetry, QC, and provenance records.

  • lat (float, optional) – Geographic coordinates in decimal degrees. Latitude is validated to [-90, 90]; longitude accepts both [-180, 180] and [0, 360] conventions.

  • lon (float, optional) – Geographic coordinates in decimal degrees. Latitude is validated to [-90, 90]; longitude accepts both [-180, 180] and [0, 360] conventions.

  • elevation (float, optional) – Ground elevation in metres.

  • profile (str, optional) – Survey line/profile label the station belongs to (e.g. "L1").

  • position_m (float, optional) – Chainage (distance along the profile) in metres. Enables ordering of stations into a 2D section.

  • channels (list of str, optional) – Acquisition channels recorded at the station (e.g. ["ex", "ey", "hx", "hy"]).

  • dipole_length_m (float, optional) – Electric dipole length in metres, used for E-field scaling and contact-resistance checks.

  • ex_azimuth_deg (float, optional) – Orientation of the Ex/Ey electric dipoles, degrees clockwise from geographic north.

  • ey_azimuth_deg (float, optional) – Orientation of the Ex/Ey electric dipoles, degrees clockwise from geographic north.

  • device_ids (list of str, optional) – Identifiers of the IoT nodes occupying this station.

  • operator (str, optional) – Field operator responsible for the occupation.

  • notes (str) – Free-form field notes.

  • metadata (dict) – Free-form structured metadata.

station_id: str#
lat: float | None = None#
lon: float | None = None#
elevation: float | None = None#
profile: str | None = None#
position_m: float | None = None#
channels: list[str]#
dipole_length_m: float | None = None#
ex_azimuth_deg: float | None = None#
ey_azimuth_deg: float | None = None#
device_ids: list[str]#
operator: str | None = None#
notes: str = ''#
metadata: dict[str, Any]#
validate()#

Validate and normalise station fields.

Return type:

None

property has_location: bool#

Return whether latitude and longitude are both known.

property coords: tuple[float, float, float] | None#

Return (lat, lon, elevation) if a location is set.

property n_channels: int#

Number of declared acquisition channels.

attach_device(device_id)#

Associate an IoT node with this station (idempotent).

Parameters:

device_id (str)

Return type:

StationConfig

as_dict()#

Return a serialisable station dictionary.

Return type:

dict[str, Any]

classmethod from_mapping(data)#

Build a station config from a mapping, ignoring unknown keys.

Parameters:

data (Mapping[str, Any])

Return type:

StationConfig

pycsamt.iot.station_table(stations, *, api=None)#

Return one row per station describing its acquisition metadata.

Parameters:
Return type:

Any

class pycsamt.iot.FieldSession(survey_id, *, devices=None, stations=None, monitoring_config=None, sync_config=None, operator=None, method=None, metadata=None)#

Bases: PyCSAMTObject, MetadataMixin

Stateful container for one IoT-enabled field acquisition session.

Parameters:
property n_devices: int#
property n_stations: int#
property n_packets: int#
validate()#

Validate the session (required by PyCSAMTObject).

Return type:

None

add_device(device)#

Register a device, auto-creating its station when named.

Parameters:

device (DeviceConfig | Mapping[str, Any])

Return type:

DeviceConfig

add_station(station)#

Register (or merge into) a station.

Parameters:

station (StationConfig | Mapping[str, Any])

Return type:

StationConfig

add_packet(packet)#

Append one telemetry packet, registering unknown devices.

Parameters:

packet (TelemetryPacket | Mapping[str, Any])

Return type:

TelemetryPacket

add_packets(packets)#

Append many telemetry packets.

Parameters:

packets (Iterable[TelemetryPacket | Mapping[str, Any]])

Return type:

None

assess(*, now=None)#

Assess telemetry quality with the session monitoring config.

Parameters:

now (float | None)

Return type:

MonitoringStatus

station_table(*, api=None)#

Return the registered stations as a pyCSAMT table.

Parameters:

api (bool | None)

Return type:

Any

to_sites()#

Return acquisition-level station descriptors.

These are StationConfig objects enriched from telemetry (channels and method observed in packets). They are acquisition descriptors, not EDI-backed pycsamt.site.base.Site objects, which require processed impedance data.

Once impedance is available, to_edifiles() and to_sites_collection() promote these descriptors into EDI-backed sites ready for the processing pipeline.

Return type:

list[StationConfig]

to_pipeline_input()#

Return a serialisable hand-off for the processing pipeline.

Contains, per station, the location, channels, observed method, accepted frequency band, and QC acceptance — everything the downstream pyCSAMT processing flow needs to know from acquisition.

Return type:

dict[str, Any]

to_edifiles(impedance, freq=None, *, method=None, savepath=None, write=False, acqby='pycsamt.iot')#

Promote per-station impedance into EDIFile objects.

A thin convenience over pycsamt.iot.bridge.field_session_to_edifiles(): each EDI is enriched with the geometry this session recorded for the station. See that function for the accepted impedance mapping forms and the meaning of write/savepath.

Parameters:
Return type:

dict[str, Any]

to_sites_collection(impedance, freq=None, *, method=None)#

Return an EDI-backed pycsamt.site.base.Sites collection.

This is the processing hand-off: given per-station impedance, it builds one EDI per station and wraps them into a Sites collection that can be fed straight to pycsamt.pipeline.Pipeline.run(). Requires the optional geospatial stack (pyproj).

Parameters:
Return type:

Any

to_manifest()#

Build a reproducible AcquisitionManifest.

Return type:

AcquisitionManifest

export_manifest(path, *, indent=2)#

Write the acquisition manifest to path and return its path.

Parameters:
Return type:

str

to_dict()#

Return a serialisable representation of the whole session.

Return type:

dict[str, Any]

to_json(*, indent=2)#

Return the session as a JSON string.

Parameters:

indent (int)

Return type:

str

classmethod from_dict(data)#

Reconstruct a session from to_dict() output.

Parameters:

data (Mapping[str, Any])

Return type:

FieldSession

classmethod from_json(text)#

Reconstruct a session from a JSON string.

Parameters:

text (str)

Return type:

FieldSession

write_json(path, *, indent=2)#

Write the session JSON to path and return its path.

Parameters:
Return type:

str

classmethod read_json(path)#

Load a session from a JSON file.

Parameters:

path (str)

Return type:

FieldSession

class pycsamt.iot.AcquisitionPayload(kind=PacketKind.DATA, station=None, extra=<factory>, method=None, channels=<factory>, sample_rate_hz=None, frequency_hz=None, frequency_band_hz=None, n_samples=None, gain=None, duration_s=None)#

Bases: TelemetryPayload

Metadata describing one raw acquisition record/window.

Parameters:
kind: PacketKind = 'data'#

Canonical packet kind the schema describes. Overridden per subclass.

method: str | None = None#
channels: list[str]#
sample_rate_hz: float | None = None#
frequency_hz: float | None = None#
frequency_band_hz: tuple[float, float] | None = None#
n_samples: int | None = None#
gain: float | None = None#
duration_s: float | None = None#
validate()#

Validate object state.

Subclasses can override this hook. The base implementation intentionally accepts all states.

Return type:

None

classmethod from_mapping(payload)#
Parameters:

payload (Mapping[str, Any])

Return type:

AcquisitionPayload

as_dict(*, drop_none=False)#

Return a flat payload dictionary with canonical keys.

Parameters:

drop_none (bool)

Return type:

dict[str, Any]

class pycsamt.iot.EventPayload(kind=PacketKind.EVENT, station=None, extra=<factory>, event=None, severity=EventSeverity.INFO, message=None, code=None)#

Bases: TelemetryPayload

Discrete field event (state change, alarm, operator note).

Parameters:
kind: PacketKind = 'event'#

Canonical packet kind the schema describes. Overridden per subclass.

event: str | None = None#
severity: EventSeverity | str = 'info'#
message: str | None = None#
code: str | None = None#
validate()#

Validate object state.

Subclasses can override this hook. The base implementation intentionally accepts all states.

Return type:

None

classmethod from_mapping(payload)#
Parameters:

payload (Mapping[str, Any])

Return type:

EventPayload

as_dict(*, drop_none=False)#

Return a flat payload dictionary with canonical keys.

Parameters:

drop_none (bool)

Return type:

dict[str, Any]

class pycsamt.iot.EventSeverity(*values)#

Bases: str, Enum

Severity levels for event telemetry.

INFO = 'info'#
WARNING = 'warning'#
ERROR = 'error'#
CRITICAL = 'critical'#
class pycsamt.iot.HealthPayload(kind=PacketKind.HEALTH, station=None, extra=<factory>, battery_v=None, temperature_c=None, uptime_s=None, free_storage_mb=None, rssi_dbm=None, firmware=None)#

Bases: TelemetryPayload

Device-health telemetry (battery, temperature, link quality).

Parameters:
kind: PacketKind = 'health'#

Canonical packet kind the schema describes. Overridden per subclass.

battery_v: float | None = None#
temperature_c: float | None = None#
uptime_s: float | None = None#
free_storage_mb: float | None = None#
rssi_dbm: float | None = None#
firmware: str | None = None#
validate()#

Validate object state.

Subclasses can override this hook. The base implementation intentionally accepts all states.

Return type:

None

classmethod from_mapping(payload)#
Parameters:

payload (Mapping[str, Any])

Return type:

HealthPayload

as_dict(*, drop_none=False)#

Return a flat payload dictionary with canonical keys.

Parameters:

drop_none (bool)

Return type:

dict[str, Any]

class pycsamt.iot.PowerPayload(kind=PacketKind.POWER, station=None, extra=<factory>, battery_v=None, state=None, runtime_days=None, net_wh_per_day=None, solar_w=None, load_w=None)#

Bases: TelemetryPayload

Energy-budget telemetry for a field node.

Parameters:
kind: PacketKind = 'power'#

Canonical packet kind the schema describes. Overridden per subclass.

battery_v: float | None = None#
state: str | None = None#
runtime_days: float | None = None#
net_wh_per_day: float | None = None#
solar_w: float | None = None#
load_w: float | None = None#
validate()#

Validate object state.

Subclasses can override this hook. The base implementation intentionally accepts all states.

Return type:

None

classmethod from_mapping(payload)#
Parameters:

payload (Mapping[str, Any])

Return type:

PowerPayload

as_dict(*, drop_none=False)#

Return a flat payload dictionary with canonical keys.

Parameters:

drop_none (bool)

Return type:

dict[str, Any]

class pycsamt.iot.QCPayload(kind=PacketKind.QC, station=None, extra=<factory>, accepted=None, decision=None, snr_db=None, finite_coverage=None, spike_fraction=None, rms=None, method=None, channels=<factory>, frequency_band_hz=None, reasons=<factory>)#

Bases: TelemetryPayload

Edge quality-control telemetry for one acquisition window.

Parameters:
kind: PacketKind = 'qc'#

Canonical packet kind the schema describes. Overridden per subclass.

accepted: bool | None = None#
decision: str | None = None#
snr_db: float | None = None#
finite_coverage: float | None = None#
spike_fraction: float | None = None#
rms: float | None = None#
method: str | None = None#
channels: list[str]#
frequency_band_hz: tuple[float, float] | None = None#
reasons: list[str]#
validate()#

Validate object state.

Subclasses can override this hook. The base implementation intentionally accepts all states.

Return type:

None

classmethod from_mapping(payload)#
Parameters:

payload (Mapping[str, Any])

Return type:

QCPayload

as_dict(*, drop_none=False)#

Return a flat payload dictionary with canonical keys.

Parameters:

drop_none (bool)

Return type:

dict[str, Any]

class pycsamt.iot.SourcePayload(kind=PacketKind.SOURCE, station=None, extra=<factory>, source_id=None, tx_current_a=None, tx_voltage_v=None, tx_frequency_hz=None, tx_power_w=None, dipole_length_m=None, duty_cycle=None, on=None, offset_m=None, azimuth_deg=None)#

Bases: TelemetryPayload

Transmitter telemetry for a controlled-source (CSAMT/TDEM) survey.

Reported by a transmitter node so the receiver-side QC and provenance know the state of the source that produced each sounding: the injected current and voltage, the frequency currently being transmitted, the grounded-dipole geometry, and the keyed on/off state.

Parameters:
kind: PacketKind = 'source'#

Canonical packet kind the schema describes. Overridden per subclass.

source_id: str | None = None#
tx_current_a: float | None = None#
tx_voltage_v: float | None = None#
tx_frequency_hz: float | None = None#
tx_power_w: float | None = None#
dipole_length_m: float | None = None#
duty_cycle: float | None = None#
on: bool | None = None#
offset_m: float | None = None#
azimuth_deg: float | None = None#
validate()#

Validate object state.

Subclasses can override this hook. The base implementation intentionally accepts all states.

Return type:

None

classmethod from_mapping(payload)#
Parameters:

payload (Mapping[str, Any])

Return type:

SourcePayload

as_dict(*, drop_none=False)#

Return a flat payload dictionary with canonical keys.

Parameters:

drop_none (bool)

Return type:

dict[str, Any]

class pycsamt.iot.SyncPayload(kind=PacketKind.SYNC, station=None, extra=<factory>, offset_ms=None, drift_ppm=None, jitter_ms=None, gps_lock=None, n_reference_points=None, reference=None, tx_locked=None, tx_sync_offset_ms=None, tx_id=None)#

Bases: TelemetryPayload

Clock-synchronisation telemetry for a field node.

Parameters:
kind: PacketKind = 'sync'#

Canonical packet kind the schema describes. Overridden per subclass.

offset_ms: float | None = None#
drift_ppm: float | None = None#
jitter_ms: float | None = None#
gps_lock: bool | None = None#
n_reference_points: int | None = None#
reference: str | None = None#
tx_locked: bool | None = None#
tx_sync_offset_ms: float | None = None#
tx_id: str | None = None#
validate()#

Validate object state.

Subclasses can override this hook. The base implementation intentionally accepts all states.

Return type:

None

classmethod from_mapping(payload)#
Parameters:

payload (Mapping[str, Any])

Return type:

SyncPayload

as_dict(*, drop_none=False)#

Return a flat payload dictionary with canonical keys.

Parameters:

drop_none (bool)

Return type:

dict[str, Any]

class pycsamt.iot.TelemetryPayload(kind=PacketKind.DATA, station=None, extra=<factory>)#

Bases: PyCSAMTObject

Base class for canonical telemetry payloads.

Parameters:
kind: PacketKind = 'data'#

Canonical packet kind the schema describes. Overridden per subclass.

station: str | None = None#
extra: dict[str, Any]#
as_dict(*, drop_none=False)#

Return a flat payload dictionary with canonical keys.

Parameters:

drop_none (bool)

Return type:

dict[str, Any]

pycsamt.iot.parse_payload(kind, payload)#

Parse payload into the schema registered for kind.

Parameters:
Return type:

TelemetryPayload

pycsamt.iot.schema_for(kind)#

Return the payload schema registered for kind.

Parameters:

kind (PacketKind | str)

Return type:

type[TelemetryPayload]

pycsamt.iot.validate_payload(kind, payload, *, drop_none=True)#

Return a canonicalised payload dictionary for kind.

Aliased keys are folded into canonical names and values are range checked. Unknown keys are preserved. Raises ValueError when a field fails validation.

Parameters:
Return type:

dict[str, Any]

class pycsamt.iot.EdgeChannelSummary(channel, n_sample, finite_coverage, rms, mean, std, min, max, spike_fraction, accepted, reasons=<factory>)#

Bases: PyCSAMTObject

Quality-control summary for one edge data channel.

Parameters:
channel: str#
n_sample: int#
finite_coverage: float#
rms: float#
mean: float#
std: float#
min: float#
max: float#
spike_fraction: float#
accepted: bool#
reasons: list[str]#
validate()#

Validate and normalise the channel summary.

Return type:

None

as_dict()#

Return a serialisable channel summary.

Return type:

dict[str, Any]

class pycsamt.iot.EdgeDecision(*values)#

Bases: str, Enum

Acceptance state assigned by edge-side quality control.

ACCEPT and REJECT are the hard decisions. WARNING marks a window that is usable but marginal, REPEAT requests re-occupation, and UNKNOWN is used when QC could not be evaluated.

ACCEPT = 'accept'#
REJECT = 'reject'#
WARNING = 'warning'#
REPEAT = 'repeat'#
UNKNOWN = 'unknown'#
class pycsamt.iot.EdgeProcessingConfig(decimation=1, finite_threshold=0.9, sample_axis=0, channel_names=None, compute_rms=True, compute_coverage=True, compute_spikes=True, spike_threshold=6.0, max_spike_fraction=0.05, warn_finite_threshold=None, warn_spike_fraction=None, retain_payload_samples=False, metadata=<factory>)#

Bases: PyCSAMTObject

Configuration for lightweight field-side processing.

Parameters:

  • decimation (int)

  • finite_threshold (float)

  • sample_axis (int)

  • channel_names (list[str] | None)

  • compute_rms (bool)

  • compute_coverage (bool)

  • compute_spikes (bool)

  • spike_threshold (float | None)

  • max_spike_fraction (float)

  • warn_finite_threshold (float | None)

  • warn_spike_fraction (float | None)

  • retain_payload_samples (bool)

  • metadata (dict[str, Any])

decimation: int = 1#
finite_threshold: float = 0.9#
sample_axis: int = 0#
channel_names: list[str] | None = None#
compute_rms: bool = True#
compute_coverage: bool = True#
compute_spikes: bool = True#
spike_threshold: float | None = 6.0#
max_spike_fraction: float = 0.05#
warn_finite_threshold: float | None = None#
warn_spike_fraction: float | None = None#
retain_payload_samples: bool = False#
metadata: dict[str, Any]#
validate()#

Validate and normalise edge-processing options.

Return type:

None

class pycsamt.iot.EdgeProcessingResult(data, metrics, accepted, reasons=<factory>, channels=<factory>, decision_override=None)#

Bases: PyCSAMTObject

Summary returned by EdgeProcessor.

Parameters:
data: ndarray#
metrics: dict[str, Any]#
accepted: bool#
reasons: list[str]#
channels: list[EdgeChannelSummary]#
decision_override: EdgeDecision | str | None = None#
validate()#

Validate and normalise the processing result.

Return type:

None

property decision: EdgeDecision#

Return the compact QC decision.

Falls back to a plain accept/reject derived from accepted unless an explicit decision_override (e.g. WARNING) was assigned by the processor.

payload(*, include_data=None)#

Return a serialisable QC payload for telemetry.

Parameters:

include_data (bool | None)

Return type:

dict[str, Any]

to_packet(device, *, timestamp, survey_id=None, qos=0, retained=False, include_data=None)#

Encode this edge result as a QC telemetry packet.

Parameters:
Return type:

TelemetryPacket

class pycsamt.iot.EdgeProcessor(config=None)#

Bases: PyCSAMTObject

Small edge-processing block for telemetry payload reduction.

Parameters:

config (EdgeProcessingConfig | None)

process(data, *, channel_names=None)#

Decimate numeric data and compute simple quality metrics.

Parameters:
Return type:

EdgeProcessingResult

pycsamt.iot.edge_summary_table(result, *, api=None)#

Return one row per channel from edge-processing results.

Parameters:
Return type:

Any

class pycsamt.iot.FrequencyCoverage(sample_rate_hz, nyquist_hz, f_low_hz, f_high_hz, n_decades, coverage_fraction=nan, missing_bands=<factory>)#

Bases: PyCSAMTObject

Result of estimate_frequency_coverage().

Parameters:
sample_rate_hz: float#
nyquist_hz: float#
f_low_hz: float#
f_high_hz: float#
n_decades: float#
coverage_fraction: float = nan#
missing_bands: list[tuple[float, float]]#
as_dict()#
Return type:

dict[str, Any]

class pycsamt.iot.HarmonicPeak(order, frequency_hz, power_ratio, flagged)#

Bases: PyCSAMTObject

Contamination measured at one powerline harmonic.

Parameters:
order: int#
frequency_hz: float#
power_ratio: float#
flagged: bool#
as_dict()#
Return type:

dict[str, Any]

class pycsamt.iot.ImpedanceStability(n_windows, cv_magnitude, phase_std_deg, stable)#

Bases: PyCSAMTObject

Result of assess_impedance_stability().

Parameters:
n_windows: int#
cv_magnitude: float#
phase_std_deg: float#
stable: bool#
as_dict()#
Return type:

dict[str, Any]

class pycsamt.iot.PowerlineHarmonics(mains_hz, peaks=<factory>, total_ratio=0.0, contaminated=False)#

Bases: PyCSAMTObject

Result of detect_powerline_harmonics().

Parameters:
mains_hz: float#
peaks: list[HarmonicPeak]#
total_ratio: float = 0.0#
contaminated: bool = False#
property dominant: HarmonicPeak | None#

Return the strongest harmonic, if any were measured.

as_dict()#
Return type:

dict[str, Any]

class pycsamt.iot.StaticShift(shift_factor, split_decades, consistency_std, phase_diff_deg, static_shift)#

Bases: PyCSAMTObject

Result of estimate_static_shift().

Parameters:
shift_factor: float#
split_decades: float#
consistency_std: float#
phase_diff_deg: float#
static_shift: bool#
as_dict()#
Return type:

dict[str, Any]

pycsamt.iot.amt_edge_report(data, sample_rate, *, method=None, mains_hz=50.0, signal_band_hz=None)#

Run the core AMT edge diagnostics on one channel and collate them.

When method is given ("amt", "mt", "csamt", …), the diagnostics become method-aware: powerline-harmonic detection is only run for powerline-sensitive methods (it is skipped for, e.g., TDEM), and frequency coverage is scored against the method’s target bands. Passing no method preserves the original behaviour.

Parameters:
Return type:

dict[str, Any]

pycsamt.iot.amt_edge_table(reports, *, api=None)#

Flatten one or more amt_edge_report() results into a table.

Accepts a {channel: report} mapping (or (channel, report) pairs) and returns one row per channel with the headline metrics.

Parameters:
Return type:

Any

pycsamt.iot.assess_impedance_stability(z_windows, *, max_cv=0.15, max_phase_std_deg=10.0)#

Assess the stability of per-window impedance estimates.

Parameters:

  • z_windows (array-like of complex) – Impedance estimates, shape (n_windows,) or (n_windows, n_freq). Real inputs are treated as magnitudes with zero phase.

  • max_cv (float) – Maximum coefficient of variation of |Z| for a stable result.

  • max_phase_std_deg (float) – Maximum phase standard deviation (degrees) for a stable result.

Return type:

ImpedanceStability

pycsamt.iot.check_channel_saturation(data, *, limit=None, max_clip_fraction=0.01, tol=1e-09)#

Detect ADC clipping / saturation in a channel.

When limit is provided, samples with abs(x) >= limit count as saturated. Otherwise, samples equal to the observed min/max (within tol) are treated as clipped, which catches rail-to-rail saturation without a known full-scale value.

Parameters:
Return type:

dict[str, Any]

pycsamt.iot.check_contact_resistance(ex, ey=None, *, sample_rate=None, noise_rms_threshold=None)#

Proxy assessment of electrode contact quality.

True contact resistance requires a current injection measurement. On passive AMT electric channels, poor contact manifests as elevated low-frequency noise, large DC offset, and drift. This routine reports those proxies per channel (ex and optional ey) and flags a channel when its high-pass noise RMS exceeds noise_rms_threshold (when provided) or when Ex/Ey noise is strongly imbalanced.

Parameters:

  • ex (Any)

  • ey (Any)

  • sample_rate (float | None)

  • noise_rms_threshold (float | None)

Return type:

dict[str, Any]

pycsamt.iot.compute_live_spectra(data, sample_rate, *, nperseg=None)#

Return {"frequency_hz": ..., "psd": ...} for live display.

Parameters:
Return type:

dict[str, ndarray]

pycsamt.iot.detect_powerline_harmonics(data, sample_rate, *, mains_hz=50.0, n_harmonics=5, bandwidth_hz=1.0, threshold_ratio=0.05)#

Detect mains-frequency harmonics in a time series.

Parameters:

  • data (array-like) – Single-channel time series.

  • sample_rate (float) – Sampling frequency in Hz.

  • mains_hz (float) – Powerline fundamental (50 or 60 Hz typically).

  • n_harmonics (int) – Number of harmonics (including the fundamental) to test.

  • bandwidth_hz (float) – Half-width of the integration band around each harmonic.

  • threshold_ratio (float) – Per-harmonic band-power fraction above which a harmonic is flagged as contaminating.

Return type:

PowerlineHarmonics

pycsamt.iot.detect_sensor_dropout(data, *, min_flat_run=8, flat_tol=1e-12)#

Detect NaN gaps and stuck-value (flatline) runs in a channel.

Returns counts for NaN samples and the longest run of (near-)constant consecutive samples, which typically indicates a disconnected or stuck sensor.

Parameters:
Return type:

dict[str, Any]

pycsamt.iot.estimate_channel_snr(data, sample_rate=None, *, signal_band_hz=None)#

Estimate channel SNR in decibels.

Two estimators are provided:

  • If sample_rate and signal_band_hz are given, SNR is the ratio of in-band to out-of-band spectral power.

  • Otherwise a time-domain estimate is used: the signal power is the variance of the series and the noise power is derived from the variance of first differences (a white-noise proxy).

Parameters:
Return type:

float

pycsamt.iot.estimate_frequency_coverage(timeseries, sample_rate, *, target_bands=None, snr_floor_db=6.0)#

Estimate the resolvable frequency band of a recording.

The PSD noise floor is taken as its median. Frequencies whose power exceeds the floor by snr_floor_db are considered resolved; the lowest and highest such frequencies define the covered band. When target_bands are supplied, the fraction that falls inside the covered band is reported along with any missing bands.

Parameters:
Return type:

FrequencyCoverage

pycsamt.iot.estimate_static_shift(res_xy, res_yx, *, phase_xy=None, phase_yx=None, min_split_decades=0.15, max_log_std=0.15, max_phase_diff_deg=10.0)#

Flag a static shift between the two apparent-resistivity modes.

Static shift is a galvanic distortion that multiplies apparent resistivity by a frequency-independent factor while leaving phase unchanged. It therefore shows up as the xy and yx resistivity curves running parallel on a log scale (a near-constant split) even though their phases coincide – unlike true anisotropy, which splits the phases too.

Parameters:

  • res_xy (array-like) – Apparent resistivity (\(\Omega\cdot m\)) for the two off-diagonal modes, one value per frequency.

  • res_yx (array-like) – Apparent resistivity (\(\Omega\cdot m\)) for the two off-diagonal modes, one value per frequency.

  • phase_xy (array-like, optional) – Corresponding phases in degrees. When given, agreeing phases strengthen a static-shift call (and disagreeing phases veto it).

  • phase_yx (array-like, optional) – Corresponding phases in degrees. When given, agreeing phases strengthen a static-shift call (and disagreeing phases veto it).

  • min_split_decades (float) – Minimum |log10(shift_factor)| for a split to matter.

  • max_log_std (float) – Maximum standard deviation of the per-frequency log split for it to count as frequency-independent.

  • max_phase_diff_deg (float) – Maximum mean phase difference (when phases are supplied) for the distortion to read as purely galvanic.

Return type:

StaticShift

class pycsamt.iot.FieldZone(*values)#

Bases: str, Enum

Wave-propagation regime of a CSAMT measurement at one frequency.

NEAR = 'near'#
TRANSITION = 'transition'#
FAR = 'far'#
class pycsamt.iot.FieldZoneCoverage(offset_m, freq_hz=<factory>, skin_depth_m=<factory>, offset_ratio=<factory>, zones=<factory>, n_near=0, n_transition=0, n_far=0)#

Bases: PyCSAMTObject

Result of classify_field_zones().

Parameters:
offset_m: float#
freq_hz: list[float]#
skin_depth_m: list[float]#
offset_ratio: list[float]#
zones: list[str]#
n_near: int = 0#
n_transition: int = 0#
n_far: int = 0#
property far_fraction: float#

Fraction of frequencies recorded in the far field.

property all_far_field: bool#

True when every frequency satisfies the far-field assumption.

True when any frequency needs a near-field correction.

first_far_field_hz()#

Lowest frequency that reaches the far field, if any.

Return type:

float | None

as_dict()#
Return type:

dict[str, Any]

class pycsamt.iot.SourceLine(frequency_hz, snr_db, power_ratio, detected)#

Bases: PyCSAMTObject

Detection result for one expected transmitter frequency.

Parameters:
frequency_hz: float#
snr_db: float#
power_ratio: float#
detected: bool#
as_dict()#
Return type:

dict[str, Any]

class pycsamt.iot.SourceStability(current_mean_a, current_cv, current_drift_a, on_fraction, voltage_mean_v=None, stable=False, flags=<factory>)#

Bases: PyCSAMTObject

Result of assess_source_stability().

Parameters:
current_mean_a: float#
current_cv: float#
current_drift_a: float#
on_fraction: float#
voltage_mean_v: float | None = None#
stable: bool = False#
flags: list[str]#
as_dict()#
Return type:

dict[str, Any]

class pycsamt.iot.TransmitterComb(lines=<factory>)#

Bases: PyCSAMTObject

Result of detect_transmitter_frequencies().

Parameters:

lines (list[SourceLine])

lines: list[SourceLine]#
property n_expected: int#
property n_detected: int#
property detection_fraction: float#
missing()#

Expected frequencies whose energy was not resolvable.

Return type:

list[float]

as_dict()#
Return type:

dict[str, Any]

pycsamt.iot.assess_source_stability(tx_current, tx_voltage=None, *, max_cv=0.05, on_threshold_frac=0.5)#

Assess transmitter current (and optional voltage) steadiness.

The transmitter current sets the signal level of every sounding, so its steadiness bounds data quality. This reports the coefficient of variation and linear drift of the on-state current, together with the on-fraction of an on/off-keyed source.

Parameters:

  • tx_current (array-like) – Transmitter current samples (A).

  • tx_voltage (array-like, optional) – Transmitter voltage samples (V).

  • max_cv (float) – Maximum on-state current coefficient of variation for a stable source.

  • on_threshold_frac (float) – Fraction of the peak current above which the source counts as “on”; on-state statistics use only those samples.

Return type:

SourceStability

pycsamt.iot.classify_field_zones(freq, resistivity, offset_m, *, near_ratio=1.0, far_ratio=3.0)#

Classify each frequency as near, transition, or far field.

The controlling quantity is the transmitter-receiver offset expressed in skin depths, \(r / \delta\). A larger ratio means the receiver is electrically farther from the source and the plane-wave assumption is safer.

Parameters:

  • freq (array-like of float) – Frequencies in Hz.

  • resistivity (float or array-like) – (Apparent) resistivity in \(\Omega\cdot m\). A scalar is used as a half-space value for every frequency; an array must match freq.

  • offset_m (float) – Transmitter-receiver separation \(r\) in metres.

  • near_ratio (float) – r / delta at or below which a frequency is near field.

  • far_ratio (float) – r / delta at or above which a frequency is far field. Values in between are the transition zone.

Return type:

FieldZoneCoverage

pycsamt.iot.csamt_edge_report(data, sample_rate, *, tx_frequencies, offset_m, resistivity, tx_current=None, tx_voltage=None, near_ratio=1.0, far_ratio=3.0)#

Run the core CSAMT edge diagnostics on one channel and collate them.

Parameters:
Return type:

dict[str, Any]

pycsamt.iot.csamt_edge_table(reports, *, api=None)#

Flatten one or more csamt_edge_report() results into a table.

Parameters:
Return type:

Any

pycsamt.iot.detect_transmitter_frequencies(data, sample_rate, tx_frequencies, *, bandwidth_hz=1.0, snr_threshold_db=6.0)#

Check that recorded energy is present at each transmitted frequency.

For every expected transmitter frequency the in-band power is compared with a local out-of-band noise estimate to form a per-line SNR. A line counts as detected when its SNR meets snr_threshold_db; lines above Nyquist are reported as undetected.

Parameters:

  • data (array-like) – Single-channel time series.

  • sample_rate (float) – Sampling frequency in Hz.

  • tx_frequencies (iterable of float) – Expected transmitter frequencies in Hz.

  • bandwidth_hz (float) – Half-width of the signal band around each line.

  • snr_threshold_db (float) – Minimum in-band/out-of-band SNR for a line to count as detected.

Return type:

TransmitterComb

pycsamt.iot.skin_depth_m(resistivity, freq)#

Electromagnetic skin depth in metres.

\(\delta = 503.29\,\sqrt{\rho / f}\) with \(\rho\) in \(\Omega\cdot m\) and \(f\) in Hz. Scalars and arrays broadcast together; non-positive or non-finite inputs yield nan.

Parameters:

  • resistivity (Any)

  • freq (Any)

Return type:

ndarray

class pycsamt.iot.OffsetResponse(frequency_hz, offsets_m=<factory>, amplitudes=<factory>, phases_deg=None, above_noise=<factory>, max_detectable_offset_m=None, monotonic_decay=False, dynamic_range_db=nan)#

Bases: PyCSAMTObject

Result of field_vs_offset() – one MVO/PVO curve.

Parameters:
frequency_hz: float | None#
offsets_m: list[float]#
amplitudes: list[float]#
phases_deg: list[float] | None = None#
above_noise: list[bool]#
max_detectable_offset_m: float | None = None#
monotonic_decay: bool = False#
dynamic_range_db: float = nan#
property n_offsets: int#
property n_detectable: int#
as_dict()#
Return type:

dict[str, Any]

pycsamt.iot.csem_edge_report(offsets_m, amplitudes, *, phases_deg=None, noise_floor=None, frequency_hz=None)#

Run the CSEM offset-domain diagnostics for one frequency.

Parameters:

  • offsets_m (Any)

  • amplitudes (Any)

  • phases_deg (Any)

  • noise_floor (float | None)

  • frequency_hz (float | None)

Return type:

dict[str, Any]

pycsamt.iot.csem_edge_table(reports, *, api=None)#

Flatten one or more csem_edge_report() results into a table.

Accepts a {label: report} mapping (or (label, report) pairs); the label is typically a frequency or receiver-line identifier.

Parameters:
Return type:

Any

pycsamt.iot.field_vs_offset(offsets_m, amplitudes, *, phases_deg=None, noise_floor=None, frequency_hz=None, monotonic_tol=0.05)#

Build a magnitude/phase-versus-offset curve and QC it.

Parameters:

  • offsets_m (array-like of float) – Source-receiver offsets in metres (need not be sorted).

  • amplitudes (array-like of float) – Field amplitude at each offset (e.g. normalised E-field). Must be non-negative and the same length as offsets_m.

  • phases_deg (array-like of float, optional) – Phase at each offset in degrees, carried through for PVO.

  • noise_floor (float, optional) – Amplitude below which a reading is not detectable. When given, the largest offset above the floor is reported as the detectability limit and only detectable points drive the decay/dynamic-range QC.

  • frequency_hz (float, optional) – Frequency this curve was measured at, recorded for reference.

  • monotonic_tol (float) – Fractional tolerance when checking that amplitude never increases with offset (a[i+1] <= a[i] * (1 + tol)).

Return type:

OffsetResponse

class pycsamt.iot.MethodProfile(method, frequency_band_hz, required_channels, default_sample_rate_hz=None, controlled_source=False, powerline_sensitive=False, description='')#

Bases: object

Canonical acquisition characteristics for one EM method.

Parameters:
method: EMMethod#
frequency_band_hz: tuple[float, float] | None#
required_channels: tuple[str, ...]#
default_sample_rate_hz: float | None = None#
controlled_source: bool = False#
powerline_sensitive: bool = False#
description: str = ''#
property is_natural_source: bool#

True for passive methods (no transmitter).

as_dict()#
Return type:

dict[str, Any]

pycsamt.iot.method_profile(method)#

Return the MethodProfile for method.

Parameters:

method (EMMethod or str) – Method enum member or a name/value such as "csamt".

Returns:

The registered profile, or the UNKNOWN profile for a method with no registered defaults.

Return type:

MethodProfile

pycsamt.iot.target_bands_for_method(method)#

Return per-decade sub-bands spanning a method’s frequency band.

These feed coverage checks such as estimate_frequency_coverage(), which report the fraction of target bands a recording actually resolves. A method without a defined band (e.g. time-domain) returns an empty list.

Parameters:

method (EMMethod | str)

Return type:

list[tuple[float, float]]

class pycsamt.iot.EMMethod(*values)#

Bases: str, Enum

Electromagnetic survey methods recognised by IoT monitoring.

AMT = 'amt'#
MT = 'mt'#
CSAMT = 'csamt'#
CSEM = 'csem'#
TDEM = 'tdem'#
TEM = 'tem'#
UNKNOWN = 'unknown'#
class pycsamt.iot.MonitoringConfig(method=EMMethod.UNKNOWN, expected_interval_s=None, max_latency_s=30.0, max_gap_s=None, min_packet_success_rate=0.95, min_edge_acceptance_rate=0.85, min_battery_v=None, max_clock_offset_ms=None, frequency_band_hz=None, required_channels=<factory>, metadata=<factory>)#

Bases: PyCSAMTObject, MetadataMixin

Thresholds used to monitor AMT/MT/CSAMT telemetry streams.

Parameters:
method: EMMethod | str = 'unknown'#
expected_interval_s: float | None = None#
max_latency_s: float | None = 30.0#
max_gap_s: float | None = None#
min_packet_success_rate: float = 0.95#
min_edge_acceptance_rate: float = 0.85#
min_battery_v: float | None = None#
max_clock_offset_ms: float | None = None#
frequency_band_hz: tuple[float, float] | None = None#
required_channels: list[str]#
metadata: dict[str, Any]#
classmethod for_method(method, **overrides)#

Build a config seeded with a method’s canonical defaults.

The expected frequency band and required channels are taken from the method’s MethodProfile, so the method-mismatch, missing-channel, and out-of-band checks in TelemetryMonitor become method-aware without any manual threshold tuning. Any keyword overrides win over the defaults.

Example

>>> cfg = MonitoringConfig.for_method("csamt", min_battery_v=11.5)
>>> cfg.required_channels
['ex', 'ey', 'hx', 'hy']
Parameters:
Return type:

MonitoringConfig

validate()#

Validate and normalise monitoring thresholds.

Return type:

None

class pycsamt.iot.MonitoringLevel(*values)#

Bases: str, Enum

Overall status level for a monitored telemetry stream.

OK = 'ok'#
WARNING = 'warning'#
CRITICAL = 'critical'#
NO_DATA = 'no_data'#
class pycsamt.iot.MonitoringStatus(level, n_packet, packet_success_rate, edge_acceptance_rate, mean_latency_s, max_gap_s, battery_min_v, clock_offset_max_ms, methods=<factory>, stations=<factory>, channels=<factory>, frequency_min_hz=nan, frequency_max_hz=nan, issues=<factory>)#

Bases: PyCSAMTObject

Status returned by TelemetryMonitor.

Parameters:
level: MonitoringLevel | str#
n_packet: int#
packet_success_rate: float#
edge_acceptance_rate: float#
mean_latency_s: float#
max_gap_s: float#
battery_min_v: float#
clock_offset_max_ms: float#
methods: list[str]#
stations: list[str]#
channels: list[str]#
frequency_min_hz: float = nan#
frequency_max_hz: float = nan#
issues: list[str]#
validate()#

Validate and normalise status fields.

Return type:

None

property ok: bool#

Return whether the monitored stream is acceptable.

as_dict()#

Return a flat serialisable status dictionary.

Return type:

dict[str, Any]

class pycsamt.iot.TelemetryMonitor(config=None)#

Bases: PyCSAMTObject

Monitor field telemetry from AMT, MT, CSAMT, and related surveys.

Parameters:

config (MonitoringConfig | None)

assess(packets, *, now=None)#

Assess a telemetry packet stream against configured thresholds.

Parameters:
Return type:

MonitoringStatus

table(packets, *, now=None, api=None)#

Return enriched packet rows used by the monitor.

Parameters:
Return type:

Any

pycsamt.iot.assess_telemetry(packets, *, config=None, now=None)#

Convenience wrapper around TelemetryMonitor.

Parameters:
Return type:

MonitoringStatus

pycsamt.iot.monitoring_status_table(statuses, *, api=None)#

Return monitoring statuses as a pyCSAMT table.

Parameters:
Return type:

Any

pycsamt.iot.packet_table(packets, *, api=None)#

Return telemetry packets as a pyCSAMT table.

Parameters:
Return type:

Any

pycsamt.iot.telemetry_summary(packets, *, api=None)#

Summarise telemetry packet counts by device and topic.

Parameters:
Return type:

Any

class pycsamt.iot.DevicePowerProfile(name, active_power_w, sleep_power_w=0.05, telemetry_power_w=0.0, edge_power_w=0.0, metadata=<factory>)#

Bases: PyCSAMTObject

Named power profile for a field node or recorder.

Parameters:
name: str#
active_power_w: float#
sleep_power_w: float = 0.05#
telemetry_power_w: float = 0.0#
edge_power_w: float = 0.0#
metadata: dict[str, Any]#
validate()#

Validate and normalise profile fields.

Return type:

None

apply(*, battery_wh, duty_cycle=1.0, solar_wh_per_day=0.0, **kwargs)#

Build an EnergyConfig from this profile.

Parameters:
Return type:

EnergyConfig

class pycsamt.iot.EnergyConfig(battery_wh, active_power_w, sleep_power_w=0.05, duty_cycle=1.0, solar_wh_per_day=0.0, reserve_fraction=0.0, regulator_efficiency=1.0, charge_efficiency=1.0, telemetry_power_w=0.0, telemetry_seconds_per_day=0.0, edge_power_w=0.0, edge_duty_cycle=0.0, auxiliary_wh_per_day=0.0, min_runtime_days=None, device_id=None, metadata=<factory>)#

Bases: PyCSAMTObject, MetadataMixin

Power-budget inputs for a field IoT device.

Parameters:
battery_wh: float#
active_power_w: float#
sleep_power_w: float = 0.05#
duty_cycle: float = 1.0#
solar_wh_per_day: float = 0.0#
reserve_fraction: float = 0.0#
regulator_efficiency: float = 1.0#
charge_efficiency: float = 1.0#
telemetry_power_w: float = 0.0#
telemetry_seconds_per_day: float = 0.0#
edge_power_w: float = 0.0#
edge_duty_cycle: float = 0.0#
auxiliary_wh_per_day: float = 0.0#
min_runtime_days: float | None = None#
device_id: str | None = None#
metadata: dict[str, Any]#
validate()#

Validate and normalise power-budget inputs.

Return type:

None

property usable_battery_wh: float#

Battery energy available after reserve is held back.

property base_average_power_w: float#

Average active/sleep power before auxiliary loads.

property telemetry_wh_per_day: float#

Daily energy used by radio transmission windows.

property edge_wh_per_day: float#

Daily energy used by edge processing overhead.

property harvest_wh_per_day: float#

Daily usable harvested energy after charge efficiency.

class pycsamt.iot.EnergyEstimate(average_power_w, runtime_hours, runtime_days, net_wh_per_day, load_wh_per_day=0.0, harvest_wh_per_day=0.0, usable_battery_wh=0.0, telemetry_wh_per_day=0.0, edge_wh_per_day=0.0, auxiliary_wh_per_day=0.0, reserve_wh=0.0, energy_margin_wh_per_day=0.0, autonomy_days_no_harvest=0.0, state=PowerState.OK, issues=<factory>)#

Bases: PyCSAMTObject

Estimated runtime and power draw.

Parameters:
average_power_w: float#
runtime_hours: float#
runtime_days: float#
net_wh_per_day: float#
load_wh_per_day: float = 0.0#
harvest_wh_per_day: float = 0.0#
usable_battery_wh: float = 0.0#
telemetry_wh_per_day: float = 0.0#
edge_wh_per_day: float = 0.0#
auxiliary_wh_per_day: float = 0.0#
reserve_wh: float = 0.0#
energy_margin_wh_per_day: float = 0.0#
autonomy_days_no_harvest: float = 0.0#
state: PowerState | str = 'ok'#
issues: list[str]#
validate()#

Validate and normalise estimate fields.

Return type:

None

property sustaining: bool#

Return whether harvested energy covers daily load.

as_dict()#

Return a serialisable estimate dictionary.

Return type:

dict[str, Any]

to_packet(device, *, timestamp, survey_id=None, qos=0, retained=False)#

Encode this estimate as a power telemetry packet.

Parameters:
Return type:

TelemetryPacket

class pycsamt.iot.PowerState(*values)#

Bases: str, Enum

Energy status for an IoT field device.

SUSTAINING = 'sustaining'#
OK = 'ok'#
WARNING = 'warning'#
CRITICAL = 'critical'#
pycsamt.iot.estimate_deployment_energy(configs, *, api=None)#

Estimate and tabulate energy budgets for multiple devices.

Parameters:
Return type:

Any

pycsamt.iot.estimate_energy_budget(config)#

Estimate runtime from battery, duty cycle, and optional solar input.

Parameters:

config (EnergyConfig)

Return type:

EnergyEstimate

pycsamt.iot.power_summary_table(estimates, *, device_ids=None, api=None)#

Return energy estimates as a pyCSAMT table.

Parameters:
Return type:

Any

pycsamt.iot.plot_edge_qc_summary(edge, *, figsize=(12.0, 7.5), title='Edge QC summary', output_path=None, close=False)#

Plot edge quality-control decisions and channel metrics.

Parameters:

  • edge (EdgeProcessingResult, TelemetryPacket, FieldSession, mapping, or iterable) – Edge-processing result(s), QC telemetry packet(s), a field session, or serialised mappings. Session inputs are filtered to QC packets.

  • figsize (tuple) – Matplotlib figure size in inches.

  • title (str) – Figure title.

  • output_path (str, optional) – If given, save the figure to this path.

  • close (bool) – Close the figure before returning it.

Returns:

The QC summary figure. The normalised rows are attached as fig.pycsamt_iot_edge_qc.

Return type:

matplotlib.figure.Figure

pycsamt.iot.plot_field_dashboard(session, *, now=None, figsize=(13.0, 8.0), station_axis='auto', title=None, output_path=None, close=False)#

Plot a compact IoT acquisition dashboard.

Parameters:

  • session (FieldSession or mapping) – Field session, or a mapping produced by pycsamt.iot.FieldSession.to_dict().

  • now (float, optional) – Reference timestamp used for live latency/status calculations.

  • figsize (tuple) – Matplotlib figure size in inches.

  • station_axis ({"auto", "profile", "map"}) – Station layout. "profile" uses profile chainage/index; "map" uses longitude/latitude when all stations have coordinates; "auto" chooses map only when coordinates exist.

  • title (str, optional) – Figure title. Defaults to the survey id.

  • output_path (str, optional) – If given, save the figure to this path.

  • close (bool) – Close the figure before returning it. Useful for batch report generation after saving.

Returns:

The dashboard figure. The computed data are also attached as fig.pycsamt_iot_dashboard for reproducible report workflows.

Return type:

matplotlib.figure.Figure

pycsamt.iot.plot_power_budget(power, *, figsize=(12.0, 7.5), title='IoT power budget', output_path=None, close=False)#

Plot IoT energy budget, runtime, and power-state summaries.

Parameters:

  • power (EnergyConfig, EnergyEstimate, TelemetryPacket, FieldSession, mapping, or iterable) – Power budget input(s). EnergyConfig objects are estimated before plotting. Session inputs are filtered to PacketKind.POWER packets.

  • figsize (tuple) – Matplotlib figure size in inches.

  • title (str) – Figure title.

  • output_path (str, optional) – If given, save the figure to this path.

  • close (bool) – Close the figure before returning it.

Returns:

The power-budget figure. Normalised rows are attached as fig.pycsamt_iot_power_budget.

Return type:

matplotlib.figure.Figure

pycsamt.iot.plot_sync_quality(sync, *, figsize=(12.0, 7.5), title='Clock synchronisation quality', tolerance_ms=1.0, max_drift_ppm=None, max_jitter_ms=None, output_path=None, close=False)#

Plot clock offset, drift, jitter, GPS lock, and quality grades.

Parameters:

  • sync (SyncStatus, TelemetryPacket, FieldSession, mapping, or iterable) – Synchronisation status input(s). Session inputs are filtered to PacketKind.SYNC packets.

  • figsize (tuple) – Matplotlib figure size in inches.

  • title (str) – Figure title.

  • tolerance_ms (float, optional) – Optional visual threshold lines.

  • max_drift_ppm (float, optional) – Optional visual threshold lines.

  • max_jitter_ms (float, optional) – Optional visual threshold lines.

  • output_path (str, optional) – If given, save the figure to this path.

  • close (bool) – Close the figure before returning it.

Returns:

The synchronisation figure. Normalised rows are attached as fig.pycsamt_iot_sync_quality.

Return type:

matplotlib.figure.Figure

class pycsamt.iot.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'#
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.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.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.IoTProtocol(*values)#

Bases: str, Enum

Supported telemetry protocol identifiers.

MQTT = 'mqtt'#
HTTP = 'http'#
LORA = 'lora'#
SERIAL = 'serial'#
FILE = 'file'#
WEBSOCKET = 'websocket'#
class pycsamt.iot.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.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.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.TelemetryAck(ok, protocol, packet_id, detail='')#

Bases: object

Acknowledgement returned by a telemetry client.

Parameters:
ok: bool#
protocol: str#
packet_id: str#
detail: str = ''#
class pycsamt.iot.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:
exception pycsamt.iot.TelemetryError#

Bases: RuntimeError

Raised when a telemetry transport operation fails.

class pycsamt.iot.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.build_telemetry_client(protocol, *, endpoint=None, dry_run=True, **options)#

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

class pycsamt.iot.ClockSynchronizer(config=None)#

Bases: object

Evaluate device clock status against a reference.

Parameters:

config (SyncConfig | None)

assess(device_id, local_timestamps, reference_timestamps, *, gps_lock=None)#

Return offset, drift, jitter, and an overall quality grade.

Parameters:

  • device_id (str)

  • local_timestamps (Any)

  • reference_timestamps (Any)

  • gps_lock (bool | None)

Return type:

SyncStatus

class pycsamt.iot.SyncConfig(tolerance_ms=1.0, reference='gps', max_drift_ppm=None, max_jitter_ms=None, min_reference_points=2)#

Bases: PyCSAMTObject

Clock-synchronisation tolerances.

Parameters:

  • tolerance_ms (float)

  • reference (str)

  • max_drift_ppm (float | None)

  • max_jitter_ms (float | None)

  • min_reference_points (int)

tolerance_ms: float = 1.0#
reference: str = 'gps'#
max_drift_ppm: float | None = None#
max_jitter_ms: float | None = None#
min_reference_points: int = 2#
validate()#

Validate and normalise sync tolerances.

Return type:

None

class pycsamt.iot.SyncQuality(*values)#

Bases: str, Enum

Overall synchronisation grade for a device.

EXCELLENT = 'excellent'#
GOOD = 'good'#
FAIR = 'fair'#
POOR = 'poor'#
UNKNOWN = 'unknown'#
class pycsamt.iot.SyncStatus(device_id, offset_ms, within_tolerance, reference='gps', drift_ppm=nan, jitter_ms=nan, gps_lock=None, n_reference_points=0, quality=SyncQuality.UNKNOWN)#

Bases: PyCSAMTObject

Clock-synchronisation status for one device.

Parameters:
device_id: str#
offset_ms: float#
within_tolerance: bool#
reference: str = 'gps'#
drift_ppm: float = nan#
jitter_ms: float = nan#
gps_lock: bool | None = None#
n_reference_points: int = 0#
quality: SyncQuality | str = 'unknown'#
validate()#

Validate and normalise sync-status fields.

Return type:

None

as_dict()#

Return a flat, serialisable status dictionary.

Return type:

dict[str, Any]

pycsamt.iot.assess_sync_quality(*, offset_ms, drift_ppm=nan, jitter_ms=nan, gps_lock=None, tolerance_ms=1.0)#

Grade synchronisation from offset, drift, jitter, and GPS lock.

Parameters:
Return type:

SyncQuality

pycsamt.iot.batch_assess_sync(references, *, config=None, api=None)#

Assess many devices at once and return a status table.

Parameters:

  • references (mapping) – {device_id: spec} where spec is either a (local, reference) pair or a mapping with local / reference (and optional gps_lock) keys.

  • config (SyncConfig, optional) – Shared tolerances.

  • api (bool, optional) – Frame-wrapping override passed to maybe_wrap_frame().

Return type:

Any

pycsamt.iot.detect_gps_dropout(gps_lock, timestamps=None, *, min_lock_fraction=0.9)#

Summarise GPS-lock dropouts across a sample sequence.

Parameters:

  • gps_lock (iterable) – Per-sample lock flags (bool/0-1). Non-finite entries are treated as unlocked.

  • timestamps (iterable, optional) – Matching sample timestamps (seconds), used to report the longest dropout duration.

  • min_lock_fraction (float) – Threshold below which ok is False.

Returns:

Keys: n_samples, n_locked, lock_fraction, n_dropout_events, longest_dropout_samples, longest_dropout_s, and ok.

Return type:

dict

pycsamt.iot.estimate_clock_drift_ppm(local_timestamps, reference_timestamps)#

Estimate clock drift in parts-per-million.

Fits the local-reference offset (in seconds) against reference time and returns the slope scaled to ppm. Requires at least two distinct reference points; otherwise returns nan.

Parameters:

  • local_timestamps (Any)

  • reference_timestamps (Any)

Return type:

float

pycsamt.iot.estimate_clock_jitter_ms(local_timestamps, reference_timestamps)#

Estimate timing jitter as the std of drift-corrected offsets (ms).

Parameters:

  • local_timestamps (Any)

  • reference_timestamps (Any)

Return type:

float

pycsamt.iot.estimate_clock_offset_ms(local_timestamps, reference_timestamps)#

Estimate median local-reference clock offset in milliseconds.

Parameters:

  • local_timestamps (Any)

  • reference_timestamps (Any)

Return type:

float

pycsamt.iot.sync_status_table(statuses, *, api=None)#

Return one row per device from sync-status objects.

Parameters:
Return type:

Any

class pycsamt.iot.AcquisitionManifest(survey_id, created_utc=<factory>, tool='pycsamt.iot', tool_version=<factory>, method=None, operator=None, records=<factory>, devices=<factory>, files=<factory>, qc_decisions=<factory>, processing_steps=<factory>, environment=<factory>, metadata=<factory>)#

Bases: PyCSAMTObject

Reproducible manifest describing one IoT acquisition session.

Parameters:
survey_id: str#
created_utc: str#
tool: str = 'pycsamt.iot'#
tool_version: str#
method: str | None = None#
operator: str | None = None#
records: list[ProvenanceRecord]#
devices: list[dict[str, Any]]#
files: list[dict[str, Any]]#
qc_decisions: list[dict[str, Any]]#
processing_steps: list[str]#
environment: dict[str, Any]#
metadata: dict[str, Any]#
validate()#

Validate object state.

Subclasses can override this hook. The base implementation intentionally accepts all states.

Return type:

None

add_record(record)#
Parameters:

record (ProvenanceRecord | Mapping[str, Any])

Return type:

None

add_file(path, *, algo='sha256')#
Parameters:
Return type:

dict[str, Any]

add_qc_decision(**kwargs)#
Parameters:

kwargs (Any)

Return type:

dict[str, Any]

add_processing_step(step)#
Parameters:

step (str)

Return type:

None

as_dict()#
Return type:

dict[str, Any]

to_json(*, indent=2)#
Parameters:

indent (int)

Return type:

str

write(path, *, indent=2)#
Parameters:
Return type:

str

chained_qc_decisions()#

Return the QC decisions as a tamper-evident hash chain.

Return type:

list[dict[str, Any]]

sign(key, *, algo='sha256')#

Return an HMAC-signed envelope around the manifest.

The result wraps the manifest payload under manifest alongside a signature and signature_algo, so it can be written out and later checked with verify_manifest() by a holder of key.

Parameters:
Return type:

dict[str, Any]

write_signed(path, key, *, algo='sha256', indent=2)#

Write an HMAC-signed manifest envelope to path (JSON).

Parameters:
Return type:

str

class pycsamt.iot.ProvenanceRecord(station_id, instrument_serial=None, firmware=None, operator=None, lat=None, lon=None, elevation=None, ex_azimuth_deg=None, ey_azimuth_deg=None, occupation_start=None, occupation_end=None, sample_rate_hz=None, gps_quality=None, battery_status=None, accepted_band_hz=None, field_notes='', qc_decisions=<factory>, rejected_windows=<factory>, processing_steps=<factory>, raw_files=<factory>)#

Bases: PyCSAMTObject

Per-station occupation provenance for one field session.

Parameters:
station_id: str#
instrument_serial: str | None = None#
firmware: str | None = None#
operator: str | None = None#
lat: float | None = None#
lon: float | None = None#
elevation: float | None = None#
ex_azimuth_deg: float | None = None#
ey_azimuth_deg: float | None = None#
occupation_start: float | None = None#
occupation_end: float | None = None#
sample_rate_hz: float | None = None#
gps_quality: str | None = None#
battery_status: str | None = None#
accepted_band_hz: tuple[float, float] | None = None#
field_notes: str = ''#
qc_decisions: list[dict[str, Any]]#
rejected_windows: list[Any]#
processing_steps: list[str]#
raw_files: list[dict[str, Any]]#
validate()#

Validate object state.

Subclasses can override this hook. The base implementation intentionally accepts all states.

Return type:

None

property occupation_seconds: float | None#

Occupation duration in seconds, if both bounds are known.

add_raw_file(path, *, algo='sha256')#

Hash path and attach the integrity record.

Parameters:
Return type:

dict[str, Any]

as_dict()#
Return type:

dict[str, Any]

pycsamt.iot.build_acquisition_manifest(survey_id, *, records=None, devices=None, qc_decisions=None, processing_steps=None, method=None, operator=None, metadata=None)#

Assemble an AcquisitionManifest from session components.

Parameters:
Return type:

AcquisitionManifest

pycsamt.iot.export_acquisition_manifest(survey_id, path, *, indent=2, **kwargs)#

Build an acquisition manifest and write it to path.

Thin convenience over build_acquisition_manifest() + AcquisitionManifest.write(); kwargs are forwarded to the builder (records, devices, qc_decisions, method, …). Returns the written path.

Parameters:
Return type:

str

pycsamt.iot.export_reproducibility_bundle(manifest, out_dir, *, include_raw=False, zip_bundle=False)#

Write a manifest and per-station audits into out_dir.

Parameters:

  • manifest (AcquisitionManifest) – The session manifest to export.

  • out_dir (str) – Destination directory (created if needed).

  • include_raw (bool) – When True, copy referenced raw files into out_dir/raw if they exist on disk.

  • zip_bundle (bool) – When True, also produce <out_dir>.zip of the bundle.

Returns:

Paths written: manifest, audits, optional raw and zip.

Return type:

dict

pycsamt.iot.export_station_audit(record, path, *, indent=2)#

Write a single-station provenance audit to path (JSON).

Parameters:
Return type:

str

pycsamt.iot.hash_bytes(data, *, algo='sha256')#

Return the hex digest of data using algo.

Parameters:
Return type:

str

pycsamt.iot.hash_chain(entries, *, algo='sha256', genesis='')#

Return a tamper-evident hash chain over entries.

Each returned entry is a copy of the input with seq, prev_hash, and entry_hash added, where entry_hash folds in the previous entry’s hash. Altering, inserting, or reordering any entry breaks the chain from that point on (detected by verify_hash_chain()). This suits a running log such as per-window QC decisions.

Parameters:
Return type:

list[dict[str, Any]]

pycsamt.iot.hash_mapping(mapping, *, algo='sha256')#

Return a stable hash of mapping (key-sorted JSON, UTF-8).

Parameters:
Return type:

str

pycsamt.iot.hash_raw_file(path, *, algo='sha256', chunk_size=1048576)#

Hash a raw acquisition file and return its integrity record.

Returns a mapping with path, bytes, algo, digest, and modified_utc. Raises FileNotFoundError when the file is missing.

Parameters:
Return type:

dict[str, Any]

pycsamt.iot.log_qc_decision(station, decision, *, channel=None, reasons=None, operator=None, timestamp=None, window=None)#

Return a normalised QC-decision record for the audit trail.

Parameters:
Return type:

dict[str, Any]

pycsamt.iot.sign_mapping(mapping, key, *, algo='sha256')#

Return an HMAC signature over the canonical JSON of mapping.

Unlike hash_mapping(), which anyone can recompute, an HMAC signature also proves the manifest was produced by a party holding the shared key – useful for asserting a field audit trail has not been tampered with in transit.

Parameters:
Return type:

str

pycsamt.iot.verify_hash_chain(chained, *, algo='sha256', genesis='')#

Return whether a hash_chain() sequence is intact and in order.

Parameters:
Return type:

bool

pycsamt.iot.verify_manifest(signed, key)#

Verify a signed manifest produced by AcquisitionManifest.sign().

Checks the HMAC signature over the wrapped manifest payload and, when present, that the payload’s own content_hash still matches.

Parameters:
Return type:

bool

pycsamt.iot.verify_signature(mapping, signature, key, *, algo='sha256')#

Return whether signature is a valid HMAC of mapping under key.

Uses a constant-time comparison so verification does not leak timing information.

Parameters:
Return type:

bool

class pycsamt.iot.AuthScheme(*values)#

Bases: str, Enum

Supported authentication schemes.

NONE = 'none'#
BEARER = 'bearer'#
BASIC = 'basic'#
API_KEY = 'api_key'#
class pycsamt.iot.Credential(scheme=AuthScheme.NONE, token=None, username=None, password=None, api_key=None, api_key_header='X-API-Key')#

Bases: PyCSAMTObject

Authentication credential with secret redaction.

Secrets (token, password, api_key) are never shown by repr or as_dict(); use reveal() explicitly when a real transport needs the raw values.

Parameters:
scheme: AuthScheme | str = 'none'#
token: str | None = None#
username: str | None = None#
password: str | None = None#
api_key: str | None = None#
api_key_header: str = 'X-API-Key'#
validate()#

Validate object state.

Subclasses can override this hook. The base implementation intentionally accepts all states.

Return type:

None

headers()#

Return HTTP headers implementing the configured scheme.

Return type:

dict[str, str]

reveal()#

Return the raw credential values (handle with care).

Return type:

dict[str, Any]

as_dict()#

Return a redacted, serialisable credential summary.

Return type:

dict[str, Any]

class pycsamt.iot.SecurityConfig(tls=<factory>, credential=<factory>, require_tls=False, allowed_protocols=None)#

Bases: PyCSAMTObject

Combined TLS + credential policy for telemetry transports.

Parameters:
tls: TLSConfig#
credential: Credential#
require_tls: bool = False#
allowed_protocols: list[str] | None = None#
validate()#

Validate object state.

Subclasses can override this hook. The base implementation intentionally accepts all states.

Return type:

None

allows(protocol)#

Return whether protocol is permitted by policy.

Parameters:

protocol (str)

Return type:

bool

client_options()#

Return options for build_telemetry_client().

Includes TLS flags, credential headers, and (for MQTT) username / password so a transport can authenticate. Raw secrets are included because this feeds the live client, not a log.

Return type:

dict[str, Any]

as_dict()#

Return a redacted, serialisable security summary.

Return type:

dict[str, Any]

classmethod from_env(prefix='PYCSAMT_IOT_')#

Build a security config from environment variables.

Recognised variables (with prefix): TOKEN, API_KEY, USERNAME, PASSWORD, TLS (truthy), CA_CERT, CERTFILE, KEYFILE.

Parameters:

prefix (str)

Return type:

SecurityConfig

class pycsamt.iot.TLSConfig(enabled=False, ca_cert=None, certfile=None, keyfile=None, verify=True, min_version=None)#

Bases: PyCSAMTObject

Transport-layer security material for a telemetry client.

Parameters:

  • enabled (bool)

  • ca_cert (str | None)

  • certfile (str | None)

  • keyfile (str | None)

  • verify (bool)

  • min_version (str | None)

enabled: bool = False#
ca_cert: str | None = None#
certfile: str | None = None#
keyfile: str | None = None#
verify: bool = True#
min_version: str | None = None#
validate()#

Validate object state.

Subclasses can override this hook. The base implementation intentionally accepts all states.

Return type:

None

as_dict()#

Return TLS configuration (paths are not secrets).

Return type:

dict[str, Any]

pycsamt.iot.redact_secret(value)#

Return a redaction placeholder when value is a non-empty secret.

Parameters:

value (Any)

Return type:

str | None

pycsamt.iot.simulate_amt_channel(n_samples, sample_rate, *, snr_db=20.0, mains_hz=50.0, powerline_amplitude=0.0, dropout_rate=0.0, seed=None)#

Simulate one AMT channel: band-limited signal + noise + artefacts.

Parameters:
Return type:

ndarray

pycsamt.iot.simulate_amt_station(station_id, *, channels=('ex', 'ey', 'hx', 'hy'), sample_rate=128.0, n_samples=1024, mains_hz=50.0, snr_db=20.0, powerline_amplitude=0.2, dropout_rate=0.0, survey_id=None, profile=None, position_m=None, lat=None, lon=None, timestamp=None, seed=None)#

Simulate a full AMT station: config, channel data, and packets.

Returns:

Keys: station (StationConfig), device (DeviceConfig), data ({channel: ndarray}), and packets (list of TelemetryPacket: one health, one QC).

Return type:

dict

Parameters:
pycsamt.iot.simulate_battery_decay(n_samples, *, initial_v=13.2, final_v=10.5, noise_v=0.05, seed=None)#

Simulate a monotonic-ish battery discharge curve with noise.

Parameters:
Return type:

ndarray

pycsamt.iot.simulate_gps_drift(n_samples, *, sample_interval_s=1.0, drift_ppm=5.0, jitter_ms=0.2, offset_ms=0.0, dropout_rate=0.0, start_time=1700000000.0, seed=None)#

Simulate paired reference/local clocks with drift, jitter, dropout.

Returns:

Keys reference and local (POSIX-second arrays) and gps_lock (bool array). Feed local/reference straight into estimate_clock_drift_ppm() etc.

Return type:

dict

Parameters:
pycsamt.iot.simulate_iot_network(n_stations=10, *, profiles=('L1',), channels=('ex', 'ey', 'hx', 'hy'), sample_rate=128.0, n_samples=512, mains_hz=50.0, snr_db=20.0, dropout_rate=0.05, survey_id='SIM', station_spacing_m=50.0, seed=None, detail=False)#

Simulate a network of AMT stations across one or more profiles.

Parameters:

  • n_stations (int) – Total number of stations to generate.

  • profiles (sequence of str) – Profile/line labels; stations are round-robin assigned to them.

  • detail (bool) – When True return a dict with stations and packets. When False (default) return just the flat list of packets, as in the documented example.

  • channels (Sequence[str])

  • sample_rate (float)

  • n_samples (int)

  • mains_hz (float)

  • snr_db (float)

  • dropout_rate (float)

  • survey_id (str)

  • station_spacing_m (float)

  • seed (int | None)

Return type:

list of TelemetryPacket, or dict when detail=True.

pycsamt.iot.simulate_packet_loss(packets, dropout_rate=0.05, *, seed=None)#

Return packets with a fraction randomly dropped.

Parameters:
Return type:

list[TelemetryPacket]

pycsamt.iot.simulate_powerline_noise(n_samples, sample_rate, *, mains_hz=50.0, amplitude=1.0, n_harmonics=3, seed=None)#

Return a powerline-noise time series (fundamental + harmonics).

Parameters:
Return type:

ndarray

Modules#

pycsamt.iot.core

pycsamt.iot.station

Field-station configuration for IoT-enabled AMT/CSAMT acquisition.

pycsamt.iot.schemas

Formal telemetry payload schemas.

pycsamt.iot.session

Field acquisition session — the hub of the IoT subpackage.

pycsamt.iot.protocols

Telemetry transports for IoT-enabled field acquisition.

pycsamt.iot.protocols.base

Telemetry transport base classes and the dry-run recorder.

pycsamt.iot.protocols.file

File-backed telemetry transport (newline-delimited JSON).

pycsamt.iot.protocols.http

HTTP(S) telemetry transport using the standard library.

pycsamt.iot.protocols.mqtt

MQTT telemetry transport (optional dependency: paho-mqtt).

pycsamt.iot.protocols.serial

Serial (UART) telemetry transport (optional dependency: pyserial).

pycsamt.iot.protocols.store_forward

Store-and-forward buffering for intermittent field links.

pycsamt.iot.protocols.websocket

WebSocket telemetry transport (optional dependency: websocket-client).

pycsamt.iot.edge

pycsamt.iot.edge_amt

AMT/CSAMT-specific edge quality-control metrics.

pycsamt.iot.edge_csamt

CSAMT / controlled-source specific edge quality-control metrics.

pycsamt.iot.edge_csem

CSEM controlled-source electromagnetic edge quality control.

pycsamt.iot.methods

Per-method acquisition profiles for method-aware IoT QC.

pycsamt.iot.bridge

Bridge between IoT field telemetry and the pyCSAMT data model.

pycsamt.iot.sync

Clock synchronisation audit for IoT field acquisition.

pycsamt.iot.power

pycsamt.iot.monitoring

pycsamt.iot.plot

Visualisation helpers for IoT-enabled field acquisition.

pycsamt.iot.provenance

Acquisition provenance and reproducibility for IoT field sessions.

pycsamt.iot.security

Security configuration for IoT telemetry transports.

pycsamt.iot.sim

Synthetic IoT/AMT field-network simulator.
On this page