2.32.4.13. 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)[source]

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

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

Validate and normalise the channel summary.

Return type:

None

as_dict()[source]

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

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

Validate and normalise edge-processing options.

Return type:

None

class pycsamt.iot.edge.EdgeProcessingResult(data, metrics, accepted, reasons=<factory>, channels=<factory>, decision_override=None)[source]

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

Validate and normalise the processing result.

Return type:

None

property decision: EdgeDecision[source]

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

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

Encode this edge result as a QC telemetry packet.

Parameters:
Return type:

TelemetryPacket

class pycsamt.iot.edge.EdgeProcessor(config=None)[source]

Bases: PyCSAMTObject

Small edge-processing block for telemetry payload reduction.

Parameters:

config (EdgeProcessingConfig | None)

process(data, *, channel_names=None)[source]

Decimate numeric data and compute simple quality metrics.

Parameters:
Return type:

EdgeProcessingResult

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

Return one row per channel from edge-processing results.

Parameters:
Return type:

Any