plesty.lib.monitor.base_monitor

Monitor ABC: schema-declared live views over a stream of measured data.

A Monitor is the platform’s read-only live view (issue plesty-lib#24). Where an Analyzer is a pull-style transform of a complete dataset, a monitor is a push-style projection of data as it arrives — a running experiment’s step records, a device’s telemetry events, or anything else a DataSource delivers. Watching never disturbs measuring: a monitor holds no device handle, issues no command, and writes nothing back.

The contract splits in three, so no layer depends on the one above it:

Subscription — a DataSource produces Frame objects and Monitor.pump() drains it. Sources never block and own no thread, so the clock belongs to the caller: a GUI timer, a test loop, a notebook cell.
Projection — the subclass declares the channels it consumes in input_schema and implements Monitor.update(), a pure function of (frame, monitor state) returning drawable TraceData. This layer imports no GUI toolkit.
Rendering — a Renderer draws the traces declared by Monitor.traces(). plesty.lib.ui ships the PySide6/pyqtgraph renderer and the dockable shell; NullRenderer and plain Monitor.snapshot() reads cover headless use and tests.

Minimal example — a monitor that plots the newest spectrum:

from plesty.lib.monitor import Frame, Monitor, TraceData, TraceSpec

class Spectrum(Monitor):
    title = "Spectrum"
    input_schema = {
        "wavelength": {"dtype": "array_float", "unit": "nm",
                       "description": "Wavelength axis."},
        "counts": {"dtype": "array_float", "unit": "counts",
                   "description": "Detector counts."},
    }

    def traces(self):
        return [TraceSpec("spectrum", "line", x_label="Wavelength (nm)",
                          y_label="Counts", color="cyan")]

    def update(self, frame):
        return {"spectrum": TraceData(x=frame["wavelength"], y=frame["counts"])}

monitor = Spectrum()
monitor.submit(Frame({"wavelength": [780.0, 780.1], "counts": [12.0, 90.0]}))
monitor.snapshot()["spectrum"].y  # → the counts last drawn

Attributes

`TRACE_KINDS`
`_CHANNEL_ENTRY_KEYS`

Exceptions

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

Classes

`Frame`	One immutable update delivered to a monitor.
`DataSource`	What `Monitor.pump()` needs from a stream.
`TraceSpec`	Static declaration of one drawable element of a monitor's view.
`TraceData`	Drawable payload produced for one trace by `Monitor.update()`.
`Renderer`	Draw target of a monitor — the seam between projection and toolkit.
`NullRenderer`	Renderer that records draw calls instead of drawing them.
`Monitor`	Base class for PLESTY monitors: schema-declared live views.

Functions

`utc_timestamp`(→ str)	Return the current UTC time as an ISO-8601 string.
`_describe_value`(→ Any)	Coerce a view parameter into a JSON-safe representation.

Module Contents

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

plesty.lib.monitor.base_monitor._CHANNEL_ENTRY_KEYS

exception plesty.lib.monitor.base_monitor.MonitorSchemaError

Bases: ValueError

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

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

exception plesty.lib.monitor.base_monitor.FrameValidationError

Bases: ValueError

Raised when a submitted frame violates the declared input_schema.

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

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

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

Return type:: str

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

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

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

plesty.lib.monitor.base_monitor._describe_value(value: Any) → Any

Coerce a view parameter into a JSON-safe representation.

Parameters:: value (Any)
Return type:: Any