plesty.lib.monitor ================== .. py:module:: plesty.lib.monitor .. autoapi-nested-parse:: Monitor framework: schema-declared live views over streams of measured data. The contract (:class:`Monitor`) and the streams that feed it (:mod:`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 :mod:`plesty.lib.ui`. Submodules ---------- .. toctree:: :maxdepth: 1 /reference/plesty/lib/monitor/base_monitor/index /reference/plesty/lib/monitor/sources/index /reference/plesty/lib/monitor/viz/index Attributes ---------- .. autoapisummary:: plesty.lib.monitor.TRACE_KINDS plesty.lib.monitor.FrameMapper Exceptions ---------- .. autoapisummary:: plesty.lib.monitor.FrameValidationError plesty.lib.monitor.MonitorSchemaError Classes ------- .. autoapisummary:: plesty.lib.monitor.DataSource plesty.lib.monitor.Frame plesty.lib.monitor.Monitor plesty.lib.monitor.NullRenderer plesty.lib.monitor.Renderer plesty.lib.monitor.TraceData plesty.lib.monitor.TraceSpec plesty.lib.monitor.PushSource plesty.lib.monitor.ReplaySource plesty.lib.monitor.RunSource plesty.lib.monitor.TelemetrySource plesty.lib.monitor.Viz Functions --------- .. autoapisummary:: plesty.lib.monitor.utc_timestamp plesty.lib.monitor.records Package Contents ---------------- .. py:data:: TRACE_KINDS :type: tuple[str, Ellipsis] :value: ('line', 'scatter', 'image', 'scalar', 'text') .. py:class:: DataSource Bases: :py:obj:`Protocol` What :meth:`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 :mod:`plesty.lib.monitor.sources`. .. py:method:: poll() -> Iterable[Frame] Return the frames that appeared since the previous call. .. py:class:: Frame One immutable update delivered to a monitor. :ivar values: Channel name → value, validated against ``input_schema``. :ivar seq: Monotonic index within the stream, assigned by the source. :ivar timestamp: UTC ISO-8601 time the update was produced; filled when empty. :ivar source: Identifier of the producing source (run id, device id, …). :ivar context: Extra JSON-serializable context (step id, run status, file paths) carried alongside the channels but never validated. .. py:attribute:: values :type: dict[str, Any] .. py:attribute:: seq :type: int :value: 0 .. py:attribute:: timestamp :type: str :value: '' .. py:attribute:: source :type: str :value: '' .. py:attribute:: context :type: dict[str, Any] .. py:method:: __post_init__() -> None Stamp the creation time when the producer left it empty. .. py:method:: __getitem__(channel: str) -> Any Return the value of *channel*. :param channel: Declared channel name. :returns: The channel value. :raises KeyError: If the frame carries no such channel. .. py:method:: get(channel: str, default: Any = None) -> Any Return the value of *channel*, or *default* when it is absent. :param channel: Declared channel name. :param default: Returned when the channel was not delivered. :returns: The channel value, or *default*. .. py:exception:: FrameValidationError Bases: :py:obj:`ValueError` Raised when a submitted frame violates the declared ``input_schema``. Initialize self. See help(type(self)) for accurate signature. .. py:class:: Monitor(source: Optional[DataSource] = None, *, history: Optional[int] = 1, name: Optional[str] = None, **params: Any) Bases: :py:obj:`abc.ABC` Base class for PLESTY monitors: schema-declared live views. Subclasses declare the channels they consume in :attr:`input_schema` and implement exactly two methods — :meth:`traces` (what is drawn, declared once) and :meth:`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 (:func:`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. :param source: Stream to subscribe to; may be attached later with :meth:`bind`, or omitted when frames are submitted by hand. :param history: Number of past frames kept in :attr:`frames` (``None`` keeps all, ``1`` keeps only the newest). Accumulating monitors keep their own series and leave this at 1. :param name: Instance name, used by the shell for layout persistence; defaults to the class name. :param \*\*params: View parameters (limits, decimation, calibration), introspectable via :attr:`params`. :raises MonitorSchemaError: If ``input_schema`` is invalid or :meth:`traces` declares nothing. .. py:attribute:: title :type: ClassVar[str] :value: '' Human-readable view title; falls back to the class name. .. py:attribute:: input_schema :type: ClassVar[dict[str, dict[str, Any]]] name → schema entry. :type: Channels this monitor consumes .. py:attribute:: params .. py:attribute:: name :value: 'Monitor' .. py:attribute:: source :value: None .. py:attribute:: frames :type: collections.deque[Frame] .. py:attribute:: count :value: 0 .. py:attribute:: _renderers :type: list[Renderer] :value: [] .. py:attribute:: _last :type: dict[str, TraceData] .. py:attribute:: _specs .. py:method:: traces() -> list[TraceSpec] :abstractmethod: 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. .. py:method:: update(frame: Frame) -> dict[str, TraceData] :abstractmethod: 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. :param frame: The validated update. :returns: Trace key → payload; the keys must be declared traces. .. py:method:: submit(frame: Frame) -> dict[str, TraceData] Validate a frame, project it, and push the result to the renderers. :param frame: The update to process. :returns: The projected payloads returned by :meth:`update`. :raises FrameValidationError: If the frame violates ``input_schema``, or :meth:`update` returns an undeclared trace key. .. py:method:: 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. .. py:method:: bind(source: DataSource) -> None Subscribe this monitor to *source*. :param source: The stream :meth:`pump` will drain. .. py:method:: attach(renderer: Renderer) -> Renderer Attach a renderer, build its items, and replay the last drawn data. :param renderer: The draw target. :returns: The attached renderer, for convenient chaining. .. py:method:: detach(renderer: Renderer) -> None Remove a previously attached renderer. :param renderer: The draw target to remove; unknown renderers are ignored. .. py:method:: 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()``. .. py:method:: snapshot() -> dict[str, TraceData] Return the most recent payload of every trace drawn so far. :returns: Trace key → payload; empty before the first frame. .. py:method:: trace_spec(key: str) -> TraceSpec Return the declaration of one trace. :param key: Trace key. :returns: The :class:`TraceSpec`. :raises KeyError: If the monitor declares no such trace. .. py:property:: trace_specs :type: list[TraceSpec] The declared traces, in draw order. .. py:property:: display_title :type: str :attr:`title`, or the class name. :type: Title a shell shows .. py:method:: 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. .. py:method:: 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. .. py:method:: _validate_schema() -> None Check ``input_schema`` entry by entry. :raises MonitorSchemaError: On an unknown key, a missing or unresolvable ``dtype``, or a malformed ``shape``. .. py:method:: _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. :param values: 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. .. py:method:: _conform_channel(name: str, value: Any, entry: dict[str, Any]) -> Any Conform one channel value to its declared dtype and shape. .. py:method:: _conform_array(name: str, value: Any, entry: dict[str, Any]) -> plesty.lib.data.array.PlestyArray :staticmethod: Coerce an array channel and check it against the declared shape. .. py:method:: _conform_scalar(name: str, value: Any, expected: type) -> Any :staticmethod: Check a scalar channel, accepting ints and NumPy scalars for floats. .. py:method:: _check_traces(data: dict[str, TraceData]) -> None Check that :meth:`update` returned declared trace keys only. .. py:exception:: MonitorSchemaError Bases: :py:obj:`ValueError` Raised when a monitor's channel or trace declaration is invalid. Initialize self. See help(type(self)) for accurate signature. .. py:class:: NullRenderer Bases: :py:obj:`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. .. py:attribute:: drawn :type: list[dict[str, TraceData]] :value: [] .. py:method:: draw(monitor: Monitor, data: dict[str, TraceData]) -> None Append *data* to :attr:`drawn`. :param monitor: The monitor that produced the update; unused. :param data: Trace key → payload. .. py:method:: clear(monitor: Monitor) -> None Discard the recorded draw calls. :param monitor: The monitor whose view is being reset; unused. .. py:class:: 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 (:mod:`plesty.lib.ui`) subclass it. .. py:method:: setup(monitor: Monitor) -> None Build the drawable items declared by ``monitor.traces()``. :param monitor: The monitor this renderer was attached to. .. py:method:: draw(monitor: Monitor, data: dict[str, TraceData]) -> None Draw one update. :param monitor: The monitor that produced the update. :param data: Trace key → payload, as returned by :meth:`Monitor.update`. .. py:method:: clear(monitor: Monitor) -> None Drop everything drawn so far. :param monitor: The monitor whose view is being reset. .. py:class:: TraceData Drawable payload produced for one trace by :meth:`Monitor.update`. :ivar x: Sample positions of a ``line``/``scatter`` trace. :ivar y: Sample values of a ``line``/``scatter`` trace. :ivar image: 2-D array of an ``image`` trace. :ivar value: Number shown by a ``scalar`` trace. :ivar text: Text of a ``text`` trace, or an annotation for any other kind. .. py:attribute:: x :type: Optional[Sequence[float]] :value: None .. py:attribute:: y :type: Optional[Sequence[float]] :value: None .. py:attribute:: image :type: Any :value: None .. py:attribute:: value :type: Optional[float] :value: None .. py:attribute:: text :type: Optional[str] :value: None .. py:class:: 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 :class:`TraceData` changes per frame. :ivar key: Identifier used as the key of the :meth:`Monitor.update` result. :ivar kind: One of :data:`TRACE_KINDS`. :ivar label: Legend label; falls back to *key* when empty. :ivar x_label: Axis label including the unit, e.g. ``"Wavelength (nm)"``. :ivar y_label: Axis label including the unit, e.g. ``"Power (uW)"``. :ivar color: PLESTY palette token (``"cyan"``, ``"violet"``, ``"mint"``, ``"amber"``, ``"rose"``, ``"blue"``). The renderer resolves it against the theme, so monitors never hard-code hex values. :ivar width: Line width, or marker size for ``scatter``. :ivar marker: Draw point markers on a ``line`` trace. .. py:attribute:: key :type: str .. py:attribute:: kind :type: str :value: 'line' .. py:attribute:: label :type: str :value: '' .. py:attribute:: x_label :type: str :value: '' .. py:attribute:: y_label :type: str :value: '' .. py:attribute:: color :type: str :value: 'cyan' .. py:attribute:: width :type: float :value: 1.5 .. py:attribute:: marker :type: bool :value: False .. py:method:: __post_init__() -> None Validate the trace kind. :raises MonitorSchemaError: If :attr:`kind` is not a known trace kind. .. py:function:: utc_timestamp() -> str Return the current UTC time as an ISO-8601 string. .. py:data:: FrameMapper .. 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:: 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: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:: 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: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. .. py:class:: Viz(title: str, *, experiment: Union[str, Sequence[str], None] = None, run_root: Optional[str | pathlib.Path] = None) The ``monitor`` / ``render`` command line over one experiment's panels. Describe the experiment whose runs are watched. :param title: Human title of the experiment, shown in the window title and the command-line description. :param experiment: 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. :param run_root: 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. .. py:attribute:: title .. py:attribute:: experiment :value: None .. py:attribute:: run_root :value: None .. py:attribute:: _panels :type: Optional[PanelsFunc] :value: None .. py:method:: panels(func: PanelsFunc) -> PanelsFunc Register the function that turns a :class:`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. .. py:method:: build_parser() -> argparse.ArgumentParser Return the argument parser: shared options plus the two subcommands. .. py:method:: resolve_run(args: argparse.Namespace) -> plesty.lib.experiment.runs.Run Return the :class:`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. .. py:method:: main(argv: Optional[Sequence[str]] = None) -> int Run the command line and return the exit code. :param argv: Argument list; ``None`` uses ``sys.argv``. :raises RuntimeError: If no panels function was registered. :raises SystemExit: If no run exists, or the panels function yields nothing.