plesty.lib.monitor.sources

Data sources feeding monitors: run directories, telemetry, manual pushes.

Every source answers one question — what appeared since I last asked? — through a non-blocking poll(). That single method is the whole subscription contract (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:

  • RunSource — tails a running experiment’s records.jsonl (<run_root>/<run_id>/) 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.

  • TelemetrySource — registers a TelemetryEvent hook on one or more devices and turns each event into a frame.

  • PushSource — a thread-safe hand-fed buffer for tests, notebooks, and code that already has the values.

  • ReplaySource — a finished run rationed a few rows per poll, so a stored measurement plays back at the pace it happened.

Attributes

FrameMapper

LEGACY_PATTERN

Classes

PushSource

Thread-safe buffer of frames pushed in by the producer.

TelemetrySource

Frames built from the telemetry events of one or more devices.

RunSource

Frames tailed from the record file of an experiment run.

ReplaySource

A finished run handed out a few rows at a time, as if it were live.

Functions

_as_record(→ dict[str, Any])

Return value as a mapping — non-mapping results are wrapped.

_read_legacy(→ Optional[dict[str, Any]])

Load one per-step document as a record, or None if not readable yet.

_read_line(→ Optional[dict[str, Any]])

Return the record a records.jsonl line stands for, or None.

records(→ Iterable[dict[str, Any]])

Iterate the result records of a finished run, oldest first.

Module Contents

plesty.lib.monitor.sources.FrameMapper
plesty.lib.monitor.sources.LEGACY_PATTERN = 'step_*.json'
plesty.lib.monitor.sources._as_record(value: Any) → dict[str, Any]

Return value as a mapping — non-mapping results are wrapped.

Parameters:

value (Any)

Return type:

dict[str, Any]

plesty.lib.monitor.sources._read_legacy(path: pathlib.Path) → dict[str, Any] | None

Load one per-step document as a record, or None if not readable yet.

Parameters:

path (pathlib.Path)

Return type:

Optional[dict[str, Any]]

plesty.lib.monitor.sources._read_line(line: dict[str, Any], run_dir: pathlib.Path) → dict[str, Any] | None

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.

Parameters:

  • line (dict[str, Any])

  • run_dir (pathlib.Path)

Return type:

Optional[dict[str, Any]]

class plesty.lib.monitor.sources.PushSource(name: str = 'push', maxlen: int | None = 10000)

Thread-safe buffer of frames pushed in by the producer.

Producers call emit() from any thread; the consumer drains the buffer with 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.

Parameters:

  • name (str) – Identifier stamped onto every frame’s source.

  • maxlen (Optional[int]) – Maximum number of pending frames; None for unbounded.

name = 'push'
_queue: collections.deque[plesty.lib.monitor.base_monitor.Frame]
_lock
_seq = 0
emit(values: dict[str, Any], **context: Any) → plesty.lib.monitor.base_monitor.Frame

Buffer one update.

Parameters:

  • values (dict[str, Any]) – Channel name → value.

  • **context (Any) – Extra context carried on the frame.

Returns:

The buffered Frame.

Return type:

plesty.lib.monitor.base_monitor.Frame

poll() → list[plesty.lib.monitor.base_monitor.Frame]

Return and clear the buffered frames.

Returns:

The pending frames in emission order.

Return type:

list[plesty.lib.monitor.base_monitor.Frame]

close() → None

Drop everything still buffered.

Return type:

None

class plesty.lib.monitor.sources.TelemetrySource(devices: Sequence[Any], *, kinds: Sequence[str] | None = None, names: Sequence[str] | None = None, name: str = 'telemetry', maxlen: int | None = 10000)

Bases: PushSource

Frames built from the telemetry events of one or more devices.

Registers a hook on every device (see 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.

Parameters:

  • devices (Sequence[Any]) – Devices exposing register_telemetry_hook.

  • kinds (Optional[Sequence[str]]) – Event kinds to keep (see TELEMETRY_KINDS); None keeps every kind.

  • names (Optional[Sequence[str]]) – Parameter/metric names to keep; None keeps all.

  • name (str) – Identifier stamped onto every frame’s source.

  • maxlen (Optional[int]) – Maximum number of pending frames.

_devices
_kinds = None
_names = None
_on_event(event: Any) → None

Buffer one telemetry event, honouring the kind/name filters.

Parameters:

event (Any)

Return type:

None

close() → None

Unregister the hooks and drop the buffered frames.

Return type:

None

class plesty.lib.monitor.sources.RunSource(run_dir: str | pathlib.Path, *, mapper: FrameMapper | None = None, name: str | None = None)

Frames tailed from the record file of an experiment run.

The experiment appends one line per completed step to records.jsonl (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.

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.

Parameters:

  • run_dir (str | pathlib.Path) – The run’s checkpoint directory (the one holding plan.json, journal.jsonl, and records.jsonl).

  • mapper (Optional[FrameMapper]) – 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.

  • name (Optional[str]) – Identifier stamped onto every frame’s source; defaults to the run directory name.

run_dir
mapper = None
name = ''
_offset = 0
_seen: set[str]
_seq = 0
property legacy: bool

Whether the run is in the per-step-document layout (no record file).

Return type:

bool

_new_records() → list[tuple[str, dict[str, Any]]]

Return the (label, record) pairs committed since the last call.

Return type:

list[tuple[str, dict[str, Any]]]

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.

Return type:

list[plesty.lib.monitor.base_monitor.Frame]

property unread: bool

Whether committed records exist that no poll has returned yet.

Return type:

bool

rewind() → None

Forget which records were seen, so the next poll replays the run.

Return type:

None

property finished: 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.

Return type:

bool

status() → dict[str, Any]

Return the run status derived from the journal.

Returns:

The summary of plesty.lib.experiment.journal.Journal.status() — state, completed and failed step counts, last event.

Return type:

dict[str, Any]

close() → None

Release nothing — the source holds no handle; kept for symmetry.

Return type:

None

class plesty.lib.monitor.sources.ReplaySource(run_dir: str | pathlib.Path, *, mapper: FrameMapper | None = None, name: str | None = None, rows_per_poll: int = 1, loop: bool = False)

Bases: 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 RunSource — the records are the ones the experiment committed — only rationed.

The wall-clock speed is the consumer’s refresh interval times rows_per_poll: a shell ticking every 500 ms with the default of one row replays two rows per second.

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.

Parameters:

  • run_dir (str | pathlib.Path) – The run’s checkpoint directory.

  • mapper (Optional[FrameMapper]) – Turns one record into a view’s channels; see RunSource.

  • name (Optional[str]) – Identifier stamped onto every frame’s source.

  • rows_per_poll (int) – Records handed out per poll(); with the consumer’s refresh interval, this is the replay speed.

  • loop (bool) – 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.

rows_per_poll = 1
loop = False
_pending: list[plesty.lib.monitor.base_monitor.Frame] = []
poll() → list[plesty.lib.monitor.base_monitor.Frame]

Return the next few records of the run.

Returns:

At most rows_per_poll frames; empty once the run is exhausted, unless loop is set.

Return type:

list[plesty.lib.monitor.base_monitor.Frame]

property exhausted: 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.

Return type:

bool

rewind() → None

Start the replay over from the first record.

Return type:

None

plesty.lib.monitor.sources.records(run_dir: str | pathlib.Path) → Iterable[dict[str, Any]]

Iterate the result records of a finished run, oldest first.

The offline counterpart of RunSource, for replaying a run into a monitor or feeding an analyzer.

Parameters:

run_dir (str | pathlib.Path) – The run’s checkpoint directory.

Yields:

One record per readable committed result, in write order.

Return type:

Iterable[dict[str, Any]]