plesty.lib.device.telemetry

Operational telemetry for Plesty devices (issue plesty-lib#4).

Defines the unified contract for real-time status, state, and diagnostic data so that every device reports telemetry in the same shape and monitors, experiments, and services need no per-device parsing:

• DeviceStatus — pull-style point-in-time snapshot answering “what is this device doing right now”; assembled for free by TelemetrySystem.status().

• TelemetryEvent — push-style timestamped record (“a reading was taken”, “a parameter changed”, “an error occurred”) delivered through hooks registered with TelemetrySystem.register_telemetry_hook().

• TelemetrySystem — the mixin composing this contract into device models, alongside ConfigSystem and FunctionSystem.

The module owns no polling loops, threads, or transports — consumers (monitors, services, experiments) register a hook and decide what to do with the events. Minimal append-only JSONL persistence helpers are provided in the same crash-tolerant format used by the experiment journal.

Attributes

TELEMETRY_KINDS
CONNECTION_STATES

Classes

TelemetryEvent	A single timestamped telemetry record emitted by a device.
DeviceStatus	A point-in-time snapshot of a device's operational state.
TelemetrySystem	Manage telemetry hooks, event emission, and status snapshots for a device.

Functions

utc_timestamp(→ str)	Return the current UTC time as an ISO-8601 string.
append_events(→ pathlib.Path)	Append telemetry events to a JSONL log, one JSON document per line.
read_events(→ list[TelemetryEvent])	Read telemetry events from a JSONL log written by append_events().

Module Contents

plesty.lib.device.telemetry.TELEMETRY_KINDS: tuple[str, Ellipsis] = ('reading', 'state_change', 'error', 'warning')

plesty.lib.device.telemetry.CONNECTION_STATES: tuple[str, Ellipsis] = ('disconnected', 'connecting', 'connected', 'error')

plesty.lib.device.telemetry.utc_timestamp() → str

Return the current UTC time as an ISO-8601 string.

Return type:

str

class plesty.lib.device.telemetry.TelemetryEvent

A single timestamped telemetry record emitted by a device.

Variables:

• device – Identifier of the emitting device (BaseDeviceSyncModel.id).

• kind – Event category — one of TELEMETRY_KINDS.

• name – The parameter or metric the event refers to (e.g. "power").

• value – The reading, new value, or error message.

• unit – Physical unit of value, if any.

• timestamp – UTC ISO-8601 creation time; filled automatically when empty.

• context – Optional extra JSON-serializable context (step id, client, …).

device: str

kind: str

name: str

value: Any = None

unit: str | None = None

timestamp: str = ''

context: dict[str, Any] | None = None

__post_init__() → None

Validate the event kind and stamp the creation time when missing.

Return type:

None

to_dict() → dict[str, Any]

Return the event as a JSON-serializable dictionary.

Return type:

dict[str, Any]

classmethod from_dict(document: dict[str, Any]) → TelemetryEvent

Reconstruct an event from a dictionary produced by to_dict().

Parameters:

document (dict[str, Any]) – The parsed JSON document.

Returns:

The reconstructed TelemetryEvent.

Return type:

TelemetryEvent

class plesty.lib.device.telemetry.DeviceStatus

A point-in-time snapshot of a device’s operational state.

Variables:

• device – Identifier of the device (BaseDeviceSyncModel.id).

• connection – Connection state — one of CONNECTION_STATES.

• state – Operational state (e.g. "ready", "busy", "acquiring"); free-form so devices can refine it.

• parameters – Last-known configuration parameter values (no device I/O).

• last_error – Most recent error message, if any.

• timestamp – UTC ISO-8601 snapshot time; filled automatically when empty.

device: str

connection: str

state: str

parameters: dict[str, Any]

last_error: str | None = None

timestamp: str = ''

__post_init__() → None

Validate the connection state and stamp the snapshot time when missing.

Return type:

None

to_dict() → dict[str, Any]

Return the snapshot as a JSON-serializable dictionary.

Return type:

dict[str, Any]

classmethod from_dict(document: dict[str, Any]) → DeviceStatus

Reconstruct a snapshot from a dictionary produced by to_dict().

Parameters:

document (dict[str, Any]) – The parsed JSON document.

Returns:

The reconstructed DeviceStatus.

Return type:

DeviceStatus

plesty.lib.device.telemetry.append_events(path: str | pathlib.Path, events: Iterable[TelemetryEvent]) → pathlib.Path

Append telemetry events to a JSONL log, one JSON document per line.

The write is flushed and fsynced so events survive a process crash; the format matches the experiment journal (append-only JSONL).

Parameters:

• path (str | pathlib.Path) – Destination log file; parent directories are created.

• events (Iterable[TelemetryEvent]) – The events to append.

Returns:

The resolved path of the log file.

Return type:

pathlib.Path

plesty.lib.device.telemetry.read_events(path: str | pathlib.Path) → list[TelemetryEvent]

Read telemetry events from a JSONL log written by append_events().

A torn (partially written) line — possible after a crash mid-write — is silently skipped, mirroring the experiment journal’s replay semantics.

Parameters:

path (str | pathlib.Path) – Path of the JSONL log file.

Returns:

The events in file order.

Return type:

list[TelemetryEvent]

class plesty.lib.device.telemetry.TelemetrySystem

Manage telemetry hooks, event emission, and status snapshots for a device.

Relies on the composing device model for its identity (id), the operatability check, and cached configuration access — the stub methods at the end of the class document that contract and are shadowed by the real implementations in the device model’s MRO.

Initialize the hook registry and last-error record.

id: str

_telemetry_hooks: list[Callable[[TelemetryEvent], None]] = []

_last_error: str | None = None

register_telemetry_hook(hook: Callable[[TelemetryEvent], None]) → None

Register a callback invoked for every telemetry event this device emits.

Hooks receive TelemetryEvent instances synchronously on the emitting thread; they should return quickly and must not raise (exceptions are caught and logged). Typical consumers append events to a JSONL log or forward them to a monitor.

Parameters:

hook (Callable[[TelemetryEvent], None]) – The callback to register; duplicates are ignored.

Return type:

None

unregister_telemetry_hook(hook: Callable[[TelemetryEvent], None]) → None

Remove a previously registered telemetry hook.

Parameters:

hook (Callable[[TelemetryEvent], None]) – The callback to remove; unknown hooks are ignored.

Return type:

None

emit_telemetry(event: TelemetryEvent) → None

Deliver a telemetry event to all registered hooks.

Events of kind "error" additionally update the device’s last-error record surfaced by status(). A failing hook is logged and skipped — telemetry delivery never disturbs device operation.

Parameters:

event (TelemetryEvent) – The event to deliver.

Return type:

None

_emit_param_event(kind: str, name: str, value: Any) → None

Emit a parameter-related event, skipping all work when nobody listens.

Used by the device model’s write/query paths to report parameter changes ("state_change") and readings ("reading") for free.

Parameters:

• kind (str) – Event category — one of TELEMETRY_KINDS.

• name (str) – The configuration parameter key.

• value (Any) – The written value or parsed reading.

Return type:

None

device_state() → str

Return the operational state reported by status().

The base implementation only distinguishes "ready" from "unavailable" via check_operatability(); device models override this to report finer states such as "busy", "moving", or "acquiring".

Return type:

str

status() → DeviceStatus

Return a point-in-time snapshot of the device’s operational state.

The snapshot is assembled from the last-known (cached) configuration values — no device I/O is performed, so calling it is always cheap and side-effect free. Use the device model’s parameter synchronization first when fresh values are required.

Returns:

The current DeviceStatus.

Return type:

DeviceStatus

abstractmethod check_operatability() → bool

Check if the device is currently operatable — provided by the device model.

Return type:

bool

abstractmethod get_config(key: str, group: str = 'default') → Any

Return a registered configuration parameter — provided by ConfigSystem.

Parameters:

• key (str)

• group (str)

Return type:

Any

abstractmethod get_config_list(group: str = 'default') → list[str]

Return the registered configuration keys — provided by ConfigSystem.

Parameters:

group (str)

Return type:

list[str]