pycsamt.iot.edge

Functions

edge_summary_table(result, *[, api])

Return one row per channel from edge-processing results.

Classes

EdgeChannelSummary(channel, n_sample, ...[, ...])	Quality-control summary for one edge data channel.
EdgeDecision(*values)	Acceptance state assigned by edge-side quality control.
EdgeProcessingConfig([decimation, ...])	Configuration for lightweight field-side processing.
EdgeProcessingResult(data, metrics, accepted)	Summary returned by EdgeProcessor.
EdgeProcessor([config])	Small edge-processing block for telemetry payload reduction.

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

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

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:

• device (DeviceConfig)

• timestamp (float)

• survey_id (str | None)

• qos (int)

• retained (bool)

• include_data (bool | None)

Return type:

TelemetryPacket

class pycsamt.iot.edge.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:

• data (Any)

• channel_names (Iterable[str] | None)

Return type:

EdgeProcessingResult

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

Return one row per channel from edge-processing results.

Parameters:

• result (EdgeProcessingResult | Iterable[EdgeProcessingResult])

• api (bool | None)

Return type:

Any