plesty.lib.analyzer

Analyzer framework: schema-declared transforms from raw device data to results.

Submodules

Exceptions

`AnalysisValidationError`	Raised when analysis inputs or outputs violate the declared schema.
`AnalyzerSchemaError`	Raised when an analyzer's schema declaration is invalid.
`FitQualityWarning`	Warned when fitted lines show signs of a fit that did not follow its line.
`PeakFitError`	Raised when a fit does not converge.

Classes

`Analyzer`	Base class for PLESTY analyzers: schema-declared data transforms.
`Peak`	One line to fit, with its starting values and constraints.
`PeakResult`	The fitted values of one line, each with its one-sigma error.
`SpectrumFit`	Fit every line in every frame of a spectrum.
`SpectrumFitResult`	Result of `fit_spectrum()`.

Functions

`fit_peak`(→ SpectrumFitResult)	Fit one line on a background — `fit_spectrum()` with a single `Peak`.
`fit_spectrum`(→ SpectrumFitResult)	Fit one or more line shapes on a background.

Package Contents

exception plesty.lib.analyzer.AnalysisValidationError

Bases: ValueError

Raised when analysis inputs or outputs violate the declared schema.

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

class plesty.lib.analyzer.Analyzer(**params: Any)

Bases: abc.ABC

Base class for PLESTY analyzers: schema-declared data transforms.

Subclasses declare input_schema/output_schema and implement exactly one method, analyze(). Analysis parameters (including calibration objects or references) are passed as keyword arguments to the constructor and stored on params, from where they enter the provenance stamp of every result.

Schema entries map an input/output name to a dict with keys:

dtype — required: a dtype token understood by plesty.lib.data.types.resolve_dtype() (float, array_float, table2d, or an experiment-data kind such as spectrum, …).
unit — unit filled onto outputs that do not set one.
description — human-readable meaning; filled onto outputs.
shape — arrays only: expected shape; None entries are free dimensions, e.g. [None, 2].
required — inputs only: whether the input must be supplied (default True).
title — outputs only: what the output is, as a heading.
role — outputs only: "result" (the default) for what the analysis set out to find, "intermediate" for a step on the way that is kept but not reported, such as the line fits a splitting came from.

What an analysis says about its result is report().

Store the analysis parameters and validate the schema declarations.

Parameters:: **params (Any) – Analysis parameters and calibration handles. They are introspectable via params and recorded in the provenance stamp of every result.

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

name → schema entry.

Type:: Declared inputs of analyze()

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

name → schema entry.

Type:: Declared outputs of analyze()

params: dict[str, Any]: Analysis parameters passed to the constructor (calibration included).

abstractmethod analyze(**inputs: Any) → dict[str, Any]

Transform the schema-declared inputs into the schema-declared outputs.

Implementations receive validated inputs as keyword arguments and return a dict with exactly the output_schema names as keys. Output metadata (name/unit/description) may be left unset — the base machinery fills it from the schema.

Parameters:: **inputs (Any) – Validated inputs, one keyword per input_schema name.
Returns:: Mapping of output name to result value.
Return type:: dict[str, Any]

__call__(**inputs: Any) → dict[str, Any]

Validate inputs, run analyze(), validate and stamp outputs.

Parameters:: **inputs (Any) – One keyword per input_schema name; optional inputs may be omitted.
Returns:: Mapping of output name to result value; array/table results carry a provenance attribute.
Raises:: AnalysisValidationError – If inputs or outputs violate the schema.
Return type:: dict[str, Any]

report(outputs: Mapping[str, Any]) → plesty.lib.report.Report

Say what the analysis found, in text, tables and figures.

The default tabulates every "result" output — a piece of data from its kind’s declared columns, a 1-D array as one column, and the scalars together as one table — so an analyzer is reportable the day it ships. Override it to say more: a sentence on what was found, a table laid out for a paper, a figure.

The experiment saves the result beside the outputs (Report.save), where a reader renders it without importing this class.

Parameters:: outputs (Mapping[str, Any]) – What __call__() returned.
Returns:: The report, titled with the class name and stamped with provenance().
Return type:: plesty.lib.report.Report

provenance() → dict[str, Any]

Return the provenance stamp: analyzer identity, version, parameters.

Return type:: dict[str, Any]

static _validate_schema(schema: dict[str, dict[str, Any]], allowed_keys: frozenset[str], label: str) → None

Validate a schema declaration, raising AnalyzerSchemaError.

Parameters:

schema (dict[str, dict[str, Any]])
allowed_keys (frozenset[str])
label (str)

Return type:

None

_validate_inputs(inputs: dict[str, Any]) → dict[str, Any]

Check input names, dtypes, and shapes against input_schema.

Parameters:: inputs (dict[str, Any])
Return type:: dict[str, Any]

_finalize_outputs(outputs: Any) → dict[str, Any]

Validate outputs, fill schema metadata, and stamp provenance.

Parameters:: outputs (Any)
Return type:: dict[str, Any]

_conform(name: str, value: Any, entry: dict[str, Any], direction: str) → Any

Check one value against its schema entry, coercing raw arrays.

Parameters:

name (str)
value (Any)
entry (dict[str, Any])
direction (str)

Return type:

Any

_conform_array(name: str, value: Any, entry: dict[str, Any], direction: str) → plesty.lib.data.array.PlestyArray

Coerce array-likes to PlestyArray and check the shape.

Parameters:

name (str)
value (Any)
entry (dict[str, Any])
direction (str)

Return type:

plesty.lib.data.array.PlestyArray

static _conform_scalar(name: str, value: Any, expected: type, direction: str) → Any

Check a scalar value, accepting ints where floats are declared.

Parameters:

name (str)
value (Any)
expected (type)
direction (str)

Return type:

Any

exception plesty.lib.analyzer.AnalyzerSchemaError

Bases: ValueError

Raised when an analyzer’s schema declaration is invalid.

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

exception plesty.lib.analyzer.FitQualityWarning

Bases: UserWarning

Warned when fitted lines show signs of a fit that did not follow its line.

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

class plesty.lib.analyzer.Peak

One line to fit, with its starting values and constraints.

Variables:

center – Initial line position.
width – Initial FWHM; estimated from the data when None.
center_bounds – (low, high) for the position; the fitted range when None.
width_bounds – (low, high) bounds on the FWHM in the same units as x. If omitted, the width is constrained to stay above roughly one sample spacing, preventing a narrow noise spike from fitting as a line. For a Voigt profile, the bound is applied to each of the two width components separately.
nonnegative – Keep the height at or above zero.

center: float

width: float | None = None

center_bounds: tuple[float, float] | None = None

width_bounds: tuple[float, float] | None = None

nonnegative: bool = True

exception plesty.lib.analyzer.PeakFitError

Bases: RuntimeError

Raised when a fit does not converge.

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

class plesty.lib.analyzer.PeakResult

The fitted values of one line, each with its one-sigma error.

Variables:

params – The raw line-shape parameters (height, center, sigma and/or gamma).
errors – Their one-sigma errors (nan when not estimable).

center: float

center_err: float

fwhm: float

fwhm_err: float

height: float

height_err: float

area: float

area_err: float

params: dict[str, float]

errors: dict[str, float]

class plesty.lib.analyzer.SpectrumFit(peaks: str | Sequence[float] = 'auto', model: str = 'gaussian', background: str = 'linear', window: float | None = None, frames: str | int | Sequence[int] = 'all', order: Sequence[int] | None = None, track: bool = True, overlap: float = 2.0, prominence: float | None = None, distance: int = 1, tolerance: float | None = None, width: float | None = None, width_bounds: tuple[float, float] | None = None, center_tolerance: float | None = None, nonnegative: bool = True, average: str = 'mean')

Bases: plesty.lib.analyzer.base_analyzer.Analyzer

Fit every line in every frame of a spectrum.

Lines — peaks="auto" finds them in the average of the selected frames (prominence, distance); a list of positions uses those, relocated to the nearest local maximum within tolerance if set. average="median" takes the per-pixel median of the frames instead of their mean: a cosmic ray hits one frame and drops out of the median, where in the mean of a few dozen frames it can still stand out as a line.
Groups — each line’s FWHM is estimated on the average; neighbours closer than overlap × the larger FWHM are fitted together, every other line alone. overlap=0 fits every line alone.
Range — a group is fitted over window beyond its outer lines on each side, re-centred on each frame’s starting centres so the range follows a drifting line; without a window, up to halfway to the neighbouring groups.
Frames — each group is fitted to the average first; that result seeds the first frame in order, and with track each frame seeds the next. center_tolerance keeps each line within that distance of its seed. A frame whose fit fails is nan.
Quality — nothing above is changed by it: per line, frames that failed, lacked a centre error or sat at a centre or width bound, jumps wider than the line between frames, and outlying centres are counted into fit.metadata["diagnostics"], and a FitQualityWarning names every line with a count. With track, each frame’s range is centred on the previous frame’s fitted position, so when a line is weak in one frame its fit can move onto a brighter neighbouring spectral line inside the range and stay on it; these counts are how that shows.

Outputs fit (SpectrumFitData, one row per frame in frames) and curve — the fitted model on the spectrum’s axis, nan outside the fitted ranges.

Validate and store the fit parameters.

Parameters:

peaks (str | Sequence[float]) – "auto", or line positions in the wavelength unit.
model (str) – Line shape, one of MODELS.
background (str) – One of BACKGROUNDS.
window (float | None) – Fitted range beyond a group’s outer lines, each side.
frames (str | int | Sequence[int]) – "all", one frame index, or a list of them.
order (Sequence[int] | None) – The frames in the order to fit them.
track (bool) – Seed each frame with the previous frame’s fit.
overlap (float) – Grouping factor on the FWHM; 0 never groups.
prominence (float | None) – Minimum prominence for peaks="auto".
distance (int) – Minimum separation in pixels for peaks="auto".
tolerance (float | None) – Relocate given positions to a local maximum this close.
width (float | None) – Initial FWHM of every line.
width_bounds (tuple[float, float] | None) – (low, high) FWHM bounds of every line.
center_tolerance (float | None) – Maximum distance of a line from its seed position.
nonnegative (bool) – Keep heights at or above zero.
average (str) – "mean" or "median" of the frames: the spectrum lines are found in, their widths estimated on, and the first fit made to.

Raises:

ValueError – If a parameter is out of range.

input_schema

name → schema entry.

Type:: Declared inputs of analyze()

output_schema

name → schema entry.

Type:: Declared outputs of analyze()

analyze(spectrum: plesty.lib.data.spectrum.SpectrumData) → dict[str, Any]

Fit every line in every selected frame.

Parameters:

spectrum (plesty.lib.data.spectrum.SpectrumData) – The spectrum to fit.

Returns:

fit and curve, see the class description.

Raises:

ValueError – If a frame is out of range, order does not match frames, or no line is found.
PeakFitError – If a group cannot be fitted even to the average.

Return type:

dict[str, Any]

static _frames(selection: str | int | list[int], total: int) → list[int]

Return the selected frame indices, normalised and checked.

Parameters:

selection (str | int | list[int])
total (int)

Return type:

list[int]

_order(order: list[int], frames: list[int], total: int) → list[int]

Return the fitting order, which must be a permutation of frames.

Parameters:

order (list[int])
frames (list[int])
total (int)

Return type:

list[int]

_centers(spectrum: plesty.lib.data.spectrum.SpectrumData, average: numpy.ndarray) → numpy.ndarray

Return the sorted line positions to fit.

Parameters:

spectrum (plesty.lib.data.spectrum.SpectrumData)
average (numpy.ndarray)

Return type:

numpy.ndarray

_fwhms(axis: numpy.ndarray, average: numpy.ndarray, centers: numpy.ndarray) → numpy.ndarray

Estimate each line’s FWHM with a single-line fit to the average.

Parameters:

axis (numpy.ndarray)
average (numpy.ndarray)
centers (numpy.ndarray)

Return type:

numpy.ndarray

_groups(centers: numpy.ndarray, fwhms: numpy.ndarray) → list[list[int]]

Group neighbouring lines closer than overlap × the larger FWHM.

Parameters:

centers (numpy.ndarray)
fwhms (numpy.ndarray)

Return type:

list[list[int]]

_range(axis: numpy.ndarray, centers: numpy.ndarray, fwhms: numpy.ndarray, groups: list[list[int]], g: int) → tuple[float, float]

Return the fitted range of group g.

Parameters:

axis (numpy.ndarray)
centers (numpy.ndarray)
fwhms (numpy.ndarray)
groups (list[list[int]])
g (int)

Return type:

tuple[float, float]

_center_bounds(center: float, low: float, high: float) → tuple[float, float]

Return the position bounds of a line: its tolerance, within the range.

Parameters:

center (float)
low (float)
high (float)

Return type:

tuple[float, float]

static _check_bounds(diagnostics: list[dict[str, int]], group: list[int], result: SpectrumFitResult, bounded: list[Peak], x: numpy.ndarray) → None

Count, for one frame, lines without a centre error or pinned at a bound.

Parameters:

diagnostics (list[dict[str, int]])
group (list[int])
result (SpectrumFitResult)
bounded (list[Peak])
x (numpy.ndarray)

Return type:

None

static _check_tracks(diagnostics: list[dict[str, int]], center: numpy.ndarray, frames: list[int], order: list[int], fwhms: numpy.ndarray) → None

Count, per line, jumps wider than the line between frames and outlying centres.

Parameters:

diagnostics (list[dict[str, int]])
center (numpy.ndarray)
frames (list[int])
order (list[int])
fwhms (numpy.ndarray)

Return type:

None

static _warn(diagnostics: list[dict[str, int]], centers: numpy.ndarray, unit: str | None) → None

Warn once, naming every line with a count, without changing any result.

Parameters:

diagnostics (list[dict[str, int]])
centers (numpy.ndarray)
unit (str | None)

Return type:

None

static _store(table: dict[str, numpy.ndarray], row: int, group: list[int], result: SpectrumFitResult) → None

Write one group’s fit of one frame into the result table.

Parameters:

table (dict[str, numpy.ndarray])
row (int)
group (list[int])
result (SpectrumFitResult)

Return type:

None

class plesty.lib.analyzer.SpectrumFitResult

Result of fit_spectrum().

Variables:

model – The line shape of every peak.
background – "constant" or "linear".
peaks – One PeakResult per peak, in the order given.
background_params – offset (at x_ref) and slope (0 for a constant background).
background_errors – Their one-sigma errors.
x_ref – Position the background offset refers to.
r_squared – Coefficient of determination.
values – The optimiser parameter tuple in fixed order: background terms first (offset and optionally slope), then for each peak (height, center, ...) with the width parameters in the model’s native order (sigma for Gaussian, gamma for Lorentzian, sigma, gamma for Voigt). Examples: (offset, slope, h1, c1, sigma1, h2, c2, sigma2) for two Gaussian peaks on a linear background; (offset, slope, h, c, gamma) for one Lorentzian peak on a linear background; (offset, h, c, sigma, gamma) for one Voigt peak on a constant background. This is the exact order expected by _model_function() and can be reused as a seed for a later fit.
covariance – The parameter covariance matrix in the same order as values.

model: str

background: str

peaks: tuple[PeakResult, Ellipsis]

background_params: dict[str, float]

background_errors: dict[str, float]

x_ref: float

r_squared: float

values: tuple[float, Ellipsis]

covariance: numpy.ndarray

evaluate(x: numpy.typing.ArrayLike) → numpy.ndarray

Evaluate the fitted model — background plus every line — at x.

Parameters:: x (numpy.typing.ArrayLike)
Return type:: numpy.ndarray

to_dict() → dict[str, Any]

Return the result as a JSON-serializable dictionary (no covariance).

Return type:: dict[str, Any]

plesty.lib.analyzer.fit_peak(x: numpy.typing.ArrayLike, y: numpy.typing.ArrayLike, model: str = 'gaussian', *, center: float | None = None, width: float | None = None, background: str = 'linear', sigma: numpy.typing.ArrayLike | None = None, center_bounds: tuple[float, float] | None = None, width_bounds: tuple[float, float] | None = None, nonnegative: bool = True) → SpectrumFitResult

Fit one line on a background — fit_spectrum() with a single Peak.

Parameters:

x (numpy.typing.ArrayLike) – Positions, 1-D.
y (numpy.typing.ArrayLike) – Values at x.
model (str) – Line shape, one of MODELS.
center (float | None) – Initial line position; the highest point by default.
width (float | None) – Initial FWHM.
background (str) – One of BACKGROUNDS.
sigma (numpy.typing.ArrayLike | None) – One-sigma error of each y.
center_bounds (tuple[float, float] | None) – (low, high) for the position.
width_bounds (tuple[float, float] | None) – (low, high) for the FWHM.
nonnegative (bool) – Keep the height at or above zero.

Returns:

The fit result; result.peaks[0] holds the line.

Return type:

SpectrumFitResult

plesty.lib.analyzer.fit_spectrum(x: numpy.typing.ArrayLike, y: numpy.typing.ArrayLike, peaks: Sequence[Peak], model: str = 'gaussian', background: str = 'linear', *, sigma: numpy.typing.ArrayLike | None = None, seed: SpectrumFitResult | None = None, x_ref: float | None = None) → SpectrumFitResult

Fit one or more line shapes on a background.

Without a seed, starting values come from the data: the background from a fit to the points away from the lines, each height from the data at its centre, each width from the half-maximum crossings (or Peak.width). Starting values are moved inside the bounds.

Parameters:

x (numpy.typing.ArrayLike) – Positions, 1-D.
y (numpy.typing.ArrayLike) – Values at x, same shape.
peaks (Sequence[Peak]) – The lines to fit.
model (str) – Line shape of every peak, one of MODELS.
background (str) – One of BACKGROUNDS.
sigma (numpy.typing.ArrayLike | None) – One-sigma error of each y, for a weighted fit. The covariance is scaled by the reduced chi-square, as a relative weighting.
seed (SpectrumFitResult | None) – A previous result with the same peaks, model and background, whose values start this fit.
x_ref (float | None) – Position the background offset refers to; the mean of the peak centres by default.

Returns:

The fit result.

Raises:

ValueError – If an argument is unknown or the data are unusable.
PeakFitError – If the optimiser does not converge.
ImportError – If SciPy (the fit extra) is not installed.

Return type:

SpectrumFitResult