plesty.lib.monitor.sources ========================== .. py:module:: plesty.lib.monitor.sources .. autoapi-nested-parse:: Data sources feeding monitors: run directories, telemetry, manual pushes. Every source answers one question — *what appeared since I last asked?* — through a non-blocking :meth:`poll`. That single method is the whole subscription contract (:class:`~plesty.lib.monitor.base_monitor.DataSource`), and it is deliberately pull-shaped even for push-style producers: a telemetry hook fires on the device's thread and only buffers, while the frames are handed out on the consumer's thread. No source starts a thread, opens a socket, or blocks, so the clock always belongs to the caller — a GUI timer, a test loop, a notebook cell. Three sources cover the platform's streams: * :class:`RunSource` — tails a running experiment's ``records.jsonl`` (``//``) by byte offset and emits one frame per newly committed step record. This is the *generic* live view of any PLESTY experiment: the record line is the commit record of the write, so following it needs no cooperation from the experiment and cannot disturb it. * :class:`TelemetrySource` — registers a :class:`~plesty.lib.device.telemetry.TelemetryEvent` hook on one or more devices and turns each event into a frame. * :class:`PushSource` — a thread-safe hand-fed buffer for tests, notebooks, and code that already has the values. * :class:`ReplaySource` — a finished run rationed a few rows per poll, so a stored measurement plays back at the pace it happened. Attributes ---------- .. autoapisummary:: plesty.lib.monitor.sources.FrameMapper plesty.lib.monitor.sources.LEGACY_PATTERN Classes ------- .. autoapisummary:: plesty.lib.monitor.sources.PushSource plesty.lib.monitor.sources.TelemetrySource plesty.lib.monitor.sources.RunSource plesty.lib.monitor.sources.ReplaySource Functions --------- .. autoapisummary:: plesty.lib.monitor.sources._as_record plesty.lib.monitor.sources._read_legacy plesty.lib.monitor.sources._read_line plesty.lib.monitor.sources.records Module Contents --------------- .. py:data:: FrameMapper .. py:data:: LEGACY_PATTERN :value: 'step_*.json' .. py:function:: _as_record(value: Any) -> dict[str, Any] Return *value* as a mapping — non-mapping results are wrapped. .. py:function:: _read_legacy(path: pathlib.Path) -> Optional[dict[str, Any]] Load one per-step document as a record, or ``None`` if not readable yet. .. py:function:: _read_line(line: dict[str, Any], run_dir: pathlib.Path) -> Optional[dict[str, Any]] Return the record a ``records.jsonl`` line stands for, or ``None``. Blob-backed results (arrays, encoded images) are loaded and wrapped in ``{"value": …}`` so every record is a mapping; a blob that cannot be read drops its line — the commit line is written after the blob, so this is a damaged run, not a race. .. py:class:: PushSource(name: str = 'push', maxlen: Optional[int] = 10000) Thread-safe buffer of frames pushed in by the producer. Producers call :meth:`emit` from any thread; the consumer drains the buffer with :meth:`poll` on its own thread. When *maxlen* is set the buffer drops the oldest pending frames instead of growing without bound — a slow consumer never turns into a memory leak. Initialize the buffer. :param name: Identifier stamped onto every frame's ``source``. :param maxlen: Maximum number of pending frames; ``None`` for unbounded. .. py:attribute:: name :value: 'push' .. py:attribute:: _queue :type: collections.deque[plesty.lib.monitor.base_monitor.Frame] .. py:attribute:: _lock .. py:attribute:: _seq :value: 0 .. py:method:: emit(values: dict[str, Any], **context: Any) -> plesty.lib.monitor.base_monitor.Frame Buffer one update. :param values: Channel name → value. :param \*\*context: Extra context carried on the frame. :returns: The buffered :class:`~plesty.lib.monitor.base_monitor.Frame`. .. py:method:: poll() -> list[plesty.lib.monitor.base_monitor.Frame] Return and clear the buffered frames. :returns: The pending frames in emission order. .. py:method:: close() -> None Drop everything still buffered. .. py:class:: TelemetrySource(devices: Sequence[Any], *, kinds: Optional[Sequence[str]] = None, names: Optional[Sequence[str]] = None, name: str = 'telemetry', maxlen: Optional[int] = 10000) Bases: :py:obj:`PushSource` Frames built from the telemetry events of one or more devices. Registers a hook on every device (see :meth:`~plesty.lib.device.telemetry.TelemetrySystem.register_telemetry_hook`) and buffers each event as a frame with the channels ``device``, ``kind``, ``name``, ``value``, and ``unit``, plus the event's own context. Hooks run on the emitting device's thread, so buffering is all that happens there. Subscribe to the devices' telemetry. :param devices: Devices exposing ``register_telemetry_hook``. :param kinds: Event kinds to keep (see :data:`~plesty.lib.device.telemetry.TELEMETRY_KINDS`); ``None`` keeps every kind. :param names: Parameter/metric names to keep; ``None`` keeps all. :param name: Identifier stamped onto every frame's ``source``. :param maxlen: Maximum number of pending frames. .. py:attribute:: _devices .. py:attribute:: _kinds :value: None .. py:attribute:: _names :value: None .. py:method:: _on_event(event: Any) -> None Buffer one telemetry event, honouring the kind/name filters. .. py:method:: close() -> None Unregister the hooks and drop the buffered frames. .. py:class:: RunSource(run_dir: str | pathlib.Path, *, mapper: Optional[FrameMapper] = None, name: Optional[str] = None) Frames tailed from the record file of an experiment run. The experiment appends one line per completed step to ``records.jsonl`` (:func:`~plesty.lib.data.io.append_record`), and that line is the commit record: it is complete only once the step is fully persisted. Tailing the file by offset is therefore both safe and cheap — the source reads, never writes, costs one small read per poll however long the run, and can attach to a run that started hours ago and still see every row, including the ones it missed. Records are emitted in write order; a line still being written (no newline yet) waits for the next poll. Runs written by plesty-lib ≤ 0.3.4 (one ``data/step_*.json`` per step) are read the old way. .. code-block:: python source = RunSource("runs/qd_fss_20260804-214826", mapper=lambda rec: {"angle_deg": rec["hwp_deg"]}) monitor.bind(source) monitor.pump() # → submits every row persisted so far Follow a run directory. :param run_dir: The run's checkpoint directory (the one holding ``plan.json``, ``journal.jsonl``, and ``records.jsonl``). :param mapper: Turns one record into the channel mapping a monitor declares; returning ``None`` drops the record (useful for the bookkeeping steps a plan starts with). The default passes the record through unchanged. :param name: Identifier stamped onto every frame's ``source``; defaults to the run directory name. .. py:attribute:: run_dir .. py:attribute:: mapper :value: None .. py:attribute:: name :value: '' .. py:attribute:: _offset :value: 0 .. py:attribute:: _seen :type: set[str] .. py:attribute:: _seq :value: 0 .. py:property:: legacy :type: bool Whether the run is in the per-step-document layout (no record file). .. py:method:: _new_records() -> list[tuple[str, dict[str, Any]]] Return the (label, record) pairs committed since the last call. .. py:method:: poll() -> list[plesty.lib.monitor.base_monitor.Frame] Return frames for the records committed since last time. :returns: The new frames in write order; empty while the run directory does not exist yet, so a viewer may be started before the experiment. .. py:property:: unread :type: bool Whether committed records exist that no poll has returned yet. .. py:method:: rewind() -> None Forget which records were seen, so the next poll replays the run. .. py:property:: finished :type: bool Whether the run reached a terminal state. A completed, aborted, or canceled run writes no further records, so a viewer can stop polling (and stop recording) on its own. .. py:method:: status() -> dict[str, Any] Return the run status derived from the journal. :returns: The summary of :meth:`plesty.lib.experiment.journal.Journal.status` — state, completed and failed step counts, last event. .. py:method:: close() -> None Release nothing — the source holds no handle; kept for symmetry. .. py:class:: ReplaySource(run_dir: str | pathlib.Path, *, mapper: Optional[FrameMapper] = None, name: Optional[str] = None, rows_per_poll: int = 1, loop: bool = False) Bases: :py:obj:`RunSource` A finished run handed out a few rows at a time, as if it were live. Watching a stored run play back is how a viewer is demonstrated, how a layout is judged before the beam time that needs it, and how a run that went wrong is re-read at the pace it happened rather than as a finished picture. It is the same source as :class:`RunSource` — the records are the ones the experiment committed — only rationed. The wall-clock speed is the consumer's refresh interval times :attr:`rows_per_poll`: a shell ticking every 500 ms with the default of one row replays two rows per second. .. code-block:: python source = ReplaySource(run_dir, mapper=to_channels, rows_per_poll=2) monitor.bind(source) # the run rebuilds itself in the window Replay a run directory. :param run_dir: The run's checkpoint directory. :param mapper: Turns one record into a view's channels; see :class:`RunSource`. :param name: Identifier stamped onto every frame's ``source``. :param rows_per_poll: Records handed out per :meth:`poll`; with the consumer's refresh interval, this is the replay speed. :param loop: Start over once the run is exhausted, instead of stopping at the last row. :raises ValueError: If *rows_per_poll* is not positive — a replay that hands out nothing would look like a stalled run. .. py:attribute:: rows_per_poll :value: 1 .. py:attribute:: loop :value: False .. py:attribute:: _pending :type: list[plesty.lib.monitor.base_monitor.Frame] :value: [] .. py:method:: poll() -> list[plesty.lib.monitor.base_monitor.Frame] Return the next few records of the run. :returns: At most :attr:`rows_per_poll` frames; empty once the run is exhausted, unless :attr:`loop` is set. .. py:property:: exhausted :type: bool Whether every record of a finished run has been handed out. False while a record is still buffered or still on disk unread, and false for a run that is still measuring — there, the next record is simply not written yet. A looping replay is never exhausted: the next poll starts it over. .. py:method:: rewind() -> None Start the replay over from the first record. .. py:function:: records(run_dir: str | pathlib.Path) -> Iterable[dict[str, Any]] Iterate the result records of a finished run, oldest first. The offline counterpart of :class:`RunSource`, for replaying a run into a monitor or feeding an analyzer. :param run_dir: The run's checkpoint directory. :Yields: One record per readable committed result, in write order.