Observations

The numeric layer: radial signal data in one of two explicit modes. See Per-scan radius axes.

Observations

Observations(dataset: Dataset)

xarray-backed radial signal store (shared or per-scan radius axes).

Prefer the :meth:from_shared_axis and :meth:from_per_scan factories. The constructor accepts a pre-built :class:xarray.Dataset and validates it, which supports round-tripping and defensive checks on hand-built datasets.

Source code in src/openauc/models/observations.py

def __init__(self, dataset: xr.Dataset) -> None:
    self._validate_dataset(dataset)
    self._dataset = dataset

dataset property

dataset: Dataset

The backing :class:xarray.Dataset. Treat as immutable.

mode property

mode: RadiusAxisMode

Whether the radius axis is shared or per-scan.

signal_unit property

signal_unit: Unit

The retained signal unit.

radius_unit property

radius_unit: Unit

The retained radius unit.

scan_ids property

scan_ids: tuple[str, ...]

Ordered scan identifiers along the scan dimension.

n_scans property

n_scans: int

Number of scans.

from_shared_axis classmethod

from_shared_axis(
    *,
    radius: Sequence[float] | NDArray[float64],
    signal: Sequence[Sequence[float]] | NDArray[float64],
    scan_ids: Sequence[str],
    signal_unit: Unit = Unit.UNKNOWN,
    radius_unit: Unit = Unit.CENTIMETRE,
) -> Observations

Build shared-axis observations from one radius axis and a 2-D signal.

Parameters:

Name	Type	Description	Default
radius	Sequence[float] | NDArray[float64]	1-D shared radius axis of length m.	required
signal	Sequence[Sequence[float]] | NDArray[float64]	2-D signal of shape (n_scans, m).	required
scan_ids	Sequence[str]	n_scans unique, non-empty identifiers.	required
signal_unit	Unit	Declared signal unit (retained, not converted).	UNKNOWN
radius_unit	Unit	Declared radius unit (default centimetres).	CENTIMETRE

Source code in src/openauc/models/observations.py

@classmethod
def from_shared_axis(
    cls,
    *,
    radius: Sequence[float] | NDArray[np.float64],
    signal: Sequence[Sequence[float]] | NDArray[np.float64],
    scan_ids: Sequence[str],
    signal_unit: Unit = Unit.UNKNOWN,
    radius_unit: Unit = Unit.CENTIMETRE,
) -> Observations:
    """Build shared-axis observations from one radius axis and a 2-D signal.

    Args:
        radius: 1-D shared radius axis of length ``m``.
        signal: 2-D signal of shape ``(n_scans, m)``.
        scan_ids: ``n_scans`` unique, non-empty identifiers.
        signal_unit: Declared signal unit (retained, not converted).
        radius_unit: Declared radius unit (default centimetres).
    """
    radius_arr = np.asarray(radius, dtype=float)
    signal_arr = np.asarray(signal, dtype=float)
    if radius_arr.ndim != 1:
        raise ObservationError("shared radius axis must be 1-D")
    if signal_arr.ndim != 2:
        raise ObservationError("signal must be 2-D with dims (scan, radius)")
    n_scans, n_radius = signal_arr.shape
    if radius_arr.shape[0] != n_radius:
        raise ObservationError(
            "radius length must match the signal's radius dimension"
        )
    if len(scan_ids) != n_scans:
        raise ObservationError(
            "number of scan_ids must match the signal's scan dimension"
        )
    _unique_or_raise(scan_ids)
    if not bool(np.all(np.isfinite(radius_arr))):
        raise ObservationError("radius contains non-finite values")
    if not bool(np.all(np.isfinite(signal_arr))):
        raise ObservationError("signal contains non-finite values")
    dataset = xr.Dataset(
        data_vars={"signal": (("scan", "radius"), signal_arr)},
        coords={
            "scan_id": ("scan", list(scan_ids)),
            "radius": ("radius", radius_arr),
        },
        attrs={
            _MODE_ATTR: RadiusAxisMode.SHARED.value,
            _SIGNAL_UNIT_ATTR: Unit(signal_unit).value,
            _RADIUS_UNIT_ATTR: Unit(radius_unit).value,
        },
    )
    return cls(dataset)

from_per_scan classmethod

from_per_scan(
    *,
    radii: Sequence[Sequence[float] | NDArray[float64]],
    signals: Sequence[Sequence[float] | NDArray[float64]],
    scan_ids: Sequence[str],
    signal_unit: Unit = Unit.UNKNOWN,
    radius_unit: Unit = Unit.CENTIMETRE,
) -> Observations

Build per-scan observations, padding to a rectangle with a mask.

Each scan's radius/signal vectors may differ in length. They are stored in (scan, point) arrays padded with NaN; the accompanying mask marks the real observations. No interpolation is performed.

Source code in src/openauc/models/observations.py

@classmethod
def from_per_scan(
    cls,
    *,
    radii: Sequence[Sequence[float] | NDArray[np.float64]],
    signals: Sequence[Sequence[float] | NDArray[np.float64]],
    scan_ids: Sequence[str],
    signal_unit: Unit = Unit.UNKNOWN,
    radius_unit: Unit = Unit.CENTIMETRE,
) -> Observations:
    """Build per-scan observations, padding to a rectangle with a mask.

    Each scan's radius/signal vectors may differ in length. They are stored
    in ``(scan, point)`` arrays padded with ``NaN``; the accompanying mask
    marks the real observations. No interpolation is performed.
    """
    n_scans = len(scan_ids)
    if len(radii) != n_scans or len(signals) != n_scans:
        raise ObservationError("radii, signals and scan_ids must have equal length")
    _unique_or_raise(scan_ids)

    radius_rows = [np.asarray(r, dtype=float) for r in radii]
    signal_rows = [np.asarray(s, dtype=float) for s in signals]
    lengths: list[int] = []
    for i, (r, s) in enumerate(zip(radius_rows, signal_rows, strict=True)):
        if r.ndim != 1 or s.ndim != 1:
            raise ObservationError(f"scan {i}: radius and signal must be 1-D")
        if r.shape[0] != s.shape[0]:
            raise ObservationError(
                f"scan {i}: radius and signal lengths differ "
                f"({r.shape[0]} vs {s.shape[0]})"
            )
        if not bool(np.all(np.isfinite(r))):
            raise ObservationError(f"scan {i}: radius contains non-finite values")
        if not bool(np.all(np.isfinite(s))):
            raise ObservationError(f"scan {i}: signal contains non-finite values")
        lengths.append(int(r.shape[0]))

    max_len = max(lengths) if lengths else 0
    radius_arr = np.full((n_scans, max_len), np.nan, dtype=float)
    signal_arr = np.full((n_scans, max_len), np.nan, dtype=float)
    mask_arr = np.zeros((n_scans, max_len), dtype=bool)
    for i, (r, s) in enumerate(zip(radius_rows, signal_rows, strict=True)):
        length = lengths[i]
        radius_arr[i, :length] = r
        signal_arr[i, :length] = s
        mask_arr[i, :length] = True

    dataset = xr.Dataset(
        data_vars={
            "radius": (("scan", "point"), radius_arr),
            "signal": (("scan", "point"), signal_arr),
            "mask": (("scan", "point"), mask_arr),
        },
        coords={"scan_id": ("scan", list(scan_ids))},
        attrs={
            _MODE_ATTR: RadiusAxisMode.PER_SCAN.value,
            _SIGNAL_UNIT_ATTR: Unit(signal_unit).value,
            _RADIUS_UNIT_ATTR: Unit(radius_unit).value,
        },
    )
    return cls(dataset)

points_per_scan

points_per_scan() -> tuple[int, ...]

Count of real observations per scan (mask sum, or width if shared).

Source code in src/openauc/models/observations.py

def points_per_scan(self) -> tuple[int, ...]:
    """Count of real observations per scan (mask sum, or width if shared)."""
    if self.mode is RadiusAxisMode.SHARED:
        width = int(self._dataset.sizes["radius"])
        return tuple(width for _ in range(self.n_scans))
    mask = self._dataset["mask"].to_numpy()
    return tuple(int(row.sum()) for row in mask)

valid_radius_values

valid_radius_values() -> NDArray[np.float64]

All real radius observations, flattened (padding excluded).

Source code in src/openauc/models/observations.py

def valid_radius_values(self) -> NDArray[np.float64]:
    """All real radius observations, flattened (padding excluded)."""
    if self.mode is RadiusAxisMode.SHARED:
        return np.asarray(self._dataset["radius"].to_numpy(), dtype=float)
    radius = self._dataset["radius"].to_numpy()
    mask = self._dataset["mask"].to_numpy()
    return np.asarray(radius[mask], dtype=float)

radius_range

radius_range() -> tuple[float, float] | None

(min, max) over real radius observations, or None if empty.

Source code in src/openauc/models/observations.py

def radius_range(self) -> tuple[float, float] | None:
    """(min, max) over real radius observations, or ``None`` if empty."""
    values = self.valid_radius_values()
    if values.size == 0:
        return None
    return (float(values.min()), float(values.max()))

scan_vectors

scan_vectors(
    scan_id: str,
) -> tuple[NDArray[np.float64], NDArray[np.float64]]

The (radius, signal) vectors of one scan, padding excluded.

Values and their order are returned exactly as stored — nothing is sorted, resampled or interpolated. In shared mode the shared radius axis is returned for every scan; in per-scan mode each scan's own axis is returned, with positions whose mask entry is False removed.

Raises:

Type	Description
KeyError	if scan_id is not present in this observation set.

Source code in src/openauc/models/observations.py

def scan_vectors(
    self, scan_id: str
) -> tuple[NDArray[np.float64], NDArray[np.float64]]:
    """The ``(radius, signal)`` vectors of one scan, padding excluded.

    Values and their **order are returned exactly as stored** — nothing is
    sorted, resampled or interpolated. In shared mode the shared radius axis
    is returned for every scan; in per-scan mode each scan's own axis is
    returned, with positions whose mask entry is ``False`` removed.

    Raises:
        KeyError: if ``scan_id`` is not present in this observation set.
    """
    try:
        index = self.scan_ids.index(scan_id)
    except ValueError as exc:
        raise KeyError(f"no scan with id {scan_id!r} in observations") from exc
    if self.mode is RadiusAxisMode.SHARED:
        radius = np.asarray(self._dataset["radius"].to_numpy(), dtype=float)
        signal = np.asarray(self._dataset["signal"].to_numpy()[index], dtype=float)
        return radius, signal
    keep = self._dataset["mask"].to_numpy()[index]
    radius = np.asarray(
        self._dataset["radius"].to_numpy()[index][keep], dtype=float
    )
    signal = np.asarray(
        self._dataset["signal"].to_numpy()[index][keep], dtype=float
    )
    return radius, signal

iter_scan_vectors

iter_scan_vectors() -> Iterator[
    tuple[str, NDArray[np.float64], NDArray[np.float64]]
]

Yield (scan_id, radius, signal) for every scan, in stored order.

Source code in src/openauc/models/observations.py

def iter_scan_vectors(
    self,
) -> Iterator[tuple[str, NDArray[np.float64], NDArray[np.float64]]]:
    """Yield ``(scan_id, radius, signal)`` for every scan, in stored order."""
    for scan_id in self.scan_ids:
        radius, signal = self.scan_vectors(scan_id)
        yield scan_id, radius, signal

to_dict

to_dict() -> dict[str, Any]

Serialise to plain JSON-friendly Python types.

Per-scan padding is written as None (JSON null) so it is never mistaken for a measured value; the authoritative mask is written too.

Source code in src/openauc/models/observations.py

def to_dict(self) -> dict[str, Any]:
    """Serialise to plain JSON-friendly Python types.

    Per-scan padding is written as ``None`` (JSON ``null``) so it is never
    mistaken for a measured value; the authoritative mask is written too.
    """
    common: dict[str, Any] = {
        _MODE_ATTR: self.mode.value,
        _SIGNAL_UNIT_ATTR: self.signal_unit.value,
        _RADIUS_UNIT_ATTR: self.radius_unit.value,
        "scan_ids": list(self.scan_ids),
    }
    if self.mode is RadiusAxisMode.SHARED:
        common["radius"] = self._dataset["radius"].to_numpy().tolist()
        common["signal"] = self._dataset["signal"].to_numpy().tolist()
        return common
    mask = self._dataset["mask"].to_numpy()
    radius = self._dataset["radius"].to_numpy()
    signal = self._dataset["signal"].to_numpy()
    common["mask"] = [[bool(v) for v in row] for row in mask]
    common["radius"] = _rows_with_nulls(radius, mask)
    common["signal"] = _rows_with_nulls(signal, mask)
    return common

from_dict classmethod

from_dict(data: dict[str, Any]) -> Observations

Reconstruct from :meth:to_dict output.

Source code in src/openauc/models/observations.py

@classmethod
def from_dict(cls, data: dict[str, Any]) -> Observations:
    """Reconstruct from :meth:`to_dict` output."""
    mode = RadiusAxisMode(str(data[_MODE_ATTR]))
    signal_unit = Unit(str(data[_SIGNAL_UNIT_ATTR]))
    radius_unit = Unit(str(data[_RADIUS_UNIT_ATTR]))
    scan_ids = [str(x) for x in data["scan_ids"]]
    if mode is RadiusAxisMode.SHARED:
        return cls.from_shared_axis(
            radius=np.asarray(data["radius"], dtype=float),
            signal=np.asarray(data["signal"], dtype=float),
            scan_ids=scan_ids,
            signal_unit=signal_unit,
            radius_unit=radius_unit,
        )
    mask = data["mask"]
    radii: list[list[float]] = []
    signals: list[list[float]] = []
    for i in range(len(scan_ids)):
        row_mask = [bool(v) for v in mask[i]]
        radius_row = data["radius"][i]
        signal_row = data["signal"][i]
        radii.append(
            [float(radius_row[j]) for j, keep in enumerate(row_mask) if keep]
        )
        signals.append(
            [float(signal_row[j]) for j, keep in enumerate(row_mask) if keep]
        )
    return cls.from_per_scan(
        radii=radii,
        signals=signals,
        scan_ids=scan_ids,
        signal_unit=signal_unit,
        radius_unit=radius_unit,
    )