plesty.lib.monitor.base_monitor =============================== .. py:module:: plesty.lib.monitor.base_monitor .. autoapi-nested-parse:: Monitor ABC: schema-declared live views over a stream of measured data. A :class:`Monitor` is the platform's **read-only live view** (issue plesty-lib#24). Where an :class:`~plesty.lib.analyzer.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 :class:`~plesty.lib.monitor.sources.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 :class:`~plesty.lib.monitor.sources.DataSource` produces :class:`Frame` objects and :meth:`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 :meth:`Monitor.update`, a pure function of *(frame, monitor state)* returning drawable :class:`TraceData`. This layer imports no GUI toolkit. * **Rendering** — a :class:`Renderer` draws the traces declared by :meth:`Monitor.traces`. :mod:`plesty.lib.ui` ships the PySide6/pyqtgraph renderer and the dockable shell; :class:`NullRenderer` and plain :meth:`Monitor.snapshot` reads cover headless use and tests. Minimal example — a monitor that plots the newest spectrum: .. code-block:: python 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 ---------- .. autoapisummary:: plesty.lib.monitor.base_monitor.TRACE_KINDS plesty.lib.monitor.base_monitor._CHANNEL_ENTRY_KEYS Exceptions ---------- .. autoapisummary:: plesty.lib.monitor.base_monitor.MonitorSchemaError plesty.lib.monitor.base_monitor.FrameValidationError Classes ------- .. autoapisummary:: plesty.lib.monitor.base_monitor.Frame plesty.lib.monitor.base_monitor.DataSource plesty.lib.monitor.base_monitor.TraceSpec plesty.lib.monitor.base_monitor.TraceData plesty.lib.monitor.base_monitor.Renderer plesty.lib.monitor.base_monitor.NullRenderer plesty.lib.monitor.base_monitor.Monitor Functions --------- .. autoapisummary:: plesty.lib.monitor.base_monitor.utc_timestamp plesty.lib.monitor.base_monitor._describe_value Module Contents --------------- .. py:data:: TRACE_KINDS :type: tuple[str, Ellipsis] :value: ('line', 'scatter', 'image', 'scalar', 'text') .. py:data:: _CHANNEL_ENTRY_KEYS .. 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: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:function:: utc_timestamp() -> str Return the current UTC time as an ISO-8601 string. .. 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: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:: 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: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:: 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:: 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:: 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:function:: _describe_value(value: Any) -> Any Coerce a view parameter into a JSON-safe representation.