plesty.lib.monitor.data

Everything a run has produced, in one place, for anything that draws it.

A run writes one line per committed step. RunData reads those lines — once — and keeps them, so a view asks it a question instead of subscribing to a stream of its own.

That is the whole reason this exists. A source hands each record out exactly once, so a design where every view binds its own source needs one source per view, and a spectrum, a spectral map and a fit of the same sweep read the same file three times. Worse, two views that share a source split the rows between them and each draws half the measurement, silently. One store per run removes both: the rows are read once and every plot sees all of them.

data = RunData(Run("runs/pol_pl_20260917-113829"), reads=read_spectrum)
data.poll()                       # ingest whatever has committed
data.latest["hwp_deg"]            # the newest angle
data.column("hwp_deg")            # every angle, as an array
data.stack("counts")              # every spectrum, as one 2-D block
data.where(qd="QD-002").rows      # only that dot's rows

Blobs. A record names the file a device wrote rather than carrying the array — that is what keeps measurements off the network. reads turns one such row into the columns plots want, and its answer is cached per row, so the spectrum a map and a fit both need is loaded once however many plots ask.

Attributes

Reader

Classes

RunData

The rows a run has committed, and the answers a plot needs from them.

Functions

columns(→ list[str])

Every field name the rows carry, in the order first seen.

Module Contents

plesty.lib.monitor.data.Reader
class plesty.lib.monitor.data.RunData(run: plesty.lib.experiment.runs.Run | str, *, reads: Reader | None = None, rows: Sequence[Mapping[str, Any]] | None = None)

The rows a run has committed, and the answers a plot needs from them.

Variables:

  • run – The run being read.

  • reads – Expands a row into extra columns, or None.

Parameters:

Read the run at run.

Parameters:

  • run (plesty.lib.experiment.runs.Run | str) – The run, or a path to one.

  • reads (Optional[Reader]) – Called once per row to add columns the records only point at — a spectrum file becomes wavelength and counts. Its answer is cached, so every plot shares one load.

  • rows (Optional[Sequence[Mapping[str, Any]]]) – Rows to start from instead of reading; used by where() to build a filtered view over rows already read.

run
reads = None
_rows: list[dict[str, Any]]
_source: Any = None
_derived
poll() → int

Ingest whatever has committed since the last call.

Cheap however long the run has been going: the records file is read from the byte offset reached last time, so a poll costs one small read and never re-parses what it already holds.

Returns:

How many rows were added.

Raises:

RuntimeError – On a filtered view, which reads nothing of its own.

Return type:

int

_expand(row: dict[str, Any]) → dict[str, Any]

Return row with whatever reads adds to it.

A reader that fails is not allowed to end the watching: a spectrum still being written, or one file of a thousand that is corrupt, costs that row its extra columns and nothing else.

Parameters:

row (dict[str, Any])

Return type:

dict[str, Any]

property rows: list[dict[str, Any]]

Every row committed so far, oldest first.

Return type:

list[dict[str, Any]]

property latest: dict[str, Any] | None

The newest committed row, whatever kind of step wrote it.

A plan’s steps are not all alike: a sweep commits a spectrum per angle and then an analysis, and the analysis is the newest row. A view of the spectra wants last() instead.

Return type:

Optional[dict[str, Any]]

last(name: str) → dict[str, Any] | None

The newest row that carries name, or None.

What “the current spectrum” means when the newest row is the fit that followed it.

Parameters:

name (str) – The field the row must carry.

Returns:

The row, or None when none has it yet.

Return type:

Optional[dict[str, Any]]

column(name: str, default: Any = np.nan) → numpy.ndarray

One field across every row, as an array.

Parameters:

  • name (str) – The field.

  • default (Any) – Value for a row that does not carry it — nan by default, so a gap stays a gap instead of shifting the axis.

Returns:

One entry per row, in commit order.

Raises:

ValueError – If the field holds arrays rather than single values — naming stack(), which is what such a field is for. Without this the failure is numpy’s, about inhomogeneous shapes, which says nothing about what to do instead.

Return type:

numpy.ndarray

stack(name: str) → numpy.ndarray

An array field across every row, as one 2-D block.

Rows whose entry is missing or a different length are left out rather than padded: a spectral map with a half-row in it is a lie about what was measured.

Parameters:

name (str) – The field.

Returns:

Shape (rows, samples); empty when nothing usable was found.

Return type:

numpy.ndarray

stacked(name: str, along: str = '', default: Any = np.nan) → tuple[numpy.ndarray, numpy.ndarray | None]

The stack of name, and along for exactly the rows it kept.

Asking for the block and its coordinate separately is a bug waiting to happen: stack() leaves out rows that carry no array or the wrong width, and those rows are not necessarily at either end — a sweep that ends in an analysis drops its last row. A coordinate taken from column() would then be one row longer and misaligned, which draws every spectrum against the wrong position. One call decides which rows are usable and answers for both.

Parameters:

  • name (str) – The array field.

  • along (str) – Field holding each kept row’s coordinate; "" for none.

  • default (Any) – Coordinate of a kept row that does not carry along.

Returns:

The block, shape (rows, samples), and the coordinate of each of its rows — or None when along was not asked for.

Return type:

tuple[numpy.ndarray, Optional[numpy.ndarray]]

where(**fields: Any) → RunData

The same data, narrowed to rows whose fields equal fields.

Parameters:

**fields (Any) – Field name to required value.

Returns:

A view over the rows already read — it holds no source and is not polled; poll the data it came from and ask again.

Return type:

RunData

property config: dict[str, Any]

The configuration frozen into the run’s plan.

Return type:

dict[str, Any]

property provenance: dict[str, Any]

What was running when the run started; {} for an older run.

Return type:

dict[str, Any]

status() → dict[str, Any]

The run’s state and step counts, read from its journal.

Return type:

dict[str, Any]

property finished: bool

Whether the run reached a state that produces no more rows.

Return type:

bool

__len__() → int

How many rows have been committed.

Return type:

int

__iter__() → Iterator[dict[str, Any]]

Iterate the rows, oldest first.

Return type:

Iterator[dict[str, Any]]

__bool__() → bool

Whether any row has been committed.

Spelled out because __len__ would otherwise make an empty run falsy in a way a plot guarding on if data: might not intend.

Return type:

bool

__repr__() → str

Return RunData('<run id>', rows=<n>).

Return type:

str

plesty.lib.monitor.data.columns(rows: Iterable[Mapping[str, Any]]) → list[str]

Every field name the rows carry, in the order first seen.

Parameters:

rows (Iterable[Mapping[str, Any]]) – The rows to survey.

Returns:

The field names — what a generic view offers when it has not been told which fields matter.

Return type:

list[str]