Data-model API#
.. py:class:: BaseData(signal: int | float | ~numpy.ndarray, units, uncertainties: ~typing.Dict[str, int | float | ~numpy.ndarray] = NOTHING, weights: int | float | ~numpy.ndarray = array(1.), axes: ~typing.List[~typing.Self | None] = NOTHING, rank_of_data=0) :module: modacor.dataclasses.basedata
BaseData stores a core data array (signal) with associated uncertainties, units,
and metadata. It validates that any weights or uncertainty arrays broadcast to
the shape of signal, and provides utilities for scaling operations and unit conversion.
:ivar signal: The primary data array. All weights/uncertainty arrays must be broadcastable
to this shape.
:vartype signal: np.ndarray
:ivar units: Physical units of signal*scaling and their uncertainties.
:vartype units: pint.Unit
:ivar uncertainties: Uncertainty (as one‐sigma standard deviation) arrays keyed by type (e.g. “Poisson”,
“pixel_index”). Each array must broadcast to signal.shape. Variances are computed
as uncertainties[k]**2.
Binary operations propagate uncertainties using first-order, uncorrelated propagation.
Keys are treated as *independent sources*.
Special key: ``"propagate_to_all"``
---------------------------------
``"propagate_to_all"`` is treated as a *global* uncertainty **only when it is the sole
key present** on an operand.
The propagation rules for an operation ``result = op(a, b)`` are:
1. If both operands only contain ``"propagate_to_all"``, the result contains only
``"propagate_to_all"`` (propagated normally).
2. If one operand only contains ``"propagate_to_all"`` and the other contains one or
more non-``"propagate_to_all"`` keys, the result contains only those non-global keys.
The global uncertainty contributes as a fallback term to each of those keys.
3. If both operands contain matching non-global keys, those keys are propagated and
combined by key.
4. If both operands contain non-global keys but with no matches, the result contains
the union of keys, and each key is propagated from its originating operand only
(no cross-key combining).
:vartype uncertainties: Dict[str, np.ndarray]
:ivar weights: Weights for signal (default is a scaling 1.0) for use in averaging operations.
Must broadcast to signal.shape.
:vartype weights: np.ndarray, optional
:ivar axes: Optional metadata for each axis of signal. Defaults to an empty list.
:vartype axes: List[BaseData | None]
:ivar rank_of_data: Rank (0–3) of the data; 1 is line data, 2 is image data. Must not exceed
signal.ndim.
:vartype rank_of_data: int, default=0
:ivar Properties:
:ivar ———-:
:ivar variances: Returns {k: u**2 for k, u in uncertainties.items()}. Assigning expects a dict
of variance arrays; each is validated against signal.shape and
converted into uncertainties[k] = sqrt(var).
:vartype variances: Dict[str, np.ndarray]
:ivar shape: Shape of the signal array.
:vartype shape: tuple[int, …]
:ivar size: Size of the signal array.
:vartype size: int
.. method:: to_units(new_units: pint.Unit):
Converts internal `signal` and all `uncertainties` to `new_units` if compatible with
the existing `units`. Raises `TypeError` or `ValueError` on invalid input.
.. method:: to_base_units(): Converts internal signal and uncertainties to the base units implied by current units.
.. method:: to_dimensionless(): Converts internal signal and uncertainties to dimensionless units if possible.
.. method:: is_dimensionless() -> bool:
Returns True if `units` is compatible with dimensionless, False otherwise.
.. py:method:: BaseData.copy(with_axes: bool = True) -> ~modacor.dataclasses.basedata.BaseData :module: modacor.dataclasses.basedata
Return a new BaseData with copied signal/uncertainties/weights.
Axes are shallow-copied (list copy) by default, so axis objects
themselves are still shared.
.. py:method:: BaseData.indexed(indexer: ~typing.Any, *, rank_of_data: int | None = None) -> ~modacor.dataclasses.basedata.BaseData :module: modacor.dataclasses.basedata
Return a new BaseData corresponding to ``self`` indexed by ``indexer``.
:Parameters: * **indexer** -- Any valid NumPy indexer (int, slice, tuple of slices, boolean mask, ...),
applied consistently to ``signal`` and all uncertainty / weight arrays.
* **rank_of_data** -- Optional explicit rank_of_data for the returned BaseData. If omitted,
it will default to ``min(self.rank_of_data, result.signal.ndim)``.
.. rubric:: Notes
- Units are preserved.
- Uncertainties and weights are sliced with the same indexer where possible.
- Axes handling is conservative: existing axes are kept unchanged. If you
want axes to track slicing semantics more strictly, a higher-level
helper can be added later.
.. py:property:: BaseData.is_dimensionless :module: modacor.dataclasses.basedata :type: bool
Check if the BaseData is dimensionless.
:returns: *bool* -- True if the units are compatible with dimensionless, False otherwise.
.. py:property:: BaseData.shape :module: modacor.dataclasses.basedata :type: tuple[int, …]
Get the shape of the BaseData signal.
:returns: *tuple[int, ...]* -- The shape of the signal.
.. py:property:: BaseData.size :module: modacor.dataclasses.basedata :type: int
Get the size of the BaseData signal.
:returns: *int* -- The size of the signal.
.. py:method:: BaseData.to_base_units(*, multiplicative_conversion: bool = True) -> None :module: modacor.dataclasses.basedata
Convert the signal and uncertainties to the pint *base units* of `self.units`.
.. rubric:: Notes
- This is an in-place operation.
- For offset / non-multiplicative units (e.g. degC <-> K), this will raise unless
explicit support is implemented (see `to_units`).
.. py:method:: BaseData.to_dimensionless() -> None :module: modacor.dataclasses.basedata
Convert the signal and uncertainties to dimensionless units if possible.
.. py:method:: BaseData.to_units(new_units: pint.Unit, multiplicative_conversion=True) -> None :module: modacor.dataclasses.basedata
Convert the signal and uncertainties to new units.
.. py:property:: BaseData.variances :module: modacor.dataclasses.basedata :type: ~modacor.dataclasses.basedata._VarianceDict
A dict‐like view of variances.
• Reading: bd.variances['foo'] → returns uncertainties['foo']**2
• Writing: bd.variances['foo'] = var_array → validates + sets uncertainties['foo']=sqrt(var_array)
• Deleting: del bd.variances['foo'] → removes 'foo' from uncertainties
.. py:class:: DataBundle(*args, **kwargs) :module: modacor.dataclasses.databundle
DataBundle is a specialized data class for storing related data. It contains a dictionary of BaseData data elements, for example Signal, a wavelength and flux spectrum, Qx, Qy, Qz, Psi, etc. Process steps can add further BaseData objects to this bundle.
.. py:method:: DataBundle.setdefault(key, default=None) :module: modacor.dataclasses.databundle
Insert key with a value of default if key is not in the dictionary.
Return the value for key if key is in the dictionary, else default.
.. py:method:: DataBundle.update([E, ]**F) -> None. Update D from mapping/iterable E and F. :module: modacor.dataclasses.databundle
If E is present and has a .keys() method, then does: for k in E.keys(): D[k] = E[k]
If E is present and lacks a .keys() method, then does: for k, v in E: D[k] = v
In either case, this is followed by: for k in F: D[k] = F[k]
.. py:class:: ProcessingData :module: modacor.dataclasses.processing_data
The ProcessingData class is a dictionary-like object that stores reference to DataBundles.
.. py:class:: TraceEvent(step_id: str, module: str, label: str = ‘’, module_path: str = ‘’, version: str = ‘’, requires_steps: tuple[str, …] = NOTHING, config: dict[str, ~typing.Any] = NOTHING, datasets: dict[str, ~typing.Any] = NOTHING, messages: list[dict[str, ~typing.Any]] = NOTHING, duration_s: float | None = None, chunk_identity: dict[str, ~typing.Any] | None = None) :module: modacor.dataclasses.trace_event
A small, UI-friendly trace record for a single executed step.
Intended to be embedded into Pipeline.to_spec() so graph viewers can show: - configuration used by the step - what changed (units/dimensionality/shape/NaNs/etc.) - optional human messages (later)
.. rubric:: Notes
Keep this JSON-friendly and lightweight: no arrays, no heavy objects.
.. py:method:: TraceEvent.to_dict() -> dict[str, ~typing.Any] :module: modacor.dataclasses.trace_event
JSON-serializable representation suitable for Pipeline.to_spec().