plesty.lib.analyzer.base_analyzer

Synchronous Analyzer ABC with schema-declared inputs/outputs and provenance.

An Analyzer converts raw device data into meaningful data (issue plesty-lib#23). It is the middle stage of the platform pipeline device drivers → analyzers → experiments and is deliberately synchronous: analysis is CPU-bound compute, unlike the I/O orchestration of Experiment. Analyzers never talk to hardware — they consume data-layer objects produced by devices or experiments.

Subclasses declare their interface with two class-level schemas using the same dtype vocabulary as device schemas (plesty.lib.data.types.resolve_dtype()), and implement a single analyze() method:

from plesty.lib.analyzer import Analyzer
from plesty.lib.data import PlestyArray

class SpectrumNormalizer(Analyzer):
    input_schema = {
        "spectrum": {"dtype": "array_float", "unit": "counts",
                     "description": "Raw detector counts."},
    }
    output_schema = {
        "normalized": {"dtype": "array_float", "unit": "a.u.",
                       "description": "Counts scaled to peak 1."},
    }

    def analyze(self, spectrum: PlestyArray) -> dict:
        peak = float(spectrum.max()) or 1.0
        return {"normalized": PlestyArray(spectrum / peak)}

result = SpectrumNormalizer()(spectrum=raw_counts)["normalized"]
# result.unit == "a.u."; result.provenance records analyzer + params

Calling the analyzer (analyzer(**inputs)) runs the base machinery: inputs are validated against input_schema, analyze() executes, outputs are validated against output_schema with missing metadata (name/unit/description) filled from the schema, and every attribute-capable output (PlestyArray, PlestyTable2D) is stamped with a provenance dict recording the analyzer class, package version, and constructor parameters — calibration included, which is why calibration handles must be passed to __init__ and never hidden in module state.

Attributes

_INPUT_ENTRY_KEYS

_OUTPUT_ENTRY_KEYS

_PROVENANCE_BASIC

Exceptions

AnalyzerSchemaError

Raised when an analyzer's schema declaration is invalid.

AnalysisValidationError

Raised when analysis inputs or outputs violate the declared schema.

Classes

Analyzer

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

Functions

_distribution_version(→ str | None)

Best-effort version of the distribution that ships cls.

_provenance_value(→ Any)

Coerce a constructor parameter into a provenance-safe representation.

Module Contents

exception plesty.lib.analyzer.base_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.base_analyzer.AnalysisValidationError

Bases: ValueError

Raised when analysis inputs or outputs violate the declared schema.

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

plesty.lib.analyzer.base_analyzer._INPUT_ENTRY_KEYS
plesty.lib.analyzer.base_analyzer._OUTPUT_ENTRY_KEYS
plesty.lib.analyzer.base_analyzer._PROVENANCE_BASIC
plesty.lib.analyzer.base_analyzer._distribution_version(cls: type) → str | None

Best-effort version of the distribution that ships cls.

A module plesty.<name>... maps to the distribution plesty-<name> per the platform packaging convention. Falls back to the top-level package’s __version__ attribute, then to None.

Parameters:

cls (type)

Return type:

str | None

plesty.lib.analyzer.base_analyzer._provenance_value(value: Any) → Any

Coerce a constructor parameter into a provenance-safe representation.

Parameters:

value (Any)

Return type:

Any

class plesty.lib.analyzer.base_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, …).

  • 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).

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]

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