plesty.lib.monitor

Monitor framework: schema-declared live views over streams of measured data.

The contract (Monitor) and the streams that feed it (plesty.lib.monitor.sources) are toolkit-free — importing this package pulls in no GUI. Concrete views live in their own modules (for spectroscopy: plesty-common-monitors); the dockable window that shows them is plesty.lib.ui.

Submodules

Attributes

`TRACE_KINDS`
`FrameMapper`

Exceptions

`FrameValidationError`	Raised when a submitted frame violates the declared `input_schema`.
`MonitorSchemaError`	Raised when a monitor's channel or trace declaration is invalid.

Classes

`DataSource`	What `Monitor.pump()` needs from a stream.
`Frame`	One immutable update delivered to a monitor.
`Monitor`	Base class for PLESTY monitors: schema-declared live views.
`NullRenderer`	Renderer that records draw calls instead of drawing them.
`Renderer`	Draw target of a monitor — the seam between projection and toolkit.
`TraceData`	Drawable payload produced for one trace by `Monitor.update()`.
`TraceSpec`	Static declaration of one drawable element of a monitor's view.
`PushSource`	Thread-safe buffer of frames pushed in by the producer.
`ReplaySource`	A finished run handed out a few rows at a time, as if it were live.
`RunSource`	Frames tailed from the record file of an experiment run.
`TelemetrySource`	Frames built from the telemetry events of one or more devices.
`Viz`	The `monitor` / `render` command line over one experiment's panels.

Functions

`utc_timestamp`(→ str)	Return the current UTC time as an ISO-8601 string.
`records`(→ Iterable[dict[str, Any]])	Iterate the result records of a finished run, oldest first.

Package Contents

plesty.lib.monitor.TRACE_KINDS: tuple[str, Ellipsis] = ('line', 'scatter', 'image', 'scalar', 'text')

class plesty.lib.monitor.DataSource

Bases: Protocol

What Monitor.pump() needs from a stream.

Declared here as a protocol so the ABC module stays free of I/O; the concrete implementations live in plesty.lib.monitor.sources.

poll() → Iterable[Frame]

Return the frames that appeared since the previous call.

Return type:: Iterable[Frame]

class plesty.lib.monitor.Frame

One immutable update delivered to a monitor.

Variables:

values – Channel name → value, validated against input_schema.
seq – Monotonic index within the stream, assigned by the source.
timestamp – UTC ISO-8601 time the update was produced; filled when empty.
source – Identifier of the producing source (run id, device id, …).
context – Extra JSON-serializable context (step id, run status, file paths) carried alongside the channels but never validated.

values: dict[str, Any]

seq: int = 0

timestamp: str = ''

source: str = ''

context: dict[str, Any]

__post_init__() → None

Stamp the creation time when the producer left it empty.

Return type:: None

__getitem__(channel: str) → Any

Return the value of channel.

Parameters:: channel (str) – Declared channel name.
Returns:: The channel value.
Raises:: KeyError – If the frame carries no such channel.
Return type:: Any

get(channel: str, default: Any = None) → Any

Return the value of channel, or default when it is absent.

Parameters:

channel (str) – Declared channel name.
default (Any) – Returned when the channel was not delivered.

Returns:

The channel value, or default.

Return type:

Any

exception plesty.lib.monitor.FrameValidationError

Bases: ValueError

Raised when a submitted frame violates the declared input_schema.

Initialize self. See help(type(self)) for accurate signature.

class plesty.lib.monitor.Monitor(source: DataSource | None = None, *, history: int | None = 1, name: str | None = None, **params: Any)

Bases: abc.ABC

Base class for PLESTY monitors: schema-declared live views.

Subclasses declare the channels they consume in input_schema and implement exactly two methods — traces() (what is drawn, declared once) and update() (what to draw for one frame). Frame validation, history, renderer fan-out, and source draining are base machinery.

Channel entries use the dtype vocabulary of device and analyzer schemas (plesty.lib.data.types.resolve_dtype()) and accept the keys dtype (required), unit, description, shape (arrays only; None marks a free dimension), and required (default True).

Initialize the monitor and validate its declarations.

Parameters:

source (Optional[DataSource]) – Stream to subscribe to; may be attached later with bind(), or omitted when frames are submitted by hand.
history (Optional[int]) – Number of past frames kept in frames (None keeps all, 1 keeps only the newest). Accumulating monitors keep their own series and leave this at 1.
name (Optional[str]) – Instance name, used by the shell for layout persistence; defaults to the class name.
**params (Any) – View parameters (limits, decimation, calibration), introspectable via params.

Raises:

MonitorSchemaError – If input_schema is invalid or traces() declares nothing.

title: ClassVar[str] = '': Human-readable view title; falls back to the class name.

input_schema: ClassVar[dict[str, dict[str, Any]]]

name → schema entry.

Type:: Channels this monitor consumes

params

name = 'Monitor'

source = None

frames: collections.deque[Frame]

count = 0

_renderers: list[Renderer] = []

_last: dict[str, TraceData]

_specs

abstractmethod traces() → list[TraceSpec]

Declare the drawable elements of this view.

Called once during construction; the result must be stable for the lifetime of the instance, because renderers build their plot items from it when they attach.

Returns:: The trace specifications, in draw order.
Return type:: list[TraceSpec]

abstractmethod update(frame: Frame) → dict[str, TraceData]

Project one validated frame onto the declared traces.

Implementations are pure with respect to the outside world: they read the frame and their own accumulated state and return what to draw. They must not touch devices, files, or GUI objects. A trace left out of the result keeps whatever it was showing.

Parameters:: frame (Frame) – The validated update.
Returns:: Trace key → payload; the keys must be declared traces.
Return type:: dict[str, TraceData]

submit(frame: Frame) → dict[str, TraceData]

Validate a frame, project it, and push the result to the renderers.

Parameters:: frame (Frame) – The update to process.
Returns:: The projected payloads returned by update().
Raises:: FrameValidationError – If the frame violates input_schema, or update() returns an undeclared trace key.
Return type:: dict[str, TraceData]

pump() → int

Drain the bound source and submit every pending frame.

Runs on the caller’s clock — a GUI timer, a test loop — so the monitor owns no thread of its own.

Returns:: The number of frames submitted.
Return type:: int

bind(source: DataSource) → None

Subscribe this monitor to source.

Parameters:: source (DataSource) – The stream pump() will drain.
Return type:: None

attach(renderer: Renderer) → Renderer

Attach a renderer, build its items, and replay the last drawn data.

Parameters:: renderer (Renderer) – The draw target.
Returns:: The attached renderer, for convenient chaining.
Return type:: Renderer

detach(renderer: Renderer) → None

Remove a previously attached renderer.

Parameters:: renderer (Renderer) – The draw target to remove; unknown renderers are ignored.
Return type:: None

reset() → None

Clear history, projected state, and every attached renderer.

Subclasses that accumulate their own series override this, drop that state, and call super().reset().

Return type:: None

snapshot() → dict[str, TraceData]

Return the most recent payload of every trace drawn so far.

Returns:: Trace key → payload; empty before the first frame.
Return type:: dict[str, TraceData]

trace_spec(key: str) → TraceSpec

Return the declaration of one trace.

Parameters:: key (str) – Trace key.
Returns:: The TraceSpec.
Raises:: KeyError – If the monitor declares no such trace.
Return type:: TraceSpec

property trace_specs: list[TraceSpec]

The declared traces, in draw order.

Return type:: list[TraceSpec]

property display_title: str

title, or the class name.

Type:: Title a shell shows
Return type:: str

status() → str

Return a one-line status shown beside the title in a shell.

The default reports how many frames were consumed; subclasses override it with something the operator cares about (the newest angle, the peak wavelength, the run id).

Returns:: The status line.
Return type:: str

describe() → dict[str, Any]

Return a JSON-serializable description of this view.

Shells use it for layout persistence, and reporting tooling to record what was watched during a run.

Returns:: Monitor identity, parameters, channels, and declared traces.
Return type:: dict[str, Any]

_validate_schema() → None

Check input_schema entry by entry.

Raises:: MonitorSchemaError – On an unknown key, a missing or unresolvable dtype, or a malformed shape.
Return type:: None

_conform_values(values: dict[str, Any]) → dict[str, Any]

Check frame channels against input_schema, coercing array-likes.

Unlike analyzer inputs, live frames arrive from JSON step records and telemetry events, so array channels are coerced from plain lists here instead of being rejected.

Parameters:: values (dict[str, Any]) – The raw channel mapping of the frame.
Returns:: The conformed mapping.
Raises:: FrameValidationError – On an unknown or missing channel, a wrong dtype, or a shape contradicting the declaration.
Return type:: dict[str, Any]

_conform_channel(name: str, value: Any, entry: dict[str, Any]) → Any

Conform one channel value to its declared dtype and shape.

Parameters:

name (str)
value (Any)
entry (dict[str, Any])

Return type:

Any

static _conform_array(name: str, value: Any, entry: dict[str, Any]) → plesty.lib.data.array.PlestyArray

Coerce an array channel and check it against the declared shape.

Parameters:

name (str)
value (Any)
entry (dict[str, Any])

Return type:

plesty.lib.data.array.PlestyArray

static _conform_scalar(name: str, value: Any, expected: type) → Any

Check a scalar channel, accepting ints and NumPy scalars for floats.

Parameters:

name (str)
value (Any)
expected (type)

Return type:

Any

_check_traces(data: dict[str, TraceData]) → None

Check that update() returned declared trace keys only.

Parameters:: data (dict[str, TraceData])
Return type:: None

exception plesty.lib.monitor.MonitorSchemaError

Bases: ValueError

Raised when a monitor’s channel or trace declaration is invalid.

Initialize self. See help(type(self)) for accurate signature.

class plesty.lib.monitor.NullRenderer

Bases: Renderer

Renderer that records draw calls instead of drawing them.

Used by tests and by headless runs that only need the projected data.

Initialize the empty draw log.

drawn: list[dict[str, TraceData]] = []

draw(monitor: Monitor, data: dict[str, TraceData]) → None

Append data to drawn.

Parameters:

monitor (Monitor) – The monitor that produced the update; unused.
data (dict[str, TraceData]) – Trace key → payload.

Return type:

None

clear(monitor: Monitor) → None

Discard the recorded draw calls.

Parameters:: monitor (Monitor) – The monitor whose view is being reset; unused.
Return type:: None

class plesty.lib.monitor.Renderer

Draw target of a monitor — the seam between projection and toolkit.

The base implementation does nothing, so headless callers and tests use a monitor without any GUI dependency; GUI backends (plesty.lib.ui) subclass it.

setup(monitor: Monitor) → None

Build the drawable items declared by monitor.traces().

Parameters:: monitor (Monitor) – The monitor this renderer was attached to.
Return type:: None

draw(monitor: Monitor, data: dict[str, TraceData]) → None

Draw one update.

Parameters:

monitor (Monitor) – The monitor that produced the update.
data (dict[str, TraceData]) – Trace key → payload, as returned by Monitor.update().

Return type:

None

clear(monitor: Monitor) → None

Drop everything drawn so far.

Parameters:: monitor (Monitor) – The monitor whose view is being reset.
Return type:: None

class plesty.lib.monitor.TraceData

Drawable payload produced for one trace by Monitor.update().

Variables:

x – Sample positions of a line/scatter trace.
y – Sample values of a line/scatter trace.
image – 2-D array of an image trace.
value – Number shown by a scalar trace.
text – Text of a text trace, or an annotation for any other kind.

x: Sequence[float] | None = None

y: Sequence[float] | None = None

image: Any = None

value: float | None = None

text: str | None = None

class plesty.lib.monitor.TraceSpec

Static declaration of one drawable element of a monitor’s view.

Specs are read when a renderer attaches, to build its plot items; only TraceData changes per frame.

Variables:

key – Identifier used as the key of the Monitor.update() result.
kind – One of TRACE_KINDS.
label – Legend label; falls back to key when empty.
x_label – Axis label including the unit, e.g. "Wavelength (nm)".
y_label – Axis label including the unit, e.g. "Power (uW)".
color – PLESTY palette token ("cyan", "violet", "mint", "amber", "rose", "blue"). The renderer resolves it against the theme, so monitors never hard-code hex values.
width – Line width, or marker size for scatter.
marker – Draw point markers on a line trace.

key: str

kind: str = 'line'

label: str = ''

x_label: str = ''

y_label: str = ''

color: str = 'cyan'

width: float = 1.5

marker: bool = False

__post_init__() → None

Validate the trace kind.

Raises:: MonitorSchemaError – If kind is not a known trace kind.
Return type:: None

plesty.lib.monitor.utc_timestamp() → str

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

Return type:: str

plesty.lib.monitor.FrameMapper

class plesty.lib.monitor.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.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

class plesty.lib.monitor.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.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

plesty.lib.monitor.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]]

class plesty.lib.monitor.Viz(title: str, *, experiment: str | Sequence[str] | None = None, run_root: str | pathlib.Path | None = None)

The monitor / render command line over one experiment’s panels.

Describe the experiment whose runs are watched.

Parameters:

title (str) – Human title of the experiment, shown in the window title and the command-line description.
experiment (Union[str, Sequence[str], None]) – The experiment name(s) whose runs to look for when --run is not given — a name or several (an experiment that was renamed still owns its old runs). None takes the newest run of any experiment.
run_root (Optional[str | pathlib.Path]) – Where runs are looked for; None takes PLESTY_DATA_MOUNT (runs land on the share) and falls back to runs. --run-root overrides per call.

title

experiment = None

run_root = None

_panels: PanelsFunc | None = None

panels(func: PanelsFunc) → PanelsFunc

Register the function that turns a Run into panels.

Use as a decorator. The function is called once per invocation and may return or yield any number of panels — none for a run that has nothing to show is an error worth naming, not an empty window.

Parameters:: func (PanelsFunc)
Return type:: PanelsFunc

build_parser() → argparse.ArgumentParser

Return the argument parser: shared options plus the two subcommands.

Return type:: argparse.ArgumentParser

resolve_run(args: argparse.Namespace) → plesty.lib.experiment.runs.Run

Return the Run the arguments name, configured for the subcommand.

Raises:: SystemExit – If no run exists yet — a viewer opened before its experiment is a mistake worth naming, not an empty window.
Parameters:: args (argparse.Namespace)
Return type:: plesty.lib.experiment.runs.Run

main(argv: Sequence[str] | None = None) → int

Run the command line and return the exit code.

Parameters:

argv (Optional[Sequence[str]]) – Argument list; None uses sys.argv.

Raises:

RuntimeError – If no panels function was registered.
SystemExit – If no run exists, or the panels function yields nothing.

Return type:

int