pycsamt.iot.sync#

Clock synchronisation audit for IoT field acquisition.

Offset alone is not enough to certify field-grade timing. This module adds clock drift (ppm), jitter (ms), GPS-lock accounting, and an overall SyncQuality grade, plus batch assessment and GPS-dropout detection across a deployment.

Functions

assess_sync_quality(*, offset_ms[, ...])

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

batch_assess_sync(references, *[, config, api])

Assess many devices at once and return a status table.

detect_gps_dropout(gps_lock[, timestamps, ...])

Summarise GPS-lock dropouts across a sample sequence.

estimate_clock_drift_ppm(local_timestamps, ...)

Estimate clock drift in parts-per-million.

estimate_clock_jitter_ms(local_timestamps, ...)

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

estimate_clock_offset_ms(local_timestamps, ...)

Estimate median local-reference clock offset in milliseconds.

sync_status_table(statuses, *[, api])

Return one row per device from sync-status objects.

Classes

ClockSynchronizer([config])

Evaluate device clock status against a reference.

SyncConfig([tolerance_ms, reference, ...])

Clock-synchronisation tolerances.

SyncQuality(*values)

Overall synchronisation grade for a device.

SyncStatus(device_id, offset_ms, ...[, ...])

Clock-synchronisation status for one device.
class pycsamt.iot.sync.SyncConfig(tolerance_ms=1.0, reference='gps', max_drift_ppm=None, max_jitter_ms=None, min_reference_points=2)[source]#

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

Validate and normalise sync tolerances.

Return type:

None

class pycsamt.iot.sync.SyncQuality(*values)[source]#

Bases: str, Enum

Overall synchronisation grade for a device.

EXCELLENT = 'excellent'#
GOOD = 'good'#
FAIR = 'fair'#
POOR = 'poor'#
UNKNOWN = 'unknown'#
class pycsamt.iot.sync.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)[source]#

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

Validate and normalise sync-status fields.

Return type:

None

as_dict()[source]#

Return a flat, serialisable status dictionary.

Return type:

dict[str, Any]

class pycsamt.iot.sync.ClockSynchronizer(config=None)[source]#

Bases: object

Evaluate device clock status against a reference.

Parameters:

config (SyncConfig | None)

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

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

pycsamt.iot.sync.estimate_clock_offset_ms(local_timestamps, reference_timestamps)[source]#

Estimate median local-reference clock offset in milliseconds.

Parameters:

  • local_timestamps (Any)

  • reference_timestamps (Any)

Return type:

float

pycsamt.iot.sync.estimate_clock_drift_ppm(local_timestamps, reference_timestamps)[source]#

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.sync.estimate_clock_jitter_ms(local_timestamps, reference_timestamps)[source]#

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

Parameters:

  • local_timestamps (Any)

  • reference_timestamps (Any)

Return type:

float

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

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

Parameters:
Return type:

SyncQuality

pycsamt.iot.sync.batch_assess_sync(references, *, config=None, api=None)[source]#

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.sync.detect_gps_dropout(gps_lock, timestamps=None, *, min_lock_fraction=0.9)[source]#

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.sync.sync_status_table(statuses, *, api=None)[source]#

Return one row per device from sync-status objects.

Parameters:
Return type:

Any