plesty.lib.device.telemetry =========================== .. py:module:: plesty.lib.device.telemetry .. autoapi-nested-parse:: 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: * :class:`DeviceStatus` — pull-style point-in-time snapshot answering "what is this device doing right now"; assembled for free by :meth:`TelemetrySystem.status`. * :class:`TelemetryEvent` — push-style timestamped record ("a reading was taken", "a parameter changed", "an error occurred") delivered through hooks registered with :meth:`TelemetrySystem.register_telemetry_hook`. * :class:`TelemetrySystem` — the mixin composing this contract into device models, alongside :class:`~plesty.lib.device.params.ConfigSystem` and :class:`~plesty.lib.device.funcs.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 ---------- .. autoapisummary:: plesty.lib.device.telemetry.TELEMETRY_KINDS plesty.lib.device.telemetry.CONNECTION_STATES Classes ------- .. autoapisummary:: plesty.lib.device.telemetry.TelemetryEvent plesty.lib.device.telemetry.DeviceStatus plesty.lib.device.telemetry.TelemetrySystem Functions --------- .. autoapisummary:: plesty.lib.device.telemetry.utc_timestamp plesty.lib.device.telemetry.append_events plesty.lib.device.telemetry.read_events Module Contents --------------- .. py:data:: TELEMETRY_KINDS :type: tuple[str, Ellipsis] :value: ('reading', 'state_change', 'error', 'warning') .. py:data:: CONNECTION_STATES :type: tuple[str, Ellipsis] :value: ('disconnected', 'connecting', 'connected', 'error') .. py:function:: utc_timestamp() -> str Return the current UTC time as an ISO-8601 string. .. py:class:: TelemetryEvent A single timestamped telemetry record emitted by a device. :ivar device: Identifier of the emitting device (``BaseDeviceSyncModel.id``). :ivar kind: Event category — one of :data:`TELEMETRY_KINDS`. :ivar name: The parameter or metric the event refers to (e.g. ``"power"``). :ivar value: The reading, new value, or error message. :ivar unit: Physical unit of *value*, if any. :ivar timestamp: UTC ISO-8601 creation time; filled automatically when empty. :ivar context: Optional extra JSON-serializable context (step id, client, …). .. py:attribute:: device :type: str .. py:attribute:: kind :type: str .. py:attribute:: name :type: str .. py:attribute:: value :type: Any :value: None .. py:attribute:: unit :type: str | None :value: None .. py:attribute:: timestamp :type: str :value: '' .. py:attribute:: context :type: dict[str, Any] | None :value: None .. py:method:: __post_init__() -> None Validate the event kind and stamp the creation time when missing. .. py:method:: to_dict() -> dict[str, Any] Return the event as a JSON-serializable dictionary. .. py:method:: from_dict(document: dict[str, Any]) -> TelemetryEvent :classmethod: Reconstruct an event from a dictionary produced by :meth:`to_dict`. :param document: The parsed JSON document. :returns: The reconstructed :class:`TelemetryEvent`. .. py:class:: DeviceStatus A point-in-time snapshot of a device's operational state. :ivar device: Identifier of the device (``BaseDeviceSyncModel.id``). :ivar connection: Connection state — one of :data:`CONNECTION_STATES`. :ivar state: Operational state (e.g. ``"ready"``, ``"busy"``, ``"acquiring"``); free-form so devices can refine it. :ivar parameters: Last-known configuration parameter values (no device I/O). :ivar last_error: Most recent error message, if any. :ivar timestamp: UTC ISO-8601 snapshot time; filled automatically when empty. .. py:attribute:: device :type: str .. py:attribute:: connection :type: str .. py:attribute:: state :type: str .. py:attribute:: parameters :type: dict[str, Any] .. py:attribute:: last_error :type: str | None :value: None .. py:attribute:: timestamp :type: str :value: '' .. py:method:: __post_init__() -> None Validate the connection state and stamp the snapshot time when missing. .. py:method:: to_dict() -> dict[str, Any] Return the snapshot as a JSON-serializable dictionary. .. py:method:: from_dict(document: dict[str, Any]) -> DeviceStatus :classmethod: Reconstruct a snapshot from a dictionary produced by :meth:`to_dict`. :param document: The parsed JSON document. :returns: The reconstructed :class:`DeviceStatus`. .. py:function:: 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). :param path: Destination log file; parent directories are created. :param events: The events to append. :returns: The resolved path of the log file. .. py:function:: read_events(path: str | pathlib.Path) -> list[TelemetryEvent] Read telemetry events from a JSONL log written by :func:`append_events`. A torn (partially written) line — possible after a crash mid-write — is silently skipped, mirroring the experiment journal's replay semantics. :param path: Path of the JSONL log file. :returns: The events in file order. .. py:class:: 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. .. py:attribute:: id :type: str .. py:attribute:: _telemetry_hooks :type: list[Callable[[TelemetryEvent], None]] :value: [] .. py:attribute:: _last_error :type: str | None :value: None .. py:method:: register_telemetry_hook(hook: Callable[[TelemetryEvent], None]) -> None Register a callback invoked for every telemetry event this device emits. Hooks receive :class:`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. :param hook: The callback to register; duplicates are ignored. .. py:method:: unregister_telemetry_hook(hook: Callable[[TelemetryEvent], None]) -> None Remove a previously registered telemetry hook. :param hook: The callback to remove; unknown hooks are ignored. .. py:method:: 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 :meth:`status`. A failing hook is logged and skipped — telemetry delivery never disturbs device operation. :param event: The event to deliver. .. py:method:: _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. :param kind: Event category — one of :data:`TELEMETRY_KINDS`. :param name: The configuration parameter key. :param value: The written value or parsed reading. .. py:method:: device_state() -> str Return the operational state reported by :meth:`status`. The base implementation only distinguishes ``"ready"`` from ``"unavailable"`` via :meth:`check_operatability`; device models override this to report finer states such as ``"busy"``, ``"moving"``, or ``"acquiring"``. .. py:method:: 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 :class:`DeviceStatus`. .. py:method:: check_operatability() -> bool :abstractmethod: Check if the device is currently operatable — provided by the device model. .. py:method:: get_config(key: str, group: str = 'default') -> Any :abstractmethod: Return a registered configuration parameter — provided by ``ConfigSystem``. .. py:method:: get_config_list(group: str = 'default') -> list[str] :abstractmethod: Return the registered configuration keys — provided by ``ConfigSystem``.