plesty.lib.data

Data types and utilities for Plesty.

Submodules

Classes

`PlestyArray`	Data definition for a numerical array with meta information.
`Units`	Standardized unit expression parser and composer.
`TableHeader`	Represents the header of a PlestyTable, defining column names and types.
`PlestyTable2D`	A simple 2D table structure that holds a list of 1d PlestyArrays,
`PlestyTable3D`	A simple 3D table structure that holds a list of 2d PlestyTables,
`ExperimentData`	Base contract of a typed piece of experimental data.
`SpectrumData`	A wavelength axis and the intensity measured on it.
`SpectrumFitData`	Line-shape fit results of a spectrum: one row per fitted frame, one column per line.
`ResultDocument`	Typed model of the JSON metadata document written by `save_result()`.

Functions

`resolve_dtype`(...)	Resolve schema dtype strings into Python types.
`resolve_array_item_dtype`(→ Any)	Resolve the NumPy item dtype from an array dtype token like `array_float32`.
`normalize_shape`(→ tuple[Any, Ellipsis] \| None)	Normalize a shape specification into a tuple, or return None.
`istype`(→ bool)	Check if a value is of a specified data type, including basic and iterable types.
`cast_basic_type`(→ Any)	Cast a value to the given basic type, raising ValueError on failure.
`experiment_data_type`(→ type[ExperimentData] \| None)	Return the `ExperimentData` subclass registered under kind.
`registered_kinds`(→ tuple[str, Ellipsis])	Return every registered `kind`, sorted.
`append_record`(→ dict[str, Any])	Append one step result to the run's `records.jsonl`.
`convert_to_hdf5`(→ pathlib.Path)	Pack results written by `save_result()` into one HDF5 archive.
`load_document`(→ ResultDocument)	Load only the metadata document of a persisted result — no blob I/O.
`load_result`(→ Any)	Load a result previously written by `save_result()`.
`read_records`(→ tuple[list[dict[str, Any]], int])	Read the record lines appended to `records.jsonl` since offset.
`record_value`(→ Any)	Return the value a `records.jsonl` line stands for.
`save_result`(→ pathlib.Path)	Persist a measurement result as raw blob + JSON metadata document.

Package Contents

class plesty.lib.data.PlestyArray(shape, dtype=float, buffer=None, offset=0, strides=None, order=None)

Bases: numpy.ndarray

Data definition for a numerical array with meta information. The array inherits from numpy.ndarray, so it can be used as a regular numpy array, but it also has additional attributes for meta information and additional methods for data manipulation.

Variables:

name – Optional name for the data.
range – Optional range for the data values, can be a tuple of (min, max).
options – Optional list of possible values for the data, if applicable.
unit – Optional unit for the data, e.g., “nm”, “s”, “m/s”, etc.
description – Optional description for the data, providing more context and information.

Usage:

from plesty.lib.data import PlestyArray as PArray
data = PArray([1.0, 2.0, 3.0], name="Example Data", unit="nm",
              description="This is an example data array.")
np_arr = np.random.rand(3)
plesty_arr = PArray(np_arr, name="Random Data", unit="s",
                    description="This is a random data array.")

# NumPy-like constructors on PlestyArray
z = PArray.zeros((10, 2), unit="a.u.", description="Zero-filled matrix")
o = PArray.ones(5, name="weights")
x = PArray.arange(0, 1, 0.1, unit="s")
grid = PArray.linspace(400, 700, 5, unit="nm", name="wavelength")

# Unit-aware operations with metadata updates
speed = PArray([10.0, 12.0], name="speed", unit="m/s", description="Speed of Rocket")
time = PArray([2.0, 3.0], name="time", unit="s", description="Time")
distance = speed * time
# distance.unit -> "m"
# distance.name -> "speed(m/s) * time(s)"
# distance.description -> "speed: Speed of Rocket; time: Time"

Create a new PlestyArray instance with optional metadata.

_META_KEYS

__array_priority__ = 1000

name: str | None

range: Tuple[float, float] | Tuple[int, int] | None

options: List[Any] | None

unit: str | None

description: str | None

update_meta(**kwargs) → None

Update meta information for the PlestyArray.

Example:

arr = PlestyArray([1.0, 2.0, 3.0], name="Example Data", unit="nm")
arr.update_meta(description="Updated description", range=(0.0, 5.0))

Return type:: None

_meta_kwargs() → Dict[str, Any]

Return type:: Dict[str, Any]

classmethod _find_meta_source(items) → PlestyArray | None

Return type:: Optional[PlestyArray]

classmethod _unwrap(item: Any) → Any

Return item with every PlestyArray, at any list/tuple depth, as a plain ndarray.

NumPy dispatches __array_function__ on arrays nested in sequence arguments (np.stack([a, b])); passing one on still wrapped would dispatch straight back here (#52).

Parameters:: item (Any)
Return type:: Any

static _unit_text(unit: str | None) → str

Parameters:: unit (Optional[str])
Return type:: str

classmethod _build_binary_meta(left: PlestyArray, right: PlestyArray, op_symbol: str) → Dict[str, Any]

Parameters:

left (PlestyArray)
right (PlestyArray)
op_symbol (str)

Return type:

Dict[str, Any]

classmethod _split_meta_kwargs(kwargs: Dict[str, Any]) → Tuple[Dict[str, Any], Dict[str, Any]]

Parameters:: kwargs (Dict[str, Any])
Return type:: Tuple[Dict[str, Any], Dict[str, Any]]

classmethod _wrap_numpy_result(result: Any, meta_kwargs: Dict[str, Any])

Parameters:

result (Any)
meta_kwargs (Dict[str, Any])

classmethod _call_numpy(np_func, *args, **kwargs)

__array_finalize__(obj): Finalize the PlestyArray after creation, copying metadata from source.

__array_ufunc__(ufunc, method, *inputs, **kwargs): Handle NumPy ufuncs with unit-aware metadata propagation.

__array_function__(func, types, args, kwargs): Dispatch NumPy functions while preserving PlestyArray metadata.

__repr__(): Return a developer-friendly representation of the PlestyArray.

__str__(): Return a human-readable string representation of the PlestyArray.

class plesty.lib.data.Units(dims: Dict[str, int] | None = None, factor: float = 1.0)

Standardized unit expression parser and composer.

The internal representation tracks: - dims: dimension exponents, e.g. {“L”: 1, “T”: -1} for speed - factor: numeric scaling factor to SI base units

Initialize a Units instance with dimension exponents and a scaling factor.

Parameters:

dims (Optional[Dict[str, int]])
factor (float)

SYMBOLS: dict[str, dict[str, Any]]

BASE_SYMBOLS

dims

factor

classmethod scalar() → Units

Return a dimensionless scalar Units instance.

Return type:: Units

classmethod parse(unit_str: str | None) → Units | None

Parse a unit string into a Units instance, returning None if unparseable.

Parameters:: unit_str (Optional[str])
Return type:: Optional[Units]

classmethod standardize(unit_str: str | None) → str | None

Return a normalized whitespace-free unit string, or None for empty input.

Parameters:: unit_str (Optional[str])
Return type:: Optional[str]

compatible(other: Units) → bool

Return True if this unit has the same dimensions as other.

Parameters:: other (Units)
Return type:: bool

conversion_scale_to(target: Units) → float

Return the numeric factor to convert this unit to the target unit.

Parameters:: target (Units)
Return type:: float

multiply(other: Units) → Units

Return the unit resulting from multiplying this unit by other.

Parameters:: other (Units)
Return type:: Units

divide(other: Units) → Units

Return the unit resulting from dividing this unit by other.

Parameters:: other (Units)
Return type:: Units

power(exponent: int) → Units

Return this unit raised to an integer exponent.

Parameters:: exponent (int)
Return type:: Units

to_unit_string() → str | None

Render the unit dimensions as a human-readable string.

Return type:: Optional[str]

plesty.lib.data.resolve_dtype(dtype: Any) → type[Any] | plesty.lib.data.ctype_manager.CtypeParam | None

Resolve schema dtype strings into Python types.

Supports scalar aliases, array types in the form array_<numpy_dtype> (e.g. array_float, array_int32, array_float64), and the kind of any registered ExperimentData type (e.g. spectrum). The built-in aliases take precedence over a kind.

Parameters:: dtype (Any)
Return type:: type[Any] | plesty.lib.data.ctype_manager.CtypeParam | None

plesty.lib.data.resolve_array_item_dtype(dtype: Any) → Any

Resolve the NumPy item dtype from an array dtype token like array_float32.

Parameters:: dtype (Any)
Return type:: Any

plesty.lib.data.normalize_shape(shape: Any) → tuple[Any, Ellipsis] | None

Normalize a shape specification into a tuple, or return None.

Parameters:: shape (Any)
Return type:: tuple[Any, Ellipsis] | None

plesty.lib.data.istype(value: Any, dtype: Any) → bool

Check if a value is of a specified data type, including basic and iterable types.

Parameters:

value (Any)
dtype (Any)

Return type:

bool

plesty.lib.data.cast_basic_type(value: Any, dtype: Any) → Any

Cast a value to the given basic type, raising ValueError on failure.

Parameters:

value (Any)
dtype (Any)

Return type:

Any

class plesty.lib.data.TableHeader

Represents the header of a PlestyTable, defining column names and types.

name: str

dtype: Any = None

unit: str | None = None

description: str | None = None

class plesty.lib.data.PlestyTable2D

A simple 2D table structure that holds a list of 1d PlestyArrays, each representing a row/column of the table.

name: str

data: list[plesty.lib.data.array.PlestyArray]

header: list[TableHeader] | None = None

description: str | None = None

class plesty.lib.data.PlestyTable3D

A simple 3D table structure that holds a list of 2d PlestyTables, each representing a layer of the table.

name: str

shape: tuple[int, int]

data: list[list[plesty.lib.data.array.PlestyArray]]

rheader: list[TableHeader] | None = None

cheader: list[TableHeader] | None = None

description: str | None = None

get_item_at(row: int, col: int) → plesty.lib.data.array.PlestyArray

Get the item at the specified row and column.

Parameters:

row (int)
col (int)

Return type:

plesty.lib.data.array.PlestyArray

class plesty.lib.data.ExperimentData(metadata: dict[str, Any] | None = None, provenance: dict[str, Any] | None = None)

Bases: abc.ABC

Base contract of a typed piece of experimental data.

Class attributes a subclass sets:

kind — mandatory, unique, lower-case identifier; the registry key, the schema dtype token, and what a saved file names.
schema_version — bumped when the arrays or their meaning change; a file written by a newer version is refused, not misread.
readers — suffix → callable reading a foreign file format.
columns — what each array means to a reader, and which are the result; a report tabulates the data from it (plesty.lib.report.data_table()).

Instance attributes every piece of data carries:

metadata — acquisition context (exposure, grating, source file, processing steps, …); JSON-serializable.
provenance — who produced it (device, analyzer, parameters); JSON-serializable.

A subclass that does not set kind itself is not registered, so an intermediate base class can share code without claiming a kind.

Store the metadata and provenance, copied so callers can reuse theirs.

Parameters:

metadata (dict[str, Any] | None) – Acquisition context.
provenance (dict[str, Any] | None) – Origin of the data.

kind: ClassVar[str]: Registry key and schema dtype token, e.g. "spectrum".

schema_version: ClassVar[int] = 1: Version of the array layout this class writes and reads.

readers: ClassVar[dict[str, Callable[[pathlib.Path], Any]]]

lower-case suffix → reader(path).

Type:: Foreign formats this class reads

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

array name → entry, in reading order.

An entry’s keys, all optional:

label — the column heading, inline markup and $math$ allowed; the array name when left out.
unit — used when the array carries none of its own.
description — one line of meaning.
result — False for an array the analysis kept but a reader does not need (a fit quality, a point count); default True.
error — name of the array holding this one’s one-sigma uncertainty, so the two are shown as one value ± error.

Arrays left out are not tabulated.

Type:: What each array means to a reader

metadata: dict[str, Any]: Acquisition context; JSON-serializable.

provenance: dict[str, Any]: Origin of the data (device, analyzer, parameters); JSON-serializable.

classmethod __init_subclass__(**kwargs: Any) → None

Register a subclass that declares its own kind.

Raises:: TypeError – If the kind is malformed or already taken by a different class.
Parameters:: kwargs (Any)
Return type:: None

abstractmethod to_arrays() → dict[str, plesty.lib.data.array.PlestyArray]

Return the named arrays that fully describe this data.

The names are the keys from_arrays() receives back; each array keeps its Plesty metadata (name, unit, description, …).

Return type:: dict[str, plesty.lib.data.array.PlestyArray]

classmethod from_arrays(arrays: dict[str, plesty.lib.data.array.PlestyArray], metadata: dict[str, Any], provenance: dict[str, Any]) → Self

Abstractmethod:

Parameters:

arrays (dict[str, plesty.lib.data.array.PlestyArray])
metadata (dict[str, Any])
provenance (dict[str, Any])

Return type:

Self

Rebuild an instance from the output of to_arrays().

Parameters:

arrays (dict[str, plesty.lib.data.array.PlestyArray]) – Name → array, as written by to_arrays().
metadata (dict[str, Any]) – The saved metadata.
provenance (dict[str, Any]) – The saved provenance.

Return type:

Self

classmethod example() → Self

Abstractmethod:
Return type:: Self

Return a small valid instance — for docs, mocks and pipeline tests.

header() → dict[str, Any]

Return the JSON header describing this data without its values.

Returns:: kind, schema_version, metadata, provenance and, per array, its Plesty metadata plus shape and dtype.
Return type:: dict[str, Any]

_checked_arrays() → dict[str, plesty.lib.data.array.PlestyArray]

Return to_arrays(), refusing names that collide with the header.

Return type:: dict[str, plesty.lib.data.array.PlestyArray]

save(path: str | pathlib.Path) → pathlib.Path

Write this data to path, in the format its suffix names.

The write is atomic: a temporary sibling is written and renamed, so a crash never leaves a half-written file under the final name.

Parameters:: path (str | pathlib.Path) – Destination; the suffix picks the format, one of NATIVE_FORMATS.
Returns:: The written path.
Raises:: ValueError – If the suffix is not a native format, or the arrays cannot be laid out as a table (.csv / .txt).
Return type:: pathlib.Path

to_bytes(fmt: str) → bytes

Serialize this data in a native format.

Parameters:: fmt (str) – A suffix from NATIVE_FORMATS (the dot is optional).
Returns:: The encoded file content.
Raises:: ValueError – If fmt is not native, or the arrays cannot be laid out as a table.
Return type:: bytes

_to_npz() → bytes

Encode as .npz: one entry per array plus the JSON header.

Return type:: bytes

_to_json() → bytes

Encode as .json: the header with each array’s values inlined.

Return type:: bytes

_to_table(delimiter: str) → bytes

Encode as a delimited table: one column per 1-D array or per 2-D row.

Parameters:: delimiter (str)
Return type:: bytes

classmethod load(path: str | pathlib.Path) → Self

Read data from path — a native file or a format a reader handles.

Called on ExperimentData itself, the class is taken from the file’s header. Called on a subclass, the file must hold that subclass (or one derived from it); a plain table without a Plesty header is then handed to the subclass’s from_table().

Parameters:

path (str | pathlib.Path) – The file to read.

Returns:

The reconstructed data.

Raises:

ValueError – If the format is unknown, the kind is not registered, or the file was written by a newer schema version.
TypeError – If the file holds a different kind than requested.

Return type:

Self

classmethod _reader_for(suffix: str) → Callable[[pathlib.Path], Any] | None

Find a reader for suffix on this class, or on any registered one.

Parameters:: suffix (str)
Return type:: Callable[[pathlib.Path], Any] | None

classmethod from_bytes(payload: bytes, fmt: str) → Self

Deserialize data written by to_bytes().

Parameters:

payload (bytes) – The encoded file content.
fmt (str) – The native format it is in.

Returns:

The reconstructed data.

Raises:

ValueError – As for load().
TypeError – As for load().

Return type:

Self

classmethod _from_table_bytes(payload: bytes, delimiter: str | None) → Self

Parse a delimited table, with or without a Plesty header line.

Parameters:

payload (bytes)
delimiter (str | None)

Return type:

Self

classmethod from_text(lines: list[str], delimiter: str | None) → Self

Build an instance from a text file without a Plesty header.

The default parses one delimited table — an optional row of column names, then numbers — and hands it to from_table(). Subclasses that read a text layout other than one table override this.

Parameters:

lines (list[str]) – The file’s non-empty lines, # comments removed.
delimiter (str | None) – "," for .csv; None (whitespace) for .txt.

Return type:

Self

classmethod from_table(columns: list[numpy.ndarray], names: list[str] | None) → Self

Build an instance from a plain table without a Plesty header.

The base class cannot interpret foreign columns; subclasses that know a common layout (e.g. wavelength + intensity) override this.

Parameters:

columns (list[numpy.ndarray]) – The table’s columns, left to right.
names (list[str] | None) – The column header row, when the file had one.

Raises:

ValueError – Always, for types that do not define a layout.

Return type:

Self

classmethod _rebuild(header: dict[str, Any], raw: dict[str, numpy.ndarray]) → Self

Resolve the header’s kind, check its version, and call from_arrays().

Parameters:

header (dict[str, Any])
raw (dict[str, numpy.ndarray])

Return type:

Self

__repr__() → str

Return the kind and the shape of every array.

Return type:: str

plesty.lib.data.experiment_data_type(kind: str) → type[ExperimentData] | None

Return the ExperimentData subclass registered under kind.

Parameters:: kind (str) – The kind a subclass declares, e.g. "spectrum".
Returns:: The registered subclass, or None when no subclass declares it.
Return type:: type[ExperimentData] | None

plesty.lib.data.registered_kinds() → tuple[str, Ellipsis]

Return every registered kind, sorted.

Return type:: tuple[str, Ellipsis]

class plesty.lib.data.SpectrumData(wavelength: numpy.typing.ArrayLike, intensity: numpy.typing.ArrayLike, metadata: dict[str, Any] | None = None, provenance: dict[str, Any] | None = None, coordinates: Mapping[str, numpy.typing.ArrayLike] | None = None)

Bases: plesty.lib.data.experiment_data.ExperimentData

A wavelength axis and the intensity measured on it.

Variables:

wavelength – 1-D axis, one value per pixel (default unit nm).
intensity – (pixels,) for one spectrum or (frames, pixels) for a stack (default unit counts).
coordinates – Per-frame values of the sweep the frames were taken over, by name — e.g. hwp_deg or power_w — each 1-D with one value per frame. Empty for a spectrum that is no sweep.

Parameters:

wavelength (numpy.typing.ArrayLike)
intensity (numpy.typing.ArrayLike)
metadata (dict[str, Any] | None)
provenance (dict[str, Any] | None)
coordinates (Mapping[str, numpy.typing.ArrayLike] | None)

Validate and store the axis, the intensity and the frame coordinates.

Parameters:

wavelength (numpy.typing.ArrayLike) – 1-D axis; a plain array gets unit nm.
intensity (numpy.typing.ArrayLike) – 1-D spectrum or 2-D (frames, pixels) stack; a plain array gets unit counts.
metadata (dict[str, Any] | None) – Acquisition context.
provenance (dict[str, Any] | None) – Origin of the data.
coordinates (Mapping[str, numpy.typing.ArrayLike] | None) – Per-frame sweep values by name; a plain array gets the unit its name ends in (_deg, _w, _nm, …).

Raises:

ValueError – If the shapes do not describe a spectrum on the axis, or a coordinate does not have one value per frame.

kind = 'spectrum': Registry key and schema dtype token, e.g. "spectrum".

schema_version = 1: Version of the array layout this class writes and reads.

wavelength: plesty.lib.data.array.PlestyArray

intensity: plesty.lib.data.array.PlestyArray

coordinates: dict[str, plesty.lib.data.array.PlestyArray]

to_arrays() → dict[str, plesty.lib.data.array.PlestyArray]

Return wavelength, intensity and each coordinate_<name>.

Return type:: dict[str, plesty.lib.data.array.PlestyArray]

classmethod from_arrays(arrays: dict[str, plesty.lib.data.array.PlestyArray], metadata: dict[str, Any], provenance: dict[str, Any]) → Self

Rebuild a spectrum from wavelength, intensity and its coordinates.

Parameters:

arrays (dict[str, plesty.lib.data.array.PlestyArray]) – Must hold wavelength and intensity; every coordinate_<name> becomes a coordinate.
metadata (dict[str, Any]) – The saved metadata.
provenance (dict[str, Any]) – The saved provenance.

Return type:

Self

classmethod example() → Self

Return a 101-pixel spectrum with one Lorentzian line at 775 nm.

Return type:: Self

classmethod from_text(lines: list[str], delimiter: str | None) → Self

Build a spectrum from a text export, in blocks or as one table.

A file in which a Frame … line opens each frame — followed by wavelength intensity rows, whitespace-separated, where a comma may be the decimal separator — is read block by block. Anything else is a table, see from_table().

Parameters:

lines (list[str]) – The file’s non-empty lines, # comments removed.
delimiter (str | None) – "," for .csv; None for .txt.

Raises:

ValueError – If the blocks hold no data or frames of different lengths.

Return type:

Self

classmethod from_table(columns: list[numpy.ndarray], names: list[str] | None) → Self

Build a spectrum from a plain table.

Two layouts are understood:

LightField’s CSV export — columns Frame, Wavelength, Intensity (and Row for multi-row regions, summed per pixel); one line per pixel.
Wide — the first column is the wavelength, every further column one frame of intensity.

Parameters:

columns (list[numpy.ndarray]) – The table’s columns.
names (list[str] | None) – The header row, when the file had one.

Raises:

ValueError – If the table has fewer than two columns.

Return type:

Self

classmethod _from_lightfield_table(columns: list[numpy.ndarray], lowered: list[str]) → Self

Rebuild frames from LightField’s one-line-per-pixel CSV export.

Parameters:

columns (list[numpy.ndarray])
lowered (list[str])

Return type:

Self

classmethod from_spe(path: str | pathlib.Path, region: int = 0, wavelength: numpy.typing.ArrayLike | None = None) → Self

Read a LightField SPE 3.0 file.

Rows of a multi-row region are summed per pixel (full vertical binning); a single frame becomes a 1-D spectrum, several a stack.

Parameters:

path (str | pathlib.Path) – The .spe file.
region (int) – Index of the region of interest to read.
wavelength (numpy.typing.ArrayLike | None) – Axis to use when the file has no calibration, or to override it.

Returns:

The spectrum, with the file’s acquisition settings in metadata and its name under metadata["source"].

Raises:

ValueError – If the file has no such region, or no calibration and none was given.

Return type:

Self

classmethod from_run(run: Any, file_key: str = 'data_file', coordinates: Sequence[str] = (), where: Mapping[str, Any] | None = None, wavelength: numpy.typing.ArrayLike | None = None) → Self

Stack the spectra an experiment run recorded, in record order.

Each record holding file_key names the raw file a device wrote for that step; the path is translated to this machine through the run’s data paths (a relative path is taken from the run directory). A .spe file is read with from_spe(); a .npy frame, which carries no axis, needs wavelength, and its rows are summed per pixel; any other file goes through load(). A record whose value is itself a SpectrumData is used as it is.

Parameters:

run (Any) – A Run or a run directory.
file_key (str) – Record field naming the raw file.
coordinates (Sequence[str]) – Record fields to keep per frame, e.g. ("hwp_deg", "power_w").
where (Mapping[str, Any] | None) – Only records whose fields equal these values, e.g. {"qd": "QD-004"}.
wavelength (numpy.typing.ArrayLike | None) – Axis for files without a calibration.

Returns:

One frame per recorded frame, with the requested coordinates; metadata names the run and provenance the run and experiment.

Raises:

ValueError – If no record matches, a record lacks a coordinate, or the files do not share one wavelength axis.

Return type:

Self

classmethod _read_raw(reader: Any, host_path: str, wavelength: numpy.typing.ArrayLike | None) → SpectrumData

Read one raw device file named by a record.

Parameters:

reader (Any)
host_path (str)
wavelength (numpy.typing.ArrayLike | None)

Return type:

SpectrumData

property is_stack: bool

Whether the intensity holds several frames.

Return type:: bool

property n_frames: int

Number of frames (1 for a single spectrum).

Return type:: int

property n_pixels: int

Number of pixels on the wavelength axis.

Return type:: int

frame(index: int) → Self

Return one frame as a single spectrum.

Parameters:: index (int) – Frame index; negative indices count from the end.
Raises:: IndexError – If there is no such frame.
Return type:: Self

combine(method: Literal['mean', 'sum'] = 'mean') → Self

Collapse a stack into one spectrum; the frame coordinates are dropped.

Parameters:: method (Literal['mean', 'sum']) – "mean" or "sum" over the frames.
Raises:: ValueError – If method is unknown.
Return type:: Self

crop(start: float, stop: float) → Self

Keep the pixels whose wavelength lies in [start, stop].

Parameters:

start (float) – Lower bound, in the wavelength’s unit.
stop (float) – Upper bound, in the wavelength’s unit.

Raises:

ValueError – If no pixel lies in the range.

Return type:

Self

subtract_background(background: float | Literal['min', 'median'] | numpy.typing.ArrayLike = 'median') → Self

Subtract a background from every frame.

Parameters:: background (float | Literal['min', 'median'] | numpy.typing.ArrayLike) – A constant; "min" or "median" of each frame; or an array broadcastable to the intensity (e.g. a dark frame).
Raises:: ValueError – If background is an unknown keyword.
Return type:: Self

normalize(method: Literal['max', 'area'] = 'max') → Self

Scale every frame to a peak of 1 or a unit area; the unit becomes a.u..

Parameters:: method (Literal['max', 'area']) – "max" or "area" (trapezoidal, over the axis).
Raises:: ValueError – If method is unknown or a frame’s norm is zero.
Return type:: Self

energy() → plesty.lib.data.array.PlestyArray

Return the photon energy of each pixel, in eV.

Raises:: ValueError – If the wavelength unit is not a length.
Return type:: plesty.lib.data.array.PlestyArray

find_peaks(frame: int = 0, height: float | None = None, prominence: float | None = None, distance: int = 1) → numpy.ndarray

Find local maxima in one frame (NumPy only).

A plateau counts once, at its first pixel. prominence is the height above the higher of the two lowest points separating the peak from taller neighbours (or from the ends of the axis).

Parameters:

frame (int) – Frame to search.
height (float | None) – Minimum value of a peak.
prominence (float | None) – Minimum prominence of a peak.
distance (int) – Minimum separation in pixels; the taller peak wins.

Returns:

Pixel indices of the peaks, in ascending order.

Return type:

numpy.ndarray

peak_prominences(indices: numpy.typing.ArrayLike, frame: int = 0) → numpy.ndarray

Return the prominence of each peak at indices, as find_peaks() defines it.

Parameters:

indices (numpy.typing.ArrayLike) – Pixel indices of peaks, e.g. from find_peaks().
frame (int) – Frame the peaks are in.

Returns:

One prominence per index, in the intensity’s unit.

Raises:

IndexError – If an index is outside the axis.

Return type:

numpy.ndarray

match_peaks(reference: numpy.typing.ArrayLike, tolerance: float, frame: int = 0) → numpy.ndarray

Relocate known line positions to the strongest local maximum near each.

For each reference position, the local maxima within ±tolerance are searched and the strongest is taken; without a local maximum in range, the highest pixel in range is.

Parameters:

reference (numpy.typing.ArrayLike) – Line positions, in the wavelength’s unit — e.g. lines identified in another measurement of the same emitter.
tolerance (float) – Half-width of the search range around each position.
frame (int) – Frame to search.

Returns:

One pixel index per reference position, in the order given.

Raises:

ValueError – If tolerance is not positive or a range holds no pixel.

Return type:

numpy.ndarray

_derive(wavelength: numpy.typing.ArrayLike, intensity: numpy.typing.ArrayLike, step: dict[str, Any], coordinates: Mapping[str, numpy.typing.ArrayLike] | None = None) → Self

Return a new spectrum carrying this one’s context plus step.

The frame coordinates are kept unless coordinates replaces them.

Parameters:

wavelength (numpy.typing.ArrayLike)
intensity (numpy.typing.ArrayLike)
step (dict[str, Any])
coordinates (Mapping[str, numpy.typing.ArrayLike] | None)

Return type:

Self

class plesty.lib.data.SpectrumFitData(values: Mapping[str, numpy.typing.ArrayLike], wavelength_unit: str | None = 'nm', intensity_unit: str | None = 'counts', metadata: dict[str, Any] | None = None, provenance: dict[str, Any] | None = None, coordinates: Mapping[str, numpy.typing.ArrayLike] | None = None)

Bases: plesty.lib.data.experiment_data.ExperimentData

Line-shape fit results of a spectrum: one row per fitted frame, one column per line.

Produced by plesty.lib.analyzer.SpectrumFit. Every quantity is an array of shape (frames, peaks), with nan where a frame’s fit failed; frame names the spectrum frame each row came from, and x_ref (one per peak) the position the linear background’s offset refers to.

`center`	line position (wavelength unit)
`fwhm`	full width at half maximum (wavelength unit)
`height`	peak height above the background (intensity unit)
`area`	integrated intensity (intensity × wavelength unit)
`background_offset`	background at `x_ref` (intensity unit)
`background_slope`	background slope (intensity / wavelength unit)
`r_squared`	coefficient of determination of the peak’s group

Each quantity but r_squared has a one-sigma <name>_err array. metadata records the model, the background mode and the groups of peaks fitted together. coordinates carries the fitted frames’ sweep values (hwp_deg, power_w, …) from the spectrum, one per row — what a sweep analysis plots the quantities against.

Validate the arrays and attach their units.

Parameters:

values (Mapping[str, numpy.typing.ArrayLike]) – Every name in names, as arrays.
wavelength_unit (str | None) – Unit of the spectrum’s axis.
intensity_unit (str | None) – Unit of the spectrum’s intensity.
metadata (dict[str, Any] | None) – Fit context (model, background, groups).
provenance (dict[str, Any] | None) – Origin of the data.
coordinates (Mapping[str, numpy.typing.ArrayLike] | None) – Sweep values of the fitted frames, one per row.

Raises:

ValueError – If a name is missing or the shapes disagree.

kind = 'spectrum_fit': Registry key and schema dtype token, e.g. "spectrum".

schema_version = 1: Version of the array layout this class writes and reads.

QUANTITIES: ClassVar[tuple[str, Ellipsis]] = ('center', 'fwhm', 'height', 'area', 'background_offset', 'background_slope')

values: dict[str, plesty.lib.data.array.PlestyArray]

the quantities, their errors, r_squared, frame, x_ref.

Type:: Every array by name

coordinates: dict[str, plesty.lib.data.array.PlestyArray]: Sweep values of the fitted frames by name, one per row.

classmethod names() → tuple[str, Ellipsis]

Return every array name, in storage order.

Return type:: tuple[str, Ellipsis]

static _units(wavelength: str | None, intensity: str | None) → dict[str, str | None]

Return the unit of each array from the spectrum’s units.

Parameters:

wavelength (str | None)
intensity (str | None)

Return type:

dict[str, str | None]

__getitem__(name: str) → plesty.lib.data.array.PlestyArray

Return the array called name, e.g. fit["center"].

Parameters:: name (str)
Return type:: plesty.lib.data.array.PlestyArray

property n_frames: int

Number of fitted frames.

Return type:: int

property n_peaks: int

Number of fitted lines.

Return type:: int

peak(index: int) → dict[str, plesty.lib.data.array.PlestyArray]

Return every per-frame quantity of one line, as 1-D arrays over frames.

Parameters:: index (int) – Peak index.
Raises:: IndexError – If there is no such peak.
Return type:: dict[str, plesty.lib.data.array.PlestyArray]

to_arrays() → dict[str, plesty.lib.data.array.PlestyArray]

Return every array by name, coordinates as coordinate_<name>.

Return type:: dict[str, plesty.lib.data.array.PlestyArray]

classmethod from_arrays(arrays: dict[str, plesty.lib.data.array.PlestyArray], metadata: dict[str, Any], provenance: dict[str, Any]) → Self

Rebuild fit results from to_arrays(); units travel with the arrays.

Parameters:

arrays (dict[str, plesty.lib.data.array.PlestyArray])
metadata (dict[str, Any])
provenance (dict[str, Any])

Return type:

Self

classmethod example() → Self

Return one Gaussian line at 775 nm fitted in two frames.

Return type:: Self

class plesty.lib.data.ResultDocument

Typed model of the JSON metadata document written by save_result().

Formalises the experimental-output metadata schema (issue plesty-lib#4): every persisted result is described by one such document, which is the commit record of the write. Blob-backed results (array, bytes) reference their raw data file via blob; plain values are inlined in value.

Variables:

type – Result category — one of RESULT_TYPES.
saved_at – UTC ISO-8601 write time.
provenance – Context recorded at save time (step id, operation, parameters, device identity, …).
blob – File name of the raw data blob next to the document (array/bytes/experiment_data results only).
format – Blob encoding ("npy", "npz", "png", "bin", …).
shape – Array shape (array results only).
dtype – Array dtype string (array results only).
meta – Plesty metadata of the array (name, range, options, unit, description; array results), or the kind and schema_version (experiment_data results).
value – The inlined JSON-serializable value (value results only).

type: str

saved_at: str = ''

provenance: dict[str, Any]

blob: str | None = None

format: str | None = None

shape: list[int] | None = None

dtype: str | None = None

meta: dict[str, Any] | None = None

value: Any = None

__post_init__() → None

Validate the result type and stamp the write time when missing.

Return type:: None

to_dict() → dict[str, Any]

Return the document as a JSON-serializable dictionary.

Blob-related fields that do not apply to the result type are omitted, matching the on-disk layout written since the schema’s introduction.

Return type:: dict[str, Any]

classmethod from_dict(document: dict[str, Any]) → ResultDocument

Reconstruct a document from a dictionary produced by to_dict().

Parameters:: document (dict[str, Any]) – The parsed JSON document.
Returns:: The reconstructed ResultDocument.
Raises:: ValueError – If the document declares an unknown result type.
Return type:: ResultDocument

plesty.lib.data.append_record(run_dir: str | pathlib.Path, index: int, result: Any, provenance: dict[str, Any] | None = None) → dict[str, Any]

Append one step result to the run’s records.jsonl.

A blob-backed result (array, bytes) is written to <run_dir>/data/step_<index>.<fmt> first, atomically; the document line — the commit record — is appended and fsynced afterwards, so a reader never sees a document whose blob is missing.

Parameters:

run_dir (str | pathlib.Path) – The run directory.
index (int) – Position of the step in the plan; stamped onto the line and used for the blob file name.
result (Any) – The value to persist (see save_result()).
provenance (Optional[dict[str, Any]]) – Context recorded verbatim in the document.

Returns:

The line that was written, as a dictionary — the document fields plus index; blob (when present) is relative to run_dir.

Raises:

TypeError – If the result is neither a PlestyArray, bytes, nor JSON-serializable.

Return type:

dict[str, Any]

plesty.lib.data.convert_to_hdf5(documents: Iterable[str | pathlib.Path], target: str | pathlib.Path) → pathlib.Path

Pack results written by save_result() into one HDF5 archive.

The raw blob + JSON layout stays the primary on-disk format; this is an optional post-hoc export to the unified HDF5 format preferred by issue plesty-lib#4. Each JSON document becomes one HDF5 group (named after the document stem) holding a data dataset — the decoded array, the raw encoded bytes, or the JSON-encoded inline value — with the document’s metadata and provenance stored as group attributes. Experiment data becomes a data subgroup with one dataset per array and the data’s JSON header as its header attribute.

Parameters:

documents (Iterable[str | pathlib.Path]) – Paths of the JSON metadata documents to include.
target (str | pathlib.Path) – Destination .h5 file; parent directories are created.

Returns:

The resolved path of the written HDF5 file.

Raises:

ImportError – If the hdf5 extra is not installed.
ValueError – If a document declares an unknown result type.

Return type:

pathlib.Path

plesty.lib.data.load_document(path: str | pathlib.Path) → ResultDocument

Load only the metadata document of a persisted result — no blob I/O.

Useful for browsing run directories (checking provenance, shapes, or timestamps) without paying the cost of decoding the referenced blobs.

Parameters:: path (str | pathlib.Path) – Path to the JSON metadata document (<stem>.json).
Returns:: The parsed ResultDocument.
Raises:: ValueError – If the document declares an unknown result type.
Return type:: ResultDocument

plesty.lib.data.load_result(path: str | pathlib.Path) → Any

Load a result previously written by save_result().

Parameters:: path (str | pathlib.Path) – Path to the JSON metadata document (<stem>.json).
Returns:: a PlestyArray with its metadata restored, the ExperimentData subclass that was saved, raw bytes for encoded blobs, or the inlined value.
Return type:: The reconstructed value
Raises:: ValueError – If the document declares an unknown result type.

plesty.lib.data.read_records(run_dir: str | pathlib.Path, offset: int = 0) → tuple[list[dict[str, Any]], int]

Read the record lines appended to records.jsonl since offset.

Only complete lines are returned; a line still being written (no trailing newline yet) is left for the next call, which is what makes tailing a run in progress safe. Lines that fail to parse are skipped.

Parameters:

run_dir (str | pathlib.Path) – The run directory.
offset (int) – Byte position to read from — pass back the returned offset to read only what appeared since.

Returns:

The parsed lines (see append_record()) and the offset just after the last complete line; ([], offset) while the file does not exist yet.

Return type:

tuple[list[dict[str, Any]], int]

plesty.lib.data.record_value(line: dict[str, Any], run_dir: str | pathlib.Path) → Any

Return the value a records.jsonl line stands for.

Parameters:

line (dict[str, Any]) – A line as returned by read_records().
run_dir (str | pathlib.Path) – The run directory the line’s blob is relative to.

Returns:

The inlined value, or the PlestyArray / ExperimentData / raw bytes loaded from the referenced blob.

Raises:

ValueError – If the line declares an unknown result type.

Return type:

Any

plesty.lib.data.save_result(result: Any, path_stem: str | pathlib.Path, provenance: dict[str, Any] | None = None) → pathlib.Path

Persist a measurement result as raw blob + JSON metadata document.

Parameters:

result (Any) – The value to persist — a PlestyArray, an ExperimentData, encoded bytes (e.g. an image), or any JSON-serializable value.
path_stem (str | pathlib.Path) – Destination path without suffix; the JSON document is written to <stem>.json and any blob next to it.
provenance (Optional[dict[str, Any]]) – Optional context (step id, operation, parameters, device identity) recorded verbatim in the document.

Returns:

The path of the JSON metadata document.

Raises:

TypeError – If the result is neither a PlestyArray, bytes, nor JSON-serializable.

Return type:

pathlib.Path