From ca1cafdd9b49703d58fc6a70a788138d04181072 Mon Sep 17 00:00:00 2001 From: Chris Meyer <34664+cmeyer@users.noreply.github.com> Date: Wed, 3 Jun 2026 13:57:14 -0700 Subject: [PATCH 01/16] Add annotated array class as successor to DataAndMetadata/Calibration. --- nion/data/AnnotatedArray.py | 283 ++++++++++++++++++++++++++++++++++++ 1 file changed, 283 insertions(+) create mode 100644 nion/data/AnnotatedArray.py diff --git a/nion/data/AnnotatedArray.py b/nion/data/AnnotatedArray.py new file mode 100644 index 0000000..0b3de20 --- /dev/null +++ b/nion/data/AnnotatedArray.py @@ -0,0 +1,283 @@ +"""Annotated n-dimensional arrays with calibrated, labeled axes. + +This module pairs a numpy array with a :class:`DataDescriptor` describing its +axes (grouped into :class:`AxisSet`/:class:`BoundAxisSet`), per-axis and +intensity :class:`Calibration` mappings between indices and physical +coordinates, and associated metadata. +""" + +from __future__ import annotations + +import dataclasses +import datetime +import typing +import types + +import numpy +import tzlocal + + +DEFAULT_CALIBRATION_KEY = "default" + + +@typing.runtime_checkable +class Calibration(typing.Protocol): + """Protocol converting between array indices and physical coordinates.""" + + @property + def unit(self) -> str: ... + + def to_coordinate(self, index: float) -> float: ... + def to_index(self, coordinate: float) -> float: ... + + +@dataclasses.dataclass(frozen=True) +class AffineCalibration(Calibration): + """Linear calibration of the form ``coordinate = offset + index * scale``.""" + + scale: float = 1.0 + offset: float = 0.0 + unit: str = "" + + def __post_init__(self) -> None: + if self.scale == 0.0: + raise ValueError("AffineCalibration scale must be non-zero") + + def to_coordinate(self, index: float) -> float: + return self.offset + index * self.scale + + def to_index(self, coordinate: float) -> float: + return (coordinate - self.offset) / self.scale + + +@dataclasses.dataclass(frozen=True) +class CalibrationSet: + """Keyed calibrations with an explicit primary calibration key.""" + + calibrations: typing.Mapping[str, Calibration] = dataclasses.field(default_factory=lambda: {DEFAULT_CALIBRATION_KEY: AffineCalibration()}) + primary_key: str = DEFAULT_CALIBRATION_KEY + + def __post_init__(self) -> None: + calibrations = dict(self.calibrations) + if not calibrations: + raise ValueError("calibrations must not be empty") + if self.primary_key not in calibrations: + raise ValueError(f"primary_key {self.primary_key!r} is not present in calibrations") + object.__setattr__(self, "calibrations", types.MappingProxyType(calibrations)) + + @property + def calibration_keys(self) -> tuple[str, ...]: + return tuple(self.calibrations.keys()) + + @property + def primary(self) -> Calibration: + return self.calibrations[self.primary_key] + + def has(self, key: str) -> bool: + return key in self.calibrations + + def get(self, key: str | None = None) -> Calibration: + target_key = self.primary_key if key is None else key + calibration = self.calibrations.get(target_key) + if calibration is None: + raise KeyError(f"Unknown calibration {target_key!r}") + return calibration + + @staticmethod + def from_calibration(calibration: Calibration, key: str = DEFAULT_CALIBRATION_KEY) -> CalibrationSet: + return CalibrationSet(calibrations={key: calibration}, primary_key=key) + + def with_primary(self, key: str) -> CalibrationSet: + if key not in self.calibrations: + raise KeyError(f"Unknown calibration {key!r}") + return dataclasses.replace(self, primary_key=key) + + def with_calibration(self, key: str, calibration: Calibration, *, make_primary: bool = False) -> CalibrationSet: + calibrations = dict(self.calibrations) + calibrations[key] = calibration + primary_key = key if make_primary else self.primary_key + return CalibrationSet(calibrations=calibrations, primary_key=primary_key) + + +@dataclasses.dataclass(frozen=True) +class Axis: + """A named dimension with a primary calibration and optional auxiliaries.""" + + label: str = "" + calibrations: CalibrationSet = dataclasses.field(default_factory=CalibrationSet) + + def get_calibration(self, key: str | None = None) -> Calibration: + return self.calibrations.get(key) + + def with_primary_calibration(self, key: str) -> Axis: + return dataclasses.replace(self, calibrations=self.calibrations.with_primary(key)) + + def with_calibration(self, key: str, calibration: Calibration, *, make_primary: bool = False) -> Axis: + return dataclasses.replace( + self, + calibrations=self.calibrations.with_calibration(key, calibration, make_primary=make_primary), + ) + + @property + def calibration_keys(self) -> tuple[str, ...]: + return self.calibrations.calibration_keys + + @property + def primary_calibration_key(self) -> str: + return self.calibrations.primary_key + + @property + def unit(self) -> str: + return self.calibrations.primary.unit + + @classmethod + def from_calibration(cls, label: str, calibration: Calibration, key: str = DEFAULT_CALIBRATION_KEY) -> Axis: + return cls(label=label, calibrations=CalibrationSet(calibrations={key: calibration}, primary_key=key)) + + +@dataclasses.dataclass(frozen=True) +class BoundAxis: + """An :class:`Axis` bound to a concrete (positive) length.""" + + axis: Axis + size: int + + def __post_init__(self) -> None: + if self.size <= 0: + raise ValueError(f"BoundAxis size must be positive, got {self.size}") + + +@dataclasses.dataclass(frozen=True) +class AxisSet: + """A named, ordered group of axes (e.g. spatial or spectral).""" + + name: str + axes: tuple[Axis, ...] = dataclasses.field(default_factory=tuple) + + @property + def rank(self) -> int: + return len(self.axes) + + @property + def units(self) -> list[str]: + return [ax.unit for ax in self.axes] + + def get_calibration(self, axis: int, key: str | None = None) -> Calibration: + return self.axes[axis].get_calibration(key) + + +@dataclasses.dataclass(frozen=True) +class BoundAxisSet: + """An :class:`AxisSet` with sized axes, exposing a concrete shape.""" + + name: str + bound_axes: tuple[BoundAxis, ...] = dataclasses.field(default_factory=tuple) + axis_set: AxisSet = dataclasses.field(init=False) + + def __post_init__(self) -> None: + object.__setattr__(self, "axis_set", AxisSet(name=self.name, axes=tuple(ba.axis for ba in self.bound_axes))) + + @staticmethod + def from_1d_size(name: str, size: int, *, label: str = "x", unit: str | None = None) -> BoundAxisSet: + """Create a 1D bound axis set with a single affine-calibrated axis.""" + calibration = CalibrationSet.from_calibration(AffineCalibration(unit=unit or "")) + return BoundAxisSet( + name=name, + bound_axes=(BoundAxis(Axis(label, calibration), size),), + ) + + @staticmethod + def from_2d_size(name: str, size: tuple[int, int], *, labels: tuple[str, str] = ("x", "y"), unit: str | None = None) -> BoundAxisSet: + """Create a 2D bound axis set with identical affine calibration units on both axes.""" + size_x, size_y = size + x_label, y_label = labels + calibration = CalibrationSet.from_calibration(AffineCalibration(unit=unit or ""), key=DEFAULT_CALIBRATION_KEY) + return BoundAxisSet( + name=name, + bound_axes=( + BoundAxis(Axis(x_label, calibration), size_x), + BoundAxis(Axis(y_label, calibration), size_y), + ), + ) + + @property + def shape(self) -> tuple[int, ...]: + return tuple(bound_axis.size for bound_axis in self.bound_axes) + + @property + def rank(self) -> int: + return len(self.bound_axes) + + @property + def units(self) -> list[str]: + return [ax.axis.unit for ax in self.bound_axes] + + def get_calibration(self, axis: int, key: str | None = None) -> Calibration: + return self.bound_axes[axis].axis.get_calibration(key) + + +@dataclasses.dataclass +class DataDescriptor: + """Metadata for an array: axis sets, intensity calibrations, and properties.""" + + bound_axis_sets: list[BoundAxisSet] = dataclasses.field(default_factory=list) + intensity_calibrations: CalibrationSet = dataclasses.field(default_factory=CalibrationSet) + properties: dict[str, typing.Any] = dataclasses.field(default_factory=dict) + timestamp: datetime.datetime = dataclasses.field(default_factory=lambda: datetime.datetime.now(tz=tzlocal.get_localzone())) + + def __post_init__(self) -> None: + if self.timestamp.tzinfo is None: + raise ValueError("timestamp must be timezone-aware") + + def get_intensity_calibration(self, key: str | None = None) -> Calibration: + return self.intensity_calibrations.get(key) + + def with_primary_intensity_calibration(self, key: str) -> DataDescriptor: + return dataclasses.replace(self, intensity_calibrations=self.intensity_calibrations.with_primary(key)) + + def with_intensity_calibration(self, key: str, calibration: Calibration, *, make_primary: bool = False) -> DataDescriptor: + return dataclasses.replace( + self, + intensity_calibrations=self.intensity_calibrations.with_calibration(key, calibration, make_primary=make_primary), + ) + + @property + def intensity_calibration_keys(self) -> tuple[str, ...]: + return self.intensity_calibrations.calibration_keys + + @property + def primary_intensity_calibration_key(self) -> str: + return self.intensity_calibrations.primary_key + + @property + def timezone(self) -> str | None: + return self.timestamp.tzinfo.tzname(self.timestamp) if self.timestamp.tzinfo else "UTC" + + @property + def timezone_offset(self) -> str | None: + return self.timestamp.strftime("%z") if self.timestamp.tzinfo else "+0000" + + +@dataclasses.dataclass +class AnnotatedArray: + """A numpy array paired with a :class:`DataDescriptor` of its axes.""" + + data: numpy.typing.NDArray[typing.Any] + descriptor: DataDescriptor = dataclasses.field(default_factory=DataDescriptor) + + def __post_init__(self) -> None: + total_axes = sum(bound_axis_set.rank for bound_axis_set in self.descriptor.bound_axis_sets) + if self.descriptor.bound_axis_sets and total_axes != self.data.ndim: + raise ValueError(f"Axis sets account for {total_axes} axes but array has {self.data.ndim}") + + def get_intensity_calibration(self, key: str | None = None) -> Calibration: + return self.descriptor.get_intensity_calibration(key) + + def get_flat_calibrations(self, key: str | None = None) -> list[Calibration]: + return [bound_axis_set.get_calibration(axis=i, key=key) for bound_axis_set in self.descriptor.bound_axis_sets for i in range(bound_axis_set.rank)] + + +def zeros_annotated_array(bound_axis_sets: typing.Sequence[BoundAxisSet], dtype: numpy.typing.DTypeLike = numpy.float64) -> AnnotatedArray: + shape = tuple(dim for bound_axis_set in bound_axis_sets for dim in bound_axis_set.shape) + descriptor = DataDescriptor(bound_axis_sets=list(bound_axis_sets)) + return AnnotatedArray(data=numpy.zeros(shape, dtype=dtype), descriptor=descriptor) From d32336266d0fac52e421e54784f8e053386abde6 Mon Sep 17 00:00:00 2001 From: Chris Meyer <34664+cmeyer@users.noreply.github.com> Date: Thu, 16 Jul 2026 15:11:40 -0700 Subject: [PATCH 02/16] WIP: update class names; eliminate identifier from axis groups, ordered only now. --- nion/data/AnnotatedArray.py | 109 +++++++++++++++++++++++++++--------- 1 file changed, 81 insertions(+), 28 deletions(-) diff --git a/nion/data/AnnotatedArray.py b/nion/data/AnnotatedArray.py index 0b3de20..32e4b85 100644 --- a/nion/data/AnnotatedArray.py +++ b/nion/data/AnnotatedArray.py @@ -1,7 +1,7 @@ """Annotated n-dimensional arrays with calibrated, labeled axes. This module pairs a numpy array with a :class:`DataDescriptor` describing its -axes (grouped into :class:`AxisSet`/:class:`BoundAxisSet`), per-axis and +axes (grouped into :class:`AxisGroup`/:class:`BoundAxisGroup`), per-axis and intensity :class:`Calibration` mappings between indices and physical coordinates, and associated metadata. """ @@ -18,9 +18,9 @@ DEFAULT_CALIBRATION_KEY = "default" +DEFAULT_COORDINATE_MAPPING_KEY = "default" -@typing.runtime_checkable class Calibration(typing.Protocol): """Protocol converting between array indices and physical coordinates.""" @@ -99,6 +99,19 @@ def with_calibration(self, key: str, calibration: Calibration, *, make_primary: return CalibrationSet(calibrations=calibrations, primary_key=primary_key) +@dataclasses.dataclass(frozen=True) +class CoordinateSystemTransform: + """Invertible affine transform between two coordinate systems.""" + + source_coordinate_system_id: str + destination_coordinate_system_id: str + matrix: numpy.typing.NDArray[numpy.float64] + + def __post_init__(self) -> None: + if self.matrix.ndim != 2 or self.matrix.shape[0] != self.matrix.shape[1]: + raise ValueError("matrix must be a square 2D affine transform matrix") + + @dataclasses.dataclass(frozen=True) class Axis: """A named dimension with a primary calibration and optional auxiliaries.""" @@ -148,10 +161,9 @@ def __post_init__(self) -> None: @dataclasses.dataclass(frozen=True) -class AxisSet: - """A named, ordered group of axes (e.g. spatial or spectral).""" +class AxisGroup: + """An ordered group of axes (e.g. spatial or spectral).""" - name: str axes: tuple[Axis, ...] = dataclasses.field(default_factory=tuple) @property @@ -167,37 +179,58 @@ def get_calibration(self, axis: int, key: str | None = None) -> Calibration: @dataclasses.dataclass(frozen=True) -class BoundAxisSet: - """An :class:`AxisSet` with sized axes, exposing a concrete shape.""" +class BoundAxisGroup: + """An :class:`AxisGroup` with sized axes, exposing a concrete shape.""" - name: str bound_axes: tuple[BoundAxis, ...] = dataclasses.field(default_factory=tuple) - axis_set: AxisSet = dataclasses.field(init=False) + coordinate_system_id: str | None = None + coordinate_mappings: typing.Mapping[str, tuple[Calibration, ...]] = dataclasses.field(default_factory=dict) + primary_mapping_key: str | None = None + axis_group: AxisGroup = dataclasses.field(init=False) def __post_init__(self) -> None: - object.__setattr__(self, "axis_set", AxisSet(name=self.name, axes=tuple(ba.axis for ba in self.bound_axes))) + object.__setattr__(self, "axis_group", AxisGroup(axes=tuple(ba.axis for ba in self.bound_axes))) + coordinate_mappings = dict(self.coordinate_mappings) + normalized_coordinate_mappings: dict[str, tuple[Calibration, ...]] = dict() + for key, mapping in coordinate_mappings.items(): + normalized_mapping = tuple(mapping) + if len(normalized_mapping) != self.rank: + raise ValueError(f"coordinate mapping {key!r} rank {len(normalized_mapping)} does not match axis group rank {self.rank}") + normalized_coordinate_mappings[key] = normalized_mapping + object.__setattr__(self, "coordinate_mappings", types.MappingProxyType(normalized_coordinate_mappings)) + + if self.primary_mapping_key is not None and self.primary_mapping_key not in normalized_coordinate_mappings: + raise ValueError(f"primary_mapping_key {self.primary_mapping_key!r} is not present in coordinate_mappings") + if not normalized_coordinate_mappings and self.primary_mapping_key is not None: + raise ValueError("primary_mapping_key must be None when coordinate_mappings is empty") @staticmethod - def from_1d_size(name: str, size: int, *, label: str = "x", unit: str | None = None) -> BoundAxisSet: - """Create a 1D bound axis set with a single affine-calibrated axis.""" + def from_1d_size(size: int, *, label: str = "x", unit: str | None = None) -> BoundAxisGroup: + """Create a 1D bound axis group with one default coordinate mapping.""" calibration = CalibrationSet.from_calibration(AffineCalibration(unit=unit or "")) - return BoundAxisSet( - name=name, + return BoundAxisGroup( bound_axes=(BoundAxis(Axis(label, calibration), size),), + coordinate_mappings={ + DEFAULT_COORDINATE_MAPPING_KEY: (calibration.primary,), + }, + primary_mapping_key=DEFAULT_COORDINATE_MAPPING_KEY, ) @staticmethod - def from_2d_size(name: str, size: tuple[int, int], *, labels: tuple[str, str] = ("x", "y"), unit: str | None = None) -> BoundAxisSet: - """Create a 2D bound axis set with identical affine calibration units on both axes.""" + def from_2d_size(size: tuple[int, int], *, labels: tuple[str, str] = ("x", "y"), unit: str | None = None) -> BoundAxisGroup: + """Create a 2D bound axis group with one default coordinate mapping.""" size_x, size_y = size x_label, y_label = labels calibration = CalibrationSet.from_calibration(AffineCalibration(unit=unit or ""), key=DEFAULT_CALIBRATION_KEY) - return BoundAxisSet( - name=name, + return BoundAxisGroup( bound_axes=( BoundAxis(Axis(x_label, calibration), size_x), BoundAxis(Axis(y_label, calibration), size_y), ), + coordinate_mappings={ + DEFAULT_COORDINATE_MAPPING_KEY: (calibration.primary, calibration.primary), + }, + primary_mapping_key=DEFAULT_COORDINATE_MAPPING_KEY, ) @property @@ -210,18 +243,34 @@ def rank(self) -> int: @property def units(self) -> list[str]: - return [ax.axis.unit for ax in self.bound_axes] + if self.primary_mapping_key is None: + return [] + return [calibration.unit for calibration in self.get_coordinate_mapping()] + + @property + def mapping_keys(self) -> tuple[str, ...]: + return tuple(self.coordinate_mappings.keys()) + + def get_coordinate_mapping(self, key: str | None = None) -> tuple[Calibration, ...]: + target_key = self.primary_mapping_key if key is None else key + if target_key is None: + raise KeyError("No primary coordinate mapping is designated") + mapping = self.coordinate_mappings.get(target_key) + if mapping is None: + raise KeyError(f"Unknown coordinate mapping {target_key!r}") + return mapping def get_calibration(self, axis: int, key: str | None = None) -> Calibration: - return self.bound_axes[axis].axis.get_calibration(key) + return self.get_coordinate_mapping(key)[axis] @dataclasses.dataclass class DataDescriptor: """Metadata for an array: axis sets, intensity calibrations, and properties.""" - bound_axis_sets: list[BoundAxisSet] = dataclasses.field(default_factory=list) + bound_axis_groups: list[BoundAxisGroup] = dataclasses.field(default_factory=list) intensity_calibrations: CalibrationSet = dataclasses.field(default_factory=CalibrationSet) + coordinate_system_transforms: list[CoordinateSystemTransform] = dataclasses.field(default_factory=list) properties: dict[str, typing.Any] = dataclasses.field(default_factory=dict) timestamp: datetime.datetime = dataclasses.field(default_factory=lambda: datetime.datetime.now(tz=tzlocal.get_localzone())) @@ -266,18 +315,22 @@ class AnnotatedArray: descriptor: DataDescriptor = dataclasses.field(default_factory=DataDescriptor) def __post_init__(self) -> None: - total_axes = sum(bound_axis_set.rank for bound_axis_set in self.descriptor.bound_axis_sets) - if self.descriptor.bound_axis_sets and total_axes != self.data.ndim: - raise ValueError(f"Axis sets account for {total_axes} axes but array has {self.data.ndim}") + if not self.descriptor.bound_axis_groups: + raise ValueError("AnnotatedArray requires at least one axis group (bound_axis_group)") + if any(bound_axis_group.rank == 0 for bound_axis_group in self.descriptor.bound_axis_groups[:-1]): + raise ValueError("Only the final axis group may have rank 0") + total_axes = sum(bound_axis_group.rank for bound_axis_group in self.descriptor.bound_axis_groups) + if total_axes != self.data.ndim: + raise ValueError(f"Axis groups account for {total_axes} axes but array has {self.data.ndim}") def get_intensity_calibration(self, key: str | None = None) -> Calibration: return self.descriptor.get_intensity_calibration(key) def get_flat_calibrations(self, key: str | None = None) -> list[Calibration]: - return [bound_axis_set.get_calibration(axis=i, key=key) for bound_axis_set in self.descriptor.bound_axis_sets for i in range(bound_axis_set.rank)] + return [bound_axis_group.get_calibration(axis=i, key=key) for bound_axis_group in self.descriptor.bound_axis_groups for i in range(bound_axis_group.rank)] -def zeros_annotated_array(bound_axis_sets: typing.Sequence[BoundAxisSet], dtype: numpy.typing.DTypeLike = numpy.float64) -> AnnotatedArray: - shape = tuple(dim for bound_axis_set in bound_axis_sets for dim in bound_axis_set.shape) - descriptor = DataDescriptor(bound_axis_sets=list(bound_axis_sets)) +def zeros_annotated_array(bound_axis_groups: typing.Sequence[BoundAxisGroup], dtype: numpy.typing.DTypeLike = numpy.float64) -> AnnotatedArray: + shape = tuple(dim for bound_axis_group in bound_axis_groups for dim in bound_axis_group.shape) + descriptor = DataDescriptor(bound_axis_groups=list(bound_axis_groups)) return AnnotatedArray(data=numpy.zeros(shape, dtype=dtype), descriptor=descriptor) From c5c4dd2b7f981e55f926137fff503f95e9ced2cf Mon Sep 17 00:00:00 2001 From: Chris Meyer <34664+cmeyer@users.noreply.github.com> Date: Thu, 16 Jul 2026 16:22:43 -0700 Subject: [PATCH 03/16] WIP: Remove intrinsic coordinate transforms. --- nion/data/AnnotatedArray.py | 14 -------------- 1 file changed, 14 deletions(-) diff --git a/nion/data/AnnotatedArray.py b/nion/data/AnnotatedArray.py index 32e4b85..22b22e9 100644 --- a/nion/data/AnnotatedArray.py +++ b/nion/data/AnnotatedArray.py @@ -99,19 +99,6 @@ def with_calibration(self, key: str, calibration: Calibration, *, make_primary: return CalibrationSet(calibrations=calibrations, primary_key=primary_key) -@dataclasses.dataclass(frozen=True) -class CoordinateSystemTransform: - """Invertible affine transform between two coordinate systems.""" - - source_coordinate_system_id: str - destination_coordinate_system_id: str - matrix: numpy.typing.NDArray[numpy.float64] - - def __post_init__(self) -> None: - if self.matrix.ndim != 2 or self.matrix.shape[0] != self.matrix.shape[1]: - raise ValueError("matrix must be a square 2D affine transform matrix") - - @dataclasses.dataclass(frozen=True) class Axis: """A named dimension with a primary calibration and optional auxiliaries.""" @@ -270,7 +257,6 @@ class DataDescriptor: bound_axis_groups: list[BoundAxisGroup] = dataclasses.field(default_factory=list) intensity_calibrations: CalibrationSet = dataclasses.field(default_factory=CalibrationSet) - coordinate_system_transforms: list[CoordinateSystemTransform] = dataclasses.field(default_factory=list) properties: dict[str, typing.Any] = dataclasses.field(default_factory=dict) timestamp: datetime.datetime = dataclasses.field(default_factory=lambda: datetime.datetime.now(tz=tzlocal.get_localzone())) From d93bc78a46139700095e7e7c69db415779caab27 Mon Sep 17 00:00:00 2001 From: Chris Meyer <34664+cmeyer@users.noreply.github.com> Date: Thu, 16 Jul 2026 17:33:58 -0700 Subject: [PATCH 04/16] WIP: separate into data, descriptor, metadata (free-form, extensions). --- nion/data/AnnotatedArray.py | 150 ++++++++++++++++++++++++++++++------ 1 file changed, 127 insertions(+), 23 deletions(-) diff --git a/nion/data/AnnotatedArray.py b/nion/data/AnnotatedArray.py index 22b22e9..7705804 100644 --- a/nion/data/AnnotatedArray.py +++ b/nion/data/AnnotatedArray.py @@ -1,9 +1,10 @@ """Annotated n-dimensional arrays with calibrated, labeled axes. -This module pairs a numpy array with a :class:`DataDescriptor` describing its -axes (grouped into :class:`AxisGroup`/:class:`BoundAxisGroup`), per-axis and -intensity :class:`Calibration` mappings between indices and physical -coordinates, and associated metadata. +This module pairs a numpy array with a structural :class:`DataDescriptor` and +contextual :class:`ArrayMetadata`. An :class:`ArrayHeader` packages those +objects with the storage data type when the buffer is passed separately. Axes +are grouped into :class:`AxisGroup`/:class:`BoundAxisGroup` objects with per-axis +and intensity :class:`Calibration` mappings. """ from __future__ import annotations @@ -251,18 +252,27 @@ def get_calibration(self, axis: int, key: str | None = None) -> Calibration: return self.get_coordinate_mapping(key)[axis] -@dataclasses.dataclass +@dataclasses.dataclass(frozen=True) class DataDescriptor: - """Metadata for an array: axis sets, intensity calibrations, and properties.""" + """Intrinsic structure and calibrations used to interpret an array.""" - bound_axis_groups: list[BoundAxisGroup] = dataclasses.field(default_factory=list) + bound_axis_groups: tuple[BoundAxisGroup, ...] intensity_calibrations: CalibrationSet = dataclasses.field(default_factory=CalibrationSet) - properties: dict[str, typing.Any] = dataclasses.field(default_factory=dict) - timestamp: datetime.datetime = dataclasses.field(default_factory=lambda: datetime.datetime.now(tz=tzlocal.get_localzone())) def __post_init__(self) -> None: - if self.timestamp.tzinfo is None: - raise ValueError("timestamp must be timezone-aware") + object.__setattr__(self, "bound_axis_groups", tuple(self.bound_axis_groups)) + if not self.bound_axis_groups: + raise ValueError("DataDescriptor requires at least one bound axis group") + if any(bound_axis_group.rank == 0 for bound_axis_group in self.bound_axis_groups[:-1]): + raise ValueError("Only the final axis group may have rank 0") + + @property + def shape(self) -> tuple[int, ...]: + return tuple(dimension for bound_axis_group in self.bound_axis_groups for dimension in bound_axis_group.shape) + + @property + def rank(self) -> int: + return sum(bound_axis_group.rank for bound_axis_group in self.bound_axis_groups) def get_intensity_calibration(self, key: str | None = None) -> Calibration: return self.intensity_calibrations.get(key) @@ -284,6 +294,79 @@ def intensity_calibration_keys(self) -> tuple[str, ...]: def primary_intensity_calibration_key(self) -> str: return self.intensity_calibrations.primary_key + +@dataclasses.dataclass(frozen=True) +class StructuredExtension: + """A versioned, encoded payload whose semantics are defined outside this module.""" + + type_identifier: str + schema_version: int + payload: bytes + + def __post_init__(self) -> None: + if not self.type_identifier: + raise ValueError("StructuredExtension type_identifier must not be empty") + if self.schema_version < 1: + raise ValueError("StructuredExtension schema_version must be positive") + if not isinstance(self.payload, bytes): + raise TypeError("StructuredExtension payload must be bytes") + + +@dataclasses.dataclass(frozen=True, init=False) +class ArrayMetadata: + """Context carried with an array but not required to interpret its data.""" + + timestamp: datetime.datetime + free_form_metadata: typing.Mapping[str, typing.Any] + __structured_extensions: typing.Mapping[str, StructuredExtension] = dataclasses.field(repr=False) + + def __init__( + self, + *, + timestamp: datetime.datetime | None = None, + free_form_metadata: typing.Mapping[str, typing.Any] | None = None, + structured_extensions: typing.Iterable[StructuredExtension] = (), + ) -> None: + timestamp = timestamp or datetime.datetime.now(tz=tzlocal.get_localzone()) + if timestamp.tzinfo is None: + raise ValueError("timestamp must be timezone-aware") + extensions: dict[str, StructuredExtension] = dict() + for extension in structured_extensions: + if extension.type_identifier in extensions: + raise ValueError(f"Duplicate structured extension type {extension.type_identifier!r}") + extensions[extension.type_identifier] = extension + object.__setattr__(self, "timestamp", timestamp) + object.__setattr__(self, "free_form_metadata", types.MappingProxyType(dict(free_form_metadata or {}))) + object.__setattr__(self, "_ArrayMetadata__structured_extensions", types.MappingProxyType(extensions)) + + @property + def structured_extension_type_identifiers(self) -> tuple[str, ...]: + return tuple(self.__structured_extensions.keys()) + + def has_structured_extension(self, type_identifier: str) -> bool: + return type_identifier in self.__structured_extensions + + def get_structured_extension(self, type_identifier: str) -> StructuredExtension: + extension = self.__structured_extensions.get(type_identifier) + if extension is None: + raise KeyError(f"Unknown structured extension {type_identifier!r}") + return extension + + def with_structured_extension(self, extension: StructuredExtension) -> ArrayMetadata: + extensions = dict(self.__structured_extensions) + extensions[extension.type_identifier] = extension + return ArrayMetadata(timestamp=self.timestamp, free_form_metadata=self.free_form_metadata, structured_extensions=extensions.values()) + + def without_structured_extension(self, type_identifier: str) -> ArrayMetadata: + extensions = dict(self.__structured_extensions) + if type_identifier not in extensions: + raise KeyError(f"Unknown structured extension {type_identifier!r}") + del extensions[type_identifier] + return ArrayMetadata(timestamp=self.timestamp, free_form_metadata=self.free_form_metadata, structured_extensions=extensions.values()) + + def with_free_form_metadata(self, free_form_metadata: typing.Mapping[str, typing.Any]) -> ArrayMetadata: + return ArrayMetadata(timestamp=self.timestamp, free_form_metadata=free_form_metadata, structured_extensions=self.__structured_extensions.values()) + @property def timezone(self) -> str | None: return self.timestamp.tzinfo.tzname(self.timestamp) if self.timestamp.tzinfo else "UTC" @@ -293,21 +376,43 @@ def timezone_offset(self) -> str | None: return self.timestamp.strftime("%z") if self.timestamp.tzinfo else "+0000" +@dataclasses.dataclass(frozen=True) +class ArrayHeader: + """Everything needed to describe an array without its underlying buffer.""" + + descriptor: DataDescriptor + dtype: numpy.typing.DTypeLike + metadata: ArrayMetadata = dataclasses.field(default_factory=ArrayMetadata) + + def __post_init__(self) -> None: + object.__setattr__(self, "dtype", numpy.dtype(self.dtype)) + + @property + def shape(self) -> tuple[int, ...]: + return self.descriptor.shape + + @dataclasses.dataclass class AnnotatedArray: - """A numpy array paired with a :class:`DataDescriptor` of its axes.""" + """A numpy array paired with its descriptor and contextual metadata.""" data: numpy.typing.NDArray[typing.Any] - descriptor: DataDescriptor = dataclasses.field(default_factory=DataDescriptor) + descriptor: DataDescriptor + metadata: ArrayMetadata = dataclasses.field(default_factory=ArrayMetadata) def __post_init__(self) -> None: - if not self.descriptor.bound_axis_groups: - raise ValueError("AnnotatedArray requires at least one axis group (bound_axis_group)") - if any(bound_axis_group.rank == 0 for bound_axis_group in self.descriptor.bound_axis_groups[:-1]): - raise ValueError("Only the final axis group may have rank 0") - total_axes = sum(bound_axis_group.rank for bound_axis_group in self.descriptor.bound_axis_groups) - if total_axes != self.data.ndim: - raise ValueError(f"Axis groups account for {total_axes} axes but array has {self.data.ndim}") + if self.descriptor.shape != self.data.shape: + raise ValueError(f"Descriptor shape {self.descriptor.shape} does not match array shape {self.data.shape}") + + @property + def header(self) -> ArrayHeader: + return ArrayHeader(self.descriptor, self.data.dtype, self.metadata) + + @classmethod + def from_header(cls, data: numpy.typing.NDArray[typing.Any], header: ArrayHeader) -> AnnotatedArray: + if header.dtype != data.dtype: + raise ValueError(f"Header dtype {header.dtype} does not match array dtype {data.dtype}") + return cls(data, header.descriptor, header.metadata) def get_intensity_calibration(self, key: str | None = None) -> Calibration: return self.descriptor.get_intensity_calibration(key) @@ -317,6 +422,5 @@ def get_flat_calibrations(self, key: str | None = None) -> list[Calibration]: def zeros_annotated_array(bound_axis_groups: typing.Sequence[BoundAxisGroup], dtype: numpy.typing.DTypeLike = numpy.float64) -> AnnotatedArray: - shape = tuple(dim for bound_axis_group in bound_axis_groups for dim in bound_axis_group.shape) - descriptor = DataDescriptor(bound_axis_groups=list(bound_axis_groups)) - return AnnotatedArray(data=numpy.zeros(shape, dtype=dtype), descriptor=descriptor) + descriptor = DataDescriptor(bound_axis_groups=tuple(bound_axis_groups)) + return AnnotatedArray(data=numpy.zeros(descriptor.shape, dtype=dtype), descriptor=descriptor) From 46e8937f4ed5f3c47084fa9a03cb6fab6fb3a91b Mon Sep 17 00:00:00 2001 From: Chris Meyer <34664+cmeyer@users.noreply.github.com> Date: Thu, 16 Jul 2026 17:34:16 -0700 Subject: [PATCH 05/16] WIP: add annotated array tests. --- nion/data/test/AnnotatedArray_test.py | 107 ++++++++++++++++++++++++++ 1 file changed, 107 insertions(+) create mode 100644 nion/data/test/AnnotatedArray_test.py diff --git a/nion/data/test/AnnotatedArray_test.py b/nion/data/test/AnnotatedArray_test.py new file mode 100644 index 0000000..429b8bb --- /dev/null +++ b/nion/data/test/AnnotatedArray_test.py @@ -0,0 +1,107 @@ +import datetime +import unittest + +import numpy + +from nion.data import AnnotatedArray + + +class TestAnnotatedArray(unittest.TestCase): + + def test_data_descriptor_describes_shape_and_rank(self) -> None: + collection = AnnotatedArray.BoundAxisGroup.from_1d_size(3) + signal = AnnotatedArray.BoundAxisGroup.from_2d_size((4, 5)) + descriptor = AnnotatedArray.DataDescriptor((collection, signal)) + + self.assertEqual((3, 4, 5), descriptor.shape) + self.assertEqual(3, descriptor.rank) + + def test_data_descriptor_requires_valid_axis_group_layout(self) -> None: + with self.assertRaisesRegex(ValueError, "at least one"): + AnnotatedArray.DataDescriptor(()) + + scalar_group = AnnotatedArray.BoundAxisGroup() + vector_group = AnnotatedArray.BoundAxisGroup.from_1d_size(3) + with self.assertRaisesRegex(ValueError, "Only the final"): + AnnotatedArray.DataDescriptor((scalar_group, vector_group)) + + def test_array_metadata_controls_structured_extension_access(self) -> None: + extension = AnnotatedArray.StructuredExtension("org.nion.test", 1, b"value=42") + metadata = AnnotatedArray.ArrayMetadata(structured_extensions=(extension,)) + + self.assertEqual(("org.nion.test",), metadata.structured_extension_type_identifiers) + self.assertTrue(metadata.has_structured_extension("org.nion.test")) + self.assertIs(extension, metadata.get_structured_extension("org.nion.test")) + with self.assertRaises(KeyError): + metadata.get_structured_extension("org.nion.missing") + + replacement = AnnotatedArray.StructuredExtension("org.nion.test", 2, b"value=43") + replaced_metadata = metadata.with_structured_extension(replacement) + self.assertEqual(2, replaced_metadata.get_structured_extension("org.nion.test").schema_version) + self.assertEqual(1, metadata.get_structured_extension("org.nion.test").schema_version) + self.assertFalse(replaced_metadata.without_structured_extension("org.nion.test").has_structured_extension("org.nion.test")) + + def test_array_metadata_validates_extensions(self) -> None: + with self.assertRaisesRegex(ValueError, "must not be empty"): + AnnotatedArray.StructuredExtension("", 1, b"") + with self.assertRaisesRegex(ValueError, "must be positive"): + AnnotatedArray.StructuredExtension("org.nion.test", 0, b"") + with self.assertRaisesRegex(TypeError, "must be bytes"): + AnnotatedArray.StructuredExtension("org.nion.test", 1, bytearray()) # type: ignore[arg-type] + + extension = AnnotatedArray.StructuredExtension("org.nion.test", 1, b"") + with self.assertRaisesRegex(ValueError, "Duplicate"): + AnnotatedArray.ArrayMetadata(structured_extensions=(extension, extension)) + + def test_array_metadata_snapshots_free_form_metadata_and_requires_timezone(self) -> None: + source = {"note": "original"} + timestamp = datetime.datetime(2026, 7, 16, tzinfo=datetime.timezone.utc) + metadata = AnnotatedArray.ArrayMetadata(timestamp=timestamp, free_form_metadata=source) + source["note"] = "changed" + + self.assertEqual("original", metadata.free_form_metadata["note"]) + self.assertEqual("UTC", metadata.timezone) + self.assertEqual("+0000", metadata.timezone_offset) + with self.assertRaises(TypeError): + metadata.free_form_metadata["note"] = "changed" # type: ignore[index] + with self.assertRaisesRegex(ValueError, "timezone-aware"): + AnnotatedArray.ArrayMetadata(timestamp=datetime.datetime(2026, 7, 16)) + + def test_array_header_can_be_passed_without_data(self) -> None: + descriptor = AnnotatedArray.DataDescriptor((AnnotatedArray.BoundAxisGroup.from_2d_size((4, 5)),)) + metadata = AnnotatedArray.ArrayMetadata(free_form_metadata={"note": "test"}) + header = AnnotatedArray.ArrayHeader(descriptor, "float32", metadata) + + self.assertEqual((4, 5), header.shape) + self.assertEqual(numpy.dtype(numpy.float32), header.dtype) + + first = AnnotatedArray.AnnotatedArray.from_header(numpy.zeros(header.shape, dtype=header.dtype), header) + second = AnnotatedArray.AnnotatedArray.from_header(numpy.ones(header.shape, dtype=header.dtype), header) + self.assertEqual(header, first.header) + self.assertIsNot(header, first.header) + self.assertEqual(first.header, second.header) + self.assertIs(descriptor, first.descriptor) + self.assertIs(metadata, first.metadata) + + def test_annotated_array_validates_data_against_header(self) -> None: + descriptor = AnnotatedArray.DataDescriptor((AnnotatedArray.BoundAxisGroup.from_1d_size(3),)) + header = AnnotatedArray.ArrayHeader(descriptor, numpy.float32) + + with self.assertRaisesRegex(ValueError, "shape"): + AnnotatedArray.AnnotatedArray(numpy.zeros((4,), dtype=numpy.float32), descriptor) + with self.assertRaisesRegex(ValueError, "dtype"): + AnnotatedArray.AnnotatedArray.from_header(numpy.zeros((3,), dtype=numpy.float64), header) + + def test_zeros_annotated_array_constructs_matching_header(self) -> None: + group = AnnotatedArray.BoundAxisGroup.from_2d_size((2, 3), unit="nm") + array = AnnotatedArray.zeros_annotated_array((group,), dtype=numpy.float32) + + self.assertEqual((2, 3), array.data.shape) + self.assertEqual(numpy.dtype(numpy.float32), array.header.dtype) + self.assertEqual((group,), array.descriptor.bound_axis_groups) + + +if __name__ == "__main__": + unittest.main() + + From c5bb240d037fa46637ae1e3c1047b0a45089aa12 Mon Sep 17 00:00:00 2001 From: Chris Meyer <34664+cmeyer@users.noreply.github.com> Date: Thu, 16 Jul 2026 17:55:45 -0700 Subject: [PATCH 06/16] WIP: improve class/field names. --- nion/data/AnnotatedArray.py | 104 +++++++++++++------------- nion/data/test/AnnotatedArray_test.py | 64 ++++++++-------- 2 files changed, 85 insertions(+), 83 deletions(-) diff --git a/nion/data/AnnotatedArray.py b/nion/data/AnnotatedArray.py index 7705804..a1b8d75 100644 --- a/nion/data/AnnotatedArray.py +++ b/nion/data/AnnotatedArray.py @@ -1,6 +1,6 @@ """Annotated n-dimensional arrays with calibrated, labeled axes. -This module pairs a numpy array with a structural :class:`DataDescriptor` and +This module pairs a numpy array with a structural :class:`ArrayDescriptor` and contextual :class:`ArrayMetadata`. An :class:`ArrayHeader` packages those objects with the storage data type when the buffer is passed separately. Axes are grouped into :class:`AxisGroup`/:class:`BoundAxisGroup` objects with per-axis @@ -253,7 +253,7 @@ def get_calibration(self, axis: int, key: str | None = None) -> Calibration: @dataclasses.dataclass(frozen=True) -class DataDescriptor: +class ArrayDescriptor: """Intrinsic structure and calibrations used to interpret an array.""" bound_axis_groups: tuple[BoundAxisGroup, ...] @@ -262,7 +262,7 @@ class DataDescriptor: def __post_init__(self) -> None: object.__setattr__(self, "bound_axis_groups", tuple(self.bound_axis_groups)) if not self.bound_axis_groups: - raise ValueError("DataDescriptor requires at least one bound axis group") + raise ValueError("ArrayDescriptor requires at least one bound axis group") if any(bound_axis_group.rank == 0 for bound_axis_group in self.bound_axis_groups[:-1]): raise ValueError("Only the final axis group may have rank 0") @@ -277,10 +277,10 @@ def rank(self) -> int: def get_intensity_calibration(self, key: str | None = None) -> Calibration: return self.intensity_calibrations.get(key) - def with_primary_intensity_calibration(self, key: str) -> DataDescriptor: + def with_primary_intensity_calibration(self, key: str) -> ArrayDescriptor: return dataclasses.replace(self, intensity_calibrations=self.intensity_calibrations.with_primary(key)) - def with_intensity_calibration(self, key: str, calibration: Calibration, *, make_primary: bool = False) -> DataDescriptor: + def with_intensity_calibration(self, key: str, calibration: Calibration, *, make_primary: bool = False) -> ArrayDescriptor: return dataclasses.replace( self, intensity_calibrations=self.intensity_calibrations.with_calibration(key, calibration, make_primary=make_primary), @@ -296,91 +296,91 @@ def primary_intensity_calibration_key(self) -> str: @dataclasses.dataclass(frozen=True) -class StructuredExtension: +class ExtensionRecord: """A versioned, encoded payload whose semantics are defined outside this module.""" - type_identifier: str + extension_type_id: str schema_version: int payload: bytes def __post_init__(self) -> None: - if not self.type_identifier: - raise ValueError("StructuredExtension type_identifier must not be empty") + if not self.extension_type_id: + raise ValueError("ExtensionRecord extension_type_id must not be empty") if self.schema_version < 1: - raise ValueError("StructuredExtension schema_version must be positive") + raise ValueError("ExtensionRecord schema_version must be positive") if not isinstance(self.payload, bytes): - raise TypeError("StructuredExtension payload must be bytes") + raise TypeError("ExtensionRecord payload must be bytes") @dataclasses.dataclass(frozen=True, init=False) class ArrayMetadata: """Context carried with an array but not required to interpret its data.""" - timestamp: datetime.datetime - free_form_metadata: typing.Mapping[str, typing.Any] - __structured_extensions: typing.Mapping[str, StructuredExtension] = dataclasses.field(repr=False) + created: datetime.datetime + attributes: typing.Mapping[str, typing.Any] + __extensions: typing.Mapping[str, ExtensionRecord] = dataclasses.field(repr=False) def __init__( self, *, - timestamp: datetime.datetime | None = None, - free_form_metadata: typing.Mapping[str, typing.Any] | None = None, - structured_extensions: typing.Iterable[StructuredExtension] = (), + created: datetime.datetime | None = None, + attributes: typing.Mapping[str, typing.Any] | None = None, + extensions: typing.Iterable[ExtensionRecord] = (), ) -> None: - timestamp = timestamp or datetime.datetime.now(tz=tzlocal.get_localzone()) - if timestamp.tzinfo is None: - raise ValueError("timestamp must be timezone-aware") - extensions: dict[str, StructuredExtension] = dict() - for extension in structured_extensions: - if extension.type_identifier in extensions: - raise ValueError(f"Duplicate structured extension type {extension.type_identifier!r}") - extensions[extension.type_identifier] = extension - object.__setattr__(self, "timestamp", timestamp) - object.__setattr__(self, "free_form_metadata", types.MappingProxyType(dict(free_form_metadata or {}))) - object.__setattr__(self, "_ArrayMetadata__structured_extensions", types.MappingProxyType(extensions)) + created = created or datetime.datetime.now(tz=tzlocal.get_localzone()) + if created.tzinfo is None: + raise ValueError("created must be timezone-aware") + extension_map: dict[str, ExtensionRecord] = dict() + for extension in extensions: + if extension.extension_type_id in extension_map: + raise ValueError(f"Duplicate extension type {extension.extension_type_id!r}") + extension_map[extension.extension_type_id] = extension + object.__setattr__(self, "created", created) + object.__setattr__(self, "attributes", types.MappingProxyType(dict(attributes or {}))) + object.__setattr__(self, "_ArrayMetadata__extensions", types.MappingProxyType(extension_map)) @property - def structured_extension_type_identifiers(self) -> tuple[str, ...]: - return tuple(self.__structured_extensions.keys()) + def extension_type_ids(self) -> tuple[str, ...]: + return tuple(self.__extensions.keys()) - def has_structured_extension(self, type_identifier: str) -> bool: - return type_identifier in self.__structured_extensions + def has_extension(self, extension_type_id: str) -> bool: + return extension_type_id in self.__extensions - def get_structured_extension(self, type_identifier: str) -> StructuredExtension: - extension = self.__structured_extensions.get(type_identifier) + def get_extension(self, extension_type_id: str) -> ExtensionRecord: + extension = self.__extensions.get(extension_type_id) if extension is None: - raise KeyError(f"Unknown structured extension {type_identifier!r}") + raise KeyError(f"Unknown extension {extension_type_id!r}") return extension - def with_structured_extension(self, extension: StructuredExtension) -> ArrayMetadata: - extensions = dict(self.__structured_extensions) - extensions[extension.type_identifier] = extension - return ArrayMetadata(timestamp=self.timestamp, free_form_metadata=self.free_form_metadata, structured_extensions=extensions.values()) + def with_extension(self, extension: ExtensionRecord) -> ArrayMetadata: + extensions = dict(self.__extensions) + extensions[extension.extension_type_id] = extension + return ArrayMetadata(created=self.created, attributes=self.attributes, extensions=extensions.values()) - def without_structured_extension(self, type_identifier: str) -> ArrayMetadata: - extensions = dict(self.__structured_extensions) - if type_identifier not in extensions: - raise KeyError(f"Unknown structured extension {type_identifier!r}") - del extensions[type_identifier] - return ArrayMetadata(timestamp=self.timestamp, free_form_metadata=self.free_form_metadata, structured_extensions=extensions.values()) + def without_extension(self, extension_type_id: str) -> ArrayMetadata: + extensions = dict(self.__extensions) + if extension_type_id not in extensions: + raise KeyError(f"Unknown extension {extension_type_id!r}") + del extensions[extension_type_id] + return ArrayMetadata(created=self.created, attributes=self.attributes, extensions=extensions.values()) - def with_free_form_metadata(self, free_form_metadata: typing.Mapping[str, typing.Any]) -> ArrayMetadata: - return ArrayMetadata(timestamp=self.timestamp, free_form_metadata=free_form_metadata, structured_extensions=self.__structured_extensions.values()) + def with_attributes(self, attributes: typing.Mapping[str, typing.Any]) -> ArrayMetadata: + return ArrayMetadata(created=self.created, attributes=attributes, extensions=self.__extensions.values()) @property def timezone(self) -> str | None: - return self.timestamp.tzinfo.tzname(self.timestamp) if self.timestamp.tzinfo else "UTC" + return self.created.tzinfo.tzname(self.created) if self.created.tzinfo else "UTC" @property def timezone_offset(self) -> str | None: - return self.timestamp.strftime("%z") if self.timestamp.tzinfo else "+0000" + return self.created.strftime("%z") if self.created.tzinfo else "+0000" @dataclasses.dataclass(frozen=True) class ArrayHeader: """Everything needed to describe an array without its underlying buffer.""" - descriptor: DataDescriptor + descriptor: ArrayDescriptor dtype: numpy.typing.DTypeLike metadata: ArrayMetadata = dataclasses.field(default_factory=ArrayMetadata) @@ -397,7 +397,7 @@ class AnnotatedArray: """A numpy array paired with its descriptor and contextual metadata.""" data: numpy.typing.NDArray[typing.Any] - descriptor: DataDescriptor + descriptor: ArrayDescriptor metadata: ArrayMetadata = dataclasses.field(default_factory=ArrayMetadata) def __post_init__(self) -> None: @@ -422,5 +422,5 @@ def get_flat_calibrations(self, key: str | None = None) -> list[Calibration]: def zeros_annotated_array(bound_axis_groups: typing.Sequence[BoundAxisGroup], dtype: numpy.typing.DTypeLike = numpy.float64) -> AnnotatedArray: - descriptor = DataDescriptor(bound_axis_groups=tuple(bound_axis_groups)) + descriptor = ArrayDescriptor(bound_axis_groups=tuple(bound_axis_groups)) return AnnotatedArray(data=numpy.zeros(descriptor.shape, dtype=dtype), descriptor=descriptor) diff --git a/nion/data/test/AnnotatedArray_test.py b/nion/data/test/AnnotatedArray_test.py index 429b8bb..6c69569 100644 --- a/nion/data/test/AnnotatedArray_test.py +++ b/nion/data/test/AnnotatedArray_test.py @@ -8,68 +8,70 @@ class TestAnnotatedArray(unittest.TestCase): - def test_data_descriptor_describes_shape_and_rank(self) -> None: + def test_array_descriptor_describes_shape_and_rank(self) -> None: collection = AnnotatedArray.BoundAxisGroup.from_1d_size(3) signal = AnnotatedArray.BoundAxisGroup.from_2d_size((4, 5)) - descriptor = AnnotatedArray.DataDescriptor((collection, signal)) + descriptor = AnnotatedArray.ArrayDescriptor((collection, signal)) self.assertEqual((3, 4, 5), descriptor.shape) self.assertEqual(3, descriptor.rank) - def test_data_descriptor_requires_valid_axis_group_layout(self) -> None: + def test_array_descriptor_requires_valid_axis_group_layout(self) -> None: with self.assertRaisesRegex(ValueError, "at least one"): - AnnotatedArray.DataDescriptor(()) + AnnotatedArray.ArrayDescriptor(()) scalar_group = AnnotatedArray.BoundAxisGroup() vector_group = AnnotatedArray.BoundAxisGroup.from_1d_size(3) with self.assertRaisesRegex(ValueError, "Only the final"): - AnnotatedArray.DataDescriptor((scalar_group, vector_group)) + AnnotatedArray.ArrayDescriptor((scalar_group, vector_group)) - def test_array_metadata_controls_structured_extension_access(self) -> None: - extension = AnnotatedArray.StructuredExtension("org.nion.test", 1, b"value=42") - metadata = AnnotatedArray.ArrayMetadata(structured_extensions=(extension,)) + def test_array_metadata_controls_extension_access(self) -> None: + extension = AnnotatedArray.ExtensionRecord("org.nion.test", 1, b"value=42") + metadata = AnnotatedArray.ArrayMetadata(extensions=(extension,)) - self.assertEqual(("org.nion.test",), metadata.structured_extension_type_identifiers) - self.assertTrue(metadata.has_structured_extension("org.nion.test")) - self.assertIs(extension, metadata.get_structured_extension("org.nion.test")) + self.assertEqual("org.nion.test", extension.extension_type_id) + self.assertEqual(("org.nion.test",), metadata.extension_type_ids) + self.assertTrue(metadata.has_extension("org.nion.test")) + self.assertIs(extension, metadata.get_extension("org.nion.test")) with self.assertRaises(KeyError): - metadata.get_structured_extension("org.nion.missing") + metadata.get_extension("org.nion.missing") - replacement = AnnotatedArray.StructuredExtension("org.nion.test", 2, b"value=43") - replaced_metadata = metadata.with_structured_extension(replacement) - self.assertEqual(2, replaced_metadata.get_structured_extension("org.nion.test").schema_version) - self.assertEqual(1, metadata.get_structured_extension("org.nion.test").schema_version) - self.assertFalse(replaced_metadata.without_structured_extension("org.nion.test").has_structured_extension("org.nion.test")) + replacement = AnnotatedArray.ExtensionRecord("org.nion.test", 2, b"value=43") + replaced_metadata = metadata.with_extension(replacement) + self.assertEqual(2, replaced_metadata.get_extension("org.nion.test").schema_version) + self.assertEqual(1, metadata.get_extension("org.nion.test").schema_version) + self.assertFalse(replaced_metadata.without_extension("org.nion.test").has_extension("org.nion.test")) def test_array_metadata_validates_extensions(self) -> None: with self.assertRaisesRegex(ValueError, "must not be empty"): - AnnotatedArray.StructuredExtension("", 1, b"") + AnnotatedArray.ExtensionRecord("", 1, b"") with self.assertRaisesRegex(ValueError, "must be positive"): - AnnotatedArray.StructuredExtension("org.nion.test", 0, b"") + AnnotatedArray.ExtensionRecord("org.nion.test", 0, b"") with self.assertRaisesRegex(TypeError, "must be bytes"): - AnnotatedArray.StructuredExtension("org.nion.test", 1, bytearray()) # type: ignore[arg-type] + AnnotatedArray.ExtensionRecord("org.nion.test", 1, bytearray()) # type: ignore[arg-type] - extension = AnnotatedArray.StructuredExtension("org.nion.test", 1, b"") + extension = AnnotatedArray.ExtensionRecord("org.nion.test", 1, b"") with self.assertRaisesRegex(ValueError, "Duplicate"): - AnnotatedArray.ArrayMetadata(structured_extensions=(extension, extension)) + AnnotatedArray.ArrayMetadata(extensions=(extension, extension)) - def test_array_metadata_snapshots_free_form_metadata_and_requires_timezone(self) -> None: + def test_array_metadata_snapshots_attributes_and_requires_timezone(self) -> None: source = {"note": "original"} - timestamp = datetime.datetime(2026, 7, 16, tzinfo=datetime.timezone.utc) - metadata = AnnotatedArray.ArrayMetadata(timestamp=timestamp, free_form_metadata=source) + created = datetime.datetime(2026, 7, 16, tzinfo=datetime.timezone.utc) + metadata = AnnotatedArray.ArrayMetadata(created=created, attributes=source) source["note"] = "changed" - self.assertEqual("original", metadata.free_form_metadata["note"]) + self.assertEqual("original", metadata.attributes["note"]) + self.assertEqual("replacement", metadata.with_attributes({"note": "replacement"}).attributes["note"]) self.assertEqual("UTC", metadata.timezone) self.assertEqual("+0000", metadata.timezone_offset) with self.assertRaises(TypeError): - metadata.free_form_metadata["note"] = "changed" # type: ignore[index] + metadata.attributes["note"] = "changed" # type: ignore[index] with self.assertRaisesRegex(ValueError, "timezone-aware"): - AnnotatedArray.ArrayMetadata(timestamp=datetime.datetime(2026, 7, 16)) + AnnotatedArray.ArrayMetadata(created=datetime.datetime(2026, 7, 16)) def test_array_header_can_be_passed_without_data(self) -> None: - descriptor = AnnotatedArray.DataDescriptor((AnnotatedArray.BoundAxisGroup.from_2d_size((4, 5)),)) - metadata = AnnotatedArray.ArrayMetadata(free_form_metadata={"note": "test"}) + descriptor = AnnotatedArray.ArrayDescriptor((AnnotatedArray.BoundAxisGroup.from_2d_size((4, 5)),)) + metadata = AnnotatedArray.ArrayMetadata(attributes={"note": "test"}) header = AnnotatedArray.ArrayHeader(descriptor, "float32", metadata) self.assertEqual((4, 5), header.shape) @@ -84,7 +86,7 @@ def test_array_header_can_be_passed_without_data(self) -> None: self.assertIs(metadata, first.metadata) def test_annotated_array_validates_data_against_header(self) -> None: - descriptor = AnnotatedArray.DataDescriptor((AnnotatedArray.BoundAxisGroup.from_1d_size(3),)) + descriptor = AnnotatedArray.ArrayDescriptor((AnnotatedArray.BoundAxisGroup.from_1d_size(3),)) header = AnnotatedArray.ArrayHeader(descriptor, numpy.float32) with self.assertRaisesRegex(ValueError, "shape"): From af4207a7a642d10bb86974c9d065f98e171dca6b Mon Sep 17 00:00:00 2001 From: Chris Meyer <34664+cmeyer@users.noreply.github.com> Date: Thu, 16 Jul 2026 18:15:11 -0700 Subject: [PATCH 07/16] WIP: more renames; merge timezone into created. --- nion/data/AnnotatedArray.py | 39 +++++++++++---------------- nion/data/test/AnnotatedArray_test.py | 24 ++++++++++++++--- 2 files changed, 37 insertions(+), 26 deletions(-) diff --git a/nion/data/AnnotatedArray.py b/nion/data/AnnotatedArray.py index a1b8d75..b5ed0d3 100644 --- a/nion/data/AnnotatedArray.py +++ b/nion/data/AnnotatedArray.py @@ -71,13 +71,13 @@ def calibration_keys(self) -> tuple[str, ...]: return tuple(self.calibrations.keys()) @property - def primary(self) -> Calibration: + def primary_calibration(self) -> Calibration: return self.calibrations[self.primary_key] - def has(self, key: str) -> bool: + def has_calibration(self, key: str) -> bool: return key in self.calibrations - def get(self, key: str | None = None) -> Calibration: + def get_calibration(self, key: str | None = None) -> Calibration: target_key = self.primary_key if key is None else key calibration = self.calibrations.get(target_key) if calibration is None: @@ -88,7 +88,7 @@ def get(self, key: str | None = None) -> Calibration: def from_calibration(calibration: Calibration, key: str = DEFAULT_CALIBRATION_KEY) -> CalibrationSet: return CalibrationSet(calibrations={key: calibration}, primary_key=key) - def with_primary(self, key: str) -> CalibrationSet: + def with_primary_calibration(self, key: str) -> CalibrationSet: if key not in self.calibrations: raise KeyError(f"Unknown calibration {key!r}") return dataclasses.replace(self, primary_key=key) @@ -102,16 +102,16 @@ def with_calibration(self, key: str, calibration: Calibration, *, make_primary: @dataclasses.dataclass(frozen=True) class Axis: - """A named dimension with a primary calibration and optional auxiliaries.""" + """A dimension with a display label, primary calibration, and optional auxiliaries.""" label: str = "" calibrations: CalibrationSet = dataclasses.field(default_factory=CalibrationSet) def get_calibration(self, key: str | None = None) -> Calibration: - return self.calibrations.get(key) + return self.calibrations.get_calibration(key) def with_primary_calibration(self, key: str) -> Axis: - return dataclasses.replace(self, calibrations=self.calibrations.with_primary(key)) + return dataclasses.replace(self, calibrations=self.calibrations.with_primary_calibration(key)) def with_calibration(self, key: str, calibration: Calibration, *, make_primary: bool = False) -> Axis: return dataclasses.replace( @@ -129,7 +129,7 @@ def primary_calibration_key(self) -> str: @property def unit(self) -> str: - return self.calibrations.primary.unit + return self.calibrations.primary_calibration.unit @classmethod def from_calibration(cls, label: str, calibration: Calibration, key: str = DEFAULT_CALIBRATION_KEY) -> Axis: @@ -199,7 +199,7 @@ def from_1d_size(size: int, *, label: str = "x", unit: str | None = None) -> Bou return BoundAxisGroup( bound_axes=(BoundAxis(Axis(label, calibration), size),), coordinate_mappings={ - DEFAULT_COORDINATE_MAPPING_KEY: (calibration.primary,), + DEFAULT_COORDINATE_MAPPING_KEY: (calibration.primary_calibration,), }, primary_mapping_key=DEFAULT_COORDINATE_MAPPING_KEY, ) @@ -216,7 +216,7 @@ def from_2d_size(size: tuple[int, int], *, labels: tuple[str, str] = ("x", "y"), BoundAxis(Axis(y_label, calibration), size_y), ), coordinate_mappings={ - DEFAULT_COORDINATE_MAPPING_KEY: (calibration.primary, calibration.primary), + DEFAULT_COORDINATE_MAPPING_KEY: (calibration.primary_calibration, calibration.primary_calibration), }, primary_mapping_key=DEFAULT_COORDINATE_MAPPING_KEY, ) @@ -271,14 +271,14 @@ def shape(self) -> tuple[int, ...]: return tuple(dimension for bound_axis_group in self.bound_axis_groups for dimension in bound_axis_group.shape) @property - def rank(self) -> int: + def ndim(self) -> int: return sum(bound_axis_group.rank for bound_axis_group in self.bound_axis_groups) def get_intensity_calibration(self, key: str | None = None) -> Calibration: - return self.intensity_calibrations.get(key) + return self.intensity_calibrations.get_calibration(key) def with_primary_intensity_calibration(self, key: str) -> ArrayDescriptor: - return dataclasses.replace(self, intensity_calibrations=self.intensity_calibrations.with_primary(key)) + return dataclasses.replace(self, intensity_calibrations=self.intensity_calibrations.with_primary_calibration(key)) def with_intensity_calibration(self, key: str, calibration: Calibration, *, make_primary: bool = False) -> ArrayDescriptor: return dataclasses.replace( @@ -328,7 +328,7 @@ def __init__( extensions: typing.Iterable[ExtensionRecord] = (), ) -> None: created = created or datetime.datetime.now(tz=tzlocal.get_localzone()) - if created.tzinfo is None: + if created.tzinfo is None or created.utcoffset() is None: raise ValueError("created must be timezone-aware") extension_map: dict[str, ExtensionRecord] = dict() for extension in extensions: @@ -367,13 +367,6 @@ def without_extension(self, extension_type_id: str) -> ArrayMetadata: def with_attributes(self, attributes: typing.Mapping[str, typing.Any]) -> ArrayMetadata: return ArrayMetadata(created=self.created, attributes=attributes, extensions=self.__extensions.values()) - @property - def timezone(self) -> str | None: - return self.created.tzinfo.tzname(self.created) if self.created.tzinfo else "UTC" - - @property - def timezone_offset(self) -> str | None: - return self.created.strftime("%z") if self.created.tzinfo else "+0000" @dataclasses.dataclass(frozen=True) @@ -385,7 +378,7 @@ class ArrayHeader: metadata: ArrayMetadata = dataclasses.field(default_factory=ArrayMetadata) def __post_init__(self) -> None: - object.__setattr__(self, "dtype", numpy.dtype(self.dtype)) + object.__setattr__(self, "dtype", numpy.dtype(typing.cast(typing.Any, self.dtype))) @property def shape(self) -> tuple[int, ...]: @@ -417,7 +410,7 @@ def from_header(cls, data: numpy.typing.NDArray[typing.Any], header: ArrayHeader def get_intensity_calibration(self, key: str | None = None) -> Calibration: return self.descriptor.get_intensity_calibration(key) - def get_flat_calibrations(self, key: str | None = None) -> list[Calibration]: + def get_flat_axis_calibrations(self, key: str | None = None) -> list[Calibration]: return [bound_axis_group.get_calibration(axis=i, key=key) for bound_axis_group in self.descriptor.bound_axis_groups for i in range(bound_axis_group.rank)] diff --git a/nion/data/test/AnnotatedArray_test.py b/nion/data/test/AnnotatedArray_test.py index 6c69569..f8bb255 100644 --- a/nion/data/test/AnnotatedArray_test.py +++ b/nion/data/test/AnnotatedArray_test.py @@ -1,5 +1,6 @@ import datetime import unittest +import zoneinfo import numpy @@ -8,13 +9,23 @@ class TestAnnotatedArray(unittest.TestCase): + def test_calibration_set_uses_explicit_calibration_accessors(self) -> None: + primary = AnnotatedArray.AffineCalibration(unit="nm") + alternate = AnnotatedArray.AffineCalibration(unit="rad") + calibrations = AnnotatedArray.CalibrationSet.from_calibration(primary).with_calibration("alternate", alternate) + + self.assertTrue(calibrations.has_calibration("alternate")) + self.assertIs(primary, calibrations.primary_calibration) + self.assertIs(alternate, calibrations.get_calibration("alternate")) + self.assertIs(alternate, calibrations.with_primary_calibration("alternate").primary_calibration) + def test_array_descriptor_describes_shape_and_rank(self) -> None: collection = AnnotatedArray.BoundAxisGroup.from_1d_size(3) signal = AnnotatedArray.BoundAxisGroup.from_2d_size((4, 5)) descriptor = AnnotatedArray.ArrayDescriptor((collection, signal)) self.assertEqual((3, 4, 5), descriptor.shape) - self.assertEqual(3, descriptor.rank) + self.assertEqual(3, descriptor.ndim) def test_array_descriptor_requires_valid_axis_group_layout(self) -> None: with self.assertRaisesRegex(ValueError, "at least one"): @@ -62,13 +73,19 @@ def test_array_metadata_snapshots_attributes_and_requires_timezone(self) -> None self.assertEqual("original", metadata.attributes["note"]) self.assertEqual("replacement", metadata.with_attributes({"note": "replacement"}).attributes["note"]) - self.assertEqual("UTC", metadata.timezone) - self.assertEqual("+0000", metadata.timezone_offset) + self.assertEqual(datetime.timedelta(0), metadata.created.utcoffset()) with self.assertRaises(TypeError): metadata.attributes["note"] = "changed" # type: ignore[index] with self.assertRaisesRegex(ValueError, "timezone-aware"): AnnotatedArray.ArrayMetadata(created=datetime.datetime(2026, 7, 16)) + def test_array_metadata_created_retains_iana_zone_and_offset(self) -> None: + created = datetime.datetime(2026, 7, 16, 12, tzinfo=zoneinfo.ZoneInfo("America/Los_Angeles")) + metadata = AnnotatedArray.ArrayMetadata(created=created) + + self.assertEqual("America/Los_Angeles", getattr(metadata.created.tzinfo, "key", None)) + self.assertEqual(datetime.timedelta(hours=-7), metadata.created.utcoffset()) + def test_array_header_can_be_passed_without_data(self) -> None: descriptor = AnnotatedArray.ArrayDescriptor((AnnotatedArray.BoundAxisGroup.from_2d_size((4, 5)),)) metadata = AnnotatedArray.ArrayMetadata(attributes={"note": "test"}) @@ -101,6 +118,7 @@ def test_zeros_annotated_array_constructs_matching_header(self) -> None: self.assertEqual((2, 3), array.data.shape) self.assertEqual(numpy.dtype(numpy.float32), array.header.dtype) self.assertEqual((group,), array.descriptor.bound_axis_groups) + self.assertEqual(2, len(array.get_flat_axis_calibrations())) if __name__ == "__main__": From d1c7290414dbd597fbdaaa6905550c5ffa4c97ae Mon Sep 17 00:00:00 2001 From: Chris Meyer <34664+cmeyer@users.noreply.github.com> Date: Tue, 21 Jul 2026 11:57:34 -0700 Subject: [PATCH 08/16] WIP: clean up where calibrations are kept. --- nion/data/AnnotatedArray.py | 54 +++++++------------------------------ 1 file changed, 9 insertions(+), 45 deletions(-) diff --git a/nion/data/AnnotatedArray.py b/nion/data/AnnotatedArray.py index b5ed0d3..8810b24 100644 --- a/nion/data/AnnotatedArray.py +++ b/nion/data/AnnotatedArray.py @@ -102,38 +102,9 @@ def with_calibration(self, key: str, calibration: Calibration, *, make_primary: @dataclasses.dataclass(frozen=True) class Axis: - """A dimension with a display label, primary calibration, and optional auxiliaries.""" + """A dimension label and display metadata independent of coordinate mappings.""" label: str = "" - calibrations: CalibrationSet = dataclasses.field(default_factory=CalibrationSet) - - def get_calibration(self, key: str | None = None) -> Calibration: - return self.calibrations.get_calibration(key) - - def with_primary_calibration(self, key: str) -> Axis: - return dataclasses.replace(self, calibrations=self.calibrations.with_primary_calibration(key)) - - def with_calibration(self, key: str, calibration: Calibration, *, make_primary: bool = False) -> Axis: - return dataclasses.replace( - self, - calibrations=self.calibrations.with_calibration(key, calibration, make_primary=make_primary), - ) - - @property - def calibration_keys(self) -> tuple[str, ...]: - return self.calibrations.calibration_keys - - @property - def primary_calibration_key(self) -> str: - return self.calibrations.primary_key - - @property - def unit(self) -> str: - return self.calibrations.primary_calibration.unit - - @classmethod - def from_calibration(cls, label: str, calibration: Calibration, key: str = DEFAULT_CALIBRATION_KEY) -> Axis: - return cls(label=label, calibrations=CalibrationSet(calibrations={key: calibration}, primary_key=key)) @dataclasses.dataclass(frozen=True) @@ -150,7 +121,7 @@ def __post_init__(self) -> None: @dataclasses.dataclass(frozen=True) class AxisGroup: - """An ordered group of axes (e.g. spatial or spectral).""" + """An ordered group of axes (e.g. spatial or spectral) without calibrations.""" axes: tuple[Axis, ...] = dataclasses.field(default_factory=tuple) @@ -158,13 +129,6 @@ class AxisGroup: def rank(self) -> int: return len(self.axes) - @property - def units(self) -> list[str]: - return [ax.unit for ax in self.axes] - - def get_calibration(self, axis: int, key: str | None = None) -> Calibration: - return self.axes[axis].get_calibration(key) - @dataclasses.dataclass(frozen=True) class BoundAxisGroup: @@ -195,11 +159,11 @@ def __post_init__(self) -> None: @staticmethod def from_1d_size(size: int, *, label: str = "x", unit: str | None = None) -> BoundAxisGroup: """Create a 1D bound axis group with one default coordinate mapping.""" - calibration = CalibrationSet.from_calibration(AffineCalibration(unit=unit or "")) + calibration = AffineCalibration(unit=unit or "") return BoundAxisGroup( - bound_axes=(BoundAxis(Axis(label, calibration), size),), + bound_axes=(BoundAxis(Axis(label), size),), coordinate_mappings={ - DEFAULT_COORDINATE_MAPPING_KEY: (calibration.primary_calibration,), + DEFAULT_COORDINATE_MAPPING_KEY: (calibration,), }, primary_mapping_key=DEFAULT_COORDINATE_MAPPING_KEY, ) @@ -209,14 +173,14 @@ def from_2d_size(size: tuple[int, int], *, labels: tuple[str, str] = ("x", "y"), """Create a 2D bound axis group with one default coordinate mapping.""" size_x, size_y = size x_label, y_label = labels - calibration = CalibrationSet.from_calibration(AffineCalibration(unit=unit or ""), key=DEFAULT_CALIBRATION_KEY) + calibration = AffineCalibration(unit=unit or "") return BoundAxisGroup( bound_axes=( - BoundAxis(Axis(x_label, calibration), size_x), - BoundAxis(Axis(y_label, calibration), size_y), + BoundAxis(Axis(x_label), size_x), + BoundAxis(Axis(y_label), size_y), ), coordinate_mappings={ - DEFAULT_COORDINATE_MAPPING_KEY: (calibration.primary_calibration, calibration.primary_calibration), + DEFAULT_COORDINATE_MAPPING_KEY: (calibration, calibration), }, primary_mapping_key=DEFAULT_COORDINATE_MAPPING_KEY, ) From f579dc98b4ccfbff26f9ac440f691285caae43c2 Mon Sep 17 00:00:00 2001 From: Chris Meyer <34664+cmeyer@users.noreply.github.com> Date: Tue, 21 Jul 2026 18:52:51 -0700 Subject: [PATCH 09/16] WIP: merge bound and unbound axis stuff. --- nion/data/AnnotatedArray.py | 77 ++++++++++----------------- nion/data/test/AnnotatedArray_test.py | 16 +++--- 2 files changed, 37 insertions(+), 56 deletions(-) diff --git a/nion/data/AnnotatedArray.py b/nion/data/AnnotatedArray.py index 8810b24..5f956c3 100644 --- a/nion/data/AnnotatedArray.py +++ b/nion/data/AnnotatedArray.py @@ -3,8 +3,8 @@ This module pairs a numpy array with a structural :class:`ArrayDescriptor` and contextual :class:`ArrayMetadata`. An :class:`ArrayHeader` packages those objects with the storage data type when the buffer is passed separately. Axes -are grouped into :class:`AxisGroup`/:class:`BoundAxisGroup` objects with per-axis -and intensity :class:`Calibration` mappings. +are grouped into :class:`Axis`/:class:`AxisGroup` objects with per-axis and +intensity :class:`Calibration` mappings. """ from __future__ import annotations @@ -102,46 +102,27 @@ def with_calibration(self, key: str, calibration: Calibration, *, make_primary: @dataclasses.dataclass(frozen=True) class Axis: - """A dimension label and display metadata independent of coordinate mappings.""" + """A dimension label, size, and coordinate metadata.""" label: str = "" - - -@dataclasses.dataclass(frozen=True) -class BoundAxis: - """An :class:`Axis` bound to a concrete (positive) length.""" - - axis: Axis - size: int + size: int = 1 def __post_init__(self) -> None: if self.size <= 0: - raise ValueError(f"BoundAxis size must be positive, got {self.size}") + raise ValueError(f"Axis size must be positive, got {self.size}") @dataclasses.dataclass(frozen=True) class AxisGroup: - """An ordered group of axes (e.g. spatial or spectral) without calibrations.""" + """An ordered group of sized axes with optional coordinate mappings.""" axes: tuple[Axis, ...] = dataclasses.field(default_factory=tuple) - - @property - def rank(self) -> int: - return len(self.axes) - - -@dataclasses.dataclass(frozen=True) -class BoundAxisGroup: - """An :class:`AxisGroup` with sized axes, exposing a concrete shape.""" - - bound_axes: tuple[BoundAxis, ...] = dataclasses.field(default_factory=tuple) coordinate_system_id: str | None = None coordinate_mappings: typing.Mapping[str, tuple[Calibration, ...]] = dataclasses.field(default_factory=dict) primary_mapping_key: str | None = None - axis_group: AxisGroup = dataclasses.field(init=False) def __post_init__(self) -> None: - object.__setattr__(self, "axis_group", AxisGroup(axes=tuple(ba.axis for ba in self.bound_axes))) + object.__setattr__(self, "axes", tuple(self.axes)) coordinate_mappings = dict(self.coordinate_mappings) normalized_coordinate_mappings: dict[str, tuple[Calibration, ...]] = dict() for key, mapping in coordinate_mappings.items(): @@ -157,11 +138,11 @@ def __post_init__(self) -> None: raise ValueError("primary_mapping_key must be None when coordinate_mappings is empty") @staticmethod - def from_1d_size(size: int, *, label: str = "x", unit: str | None = None) -> BoundAxisGroup: - """Create a 1D bound axis group with one default coordinate mapping.""" + def from_1d_size(size: int, *, label: str = "x", unit: str | None = None) -> AxisGroup: + """Create a 1D axis group with one default coordinate mapping.""" calibration = AffineCalibration(unit=unit or "") - return BoundAxisGroup( - bound_axes=(BoundAxis(Axis(label), size),), + return AxisGroup( + axes=(Axis(label, size),), coordinate_mappings={ DEFAULT_COORDINATE_MAPPING_KEY: (calibration,), }, @@ -169,15 +150,15 @@ def from_1d_size(size: int, *, label: str = "x", unit: str | None = None) -> Bou ) @staticmethod - def from_2d_size(size: tuple[int, int], *, labels: tuple[str, str] = ("x", "y"), unit: str | None = None) -> BoundAxisGroup: - """Create a 2D bound axis group with one default coordinate mapping.""" + def from_2d_size(size: tuple[int, int], *, labels: tuple[str, str] = ("x", "y"), unit: str | None = None) -> AxisGroup: + """Create a 2D axis group with one default coordinate mapping.""" size_x, size_y = size x_label, y_label = labels calibration = AffineCalibration(unit=unit or "") - return BoundAxisGroup( - bound_axes=( - BoundAxis(Axis(x_label), size_x), - BoundAxis(Axis(y_label), size_y), + return AxisGroup( + axes=( + Axis(x_label, size_x), + Axis(y_label, size_y), ), coordinate_mappings={ DEFAULT_COORDINATE_MAPPING_KEY: (calibration, calibration), @@ -187,11 +168,11 @@ def from_2d_size(size: tuple[int, int], *, labels: tuple[str, str] = ("x", "y"), @property def shape(self) -> tuple[int, ...]: - return tuple(bound_axis.size for bound_axis in self.bound_axes) + return tuple(axis.size for axis in self.axes) @property def rank(self) -> int: - return len(self.bound_axes) + return len(self.axes) @property def units(self) -> list[str]: @@ -220,23 +201,23 @@ def get_calibration(self, axis: int, key: str | None = None) -> Calibration: class ArrayDescriptor: """Intrinsic structure and calibrations used to interpret an array.""" - bound_axis_groups: tuple[BoundAxisGroup, ...] + axis_groups: tuple[AxisGroup, ...] intensity_calibrations: CalibrationSet = dataclasses.field(default_factory=CalibrationSet) def __post_init__(self) -> None: - object.__setattr__(self, "bound_axis_groups", tuple(self.bound_axis_groups)) - if not self.bound_axis_groups: - raise ValueError("ArrayDescriptor requires at least one bound axis group") - if any(bound_axis_group.rank == 0 for bound_axis_group in self.bound_axis_groups[:-1]): + object.__setattr__(self, "axis_groups", tuple(self.axis_groups)) + if not self.axis_groups: + raise ValueError("ArrayDescriptor requires at least one axis group") + if any(axis_group.rank == 0 for axis_group in self.axis_groups[:-1]): raise ValueError("Only the final axis group may have rank 0") @property def shape(self) -> tuple[int, ...]: - return tuple(dimension for bound_axis_group in self.bound_axis_groups for dimension in bound_axis_group.shape) + return tuple(dimension for axis_group in self.axis_groups for dimension in axis_group.shape) @property def ndim(self) -> int: - return sum(bound_axis_group.rank for bound_axis_group in self.bound_axis_groups) + return sum(axis_group.rank for axis_group in self.axis_groups) def get_intensity_calibration(self, key: str | None = None) -> Calibration: return self.intensity_calibrations.get_calibration(key) @@ -375,9 +356,9 @@ def get_intensity_calibration(self, key: str | None = None) -> Calibration: return self.descriptor.get_intensity_calibration(key) def get_flat_axis_calibrations(self, key: str | None = None) -> list[Calibration]: - return [bound_axis_group.get_calibration(axis=i, key=key) for bound_axis_group in self.descriptor.bound_axis_groups for i in range(bound_axis_group.rank)] + return [axis_group.get_calibration(axis=i, key=key) for axis_group in self.descriptor.axis_groups for i in range(axis_group.rank)] -def zeros_annotated_array(bound_axis_groups: typing.Sequence[BoundAxisGroup], dtype: numpy.typing.DTypeLike = numpy.float64) -> AnnotatedArray: - descriptor = ArrayDescriptor(bound_axis_groups=tuple(bound_axis_groups)) +def zeros_annotated_array(axis_groups: typing.Sequence[AxisGroup], dtype: numpy.typing.DTypeLike = numpy.float64) -> AnnotatedArray: + descriptor = ArrayDescriptor(axis_groups=tuple(axis_groups)) return AnnotatedArray(data=numpy.zeros(descriptor.shape, dtype=dtype), descriptor=descriptor) diff --git a/nion/data/test/AnnotatedArray_test.py b/nion/data/test/AnnotatedArray_test.py index f8bb255..ee5cbb3 100644 --- a/nion/data/test/AnnotatedArray_test.py +++ b/nion/data/test/AnnotatedArray_test.py @@ -20,8 +20,8 @@ def test_calibration_set_uses_explicit_calibration_accessors(self) -> None: self.assertIs(alternate, calibrations.with_primary_calibration("alternate").primary_calibration) def test_array_descriptor_describes_shape_and_rank(self) -> None: - collection = AnnotatedArray.BoundAxisGroup.from_1d_size(3) - signal = AnnotatedArray.BoundAxisGroup.from_2d_size((4, 5)) + collection = AnnotatedArray.AxisGroup.from_1d_size(3) + signal = AnnotatedArray.AxisGroup.from_2d_size((4, 5)) descriptor = AnnotatedArray.ArrayDescriptor((collection, signal)) self.assertEqual((3, 4, 5), descriptor.shape) @@ -31,8 +31,8 @@ def test_array_descriptor_requires_valid_axis_group_layout(self) -> None: with self.assertRaisesRegex(ValueError, "at least one"): AnnotatedArray.ArrayDescriptor(()) - scalar_group = AnnotatedArray.BoundAxisGroup() - vector_group = AnnotatedArray.BoundAxisGroup.from_1d_size(3) + scalar_group = AnnotatedArray.AxisGroup() + vector_group = AnnotatedArray.AxisGroup.from_1d_size(3) with self.assertRaisesRegex(ValueError, "Only the final"): AnnotatedArray.ArrayDescriptor((scalar_group, vector_group)) @@ -87,7 +87,7 @@ def test_array_metadata_created_retains_iana_zone_and_offset(self) -> None: self.assertEqual(datetime.timedelta(hours=-7), metadata.created.utcoffset()) def test_array_header_can_be_passed_without_data(self) -> None: - descriptor = AnnotatedArray.ArrayDescriptor((AnnotatedArray.BoundAxisGroup.from_2d_size((4, 5)),)) + descriptor = AnnotatedArray.ArrayDescriptor((AnnotatedArray.AxisGroup.from_2d_size((4, 5)),)) metadata = AnnotatedArray.ArrayMetadata(attributes={"note": "test"}) header = AnnotatedArray.ArrayHeader(descriptor, "float32", metadata) @@ -103,7 +103,7 @@ def test_array_header_can_be_passed_without_data(self) -> None: self.assertIs(metadata, first.metadata) def test_annotated_array_validates_data_against_header(self) -> None: - descriptor = AnnotatedArray.ArrayDescriptor((AnnotatedArray.BoundAxisGroup.from_1d_size(3),)) + descriptor = AnnotatedArray.ArrayDescriptor((AnnotatedArray.AxisGroup.from_1d_size(3),)) header = AnnotatedArray.ArrayHeader(descriptor, numpy.float32) with self.assertRaisesRegex(ValueError, "shape"): @@ -112,12 +112,12 @@ def test_annotated_array_validates_data_against_header(self) -> None: AnnotatedArray.AnnotatedArray.from_header(numpy.zeros((3,), dtype=numpy.float64), header) def test_zeros_annotated_array_constructs_matching_header(self) -> None: - group = AnnotatedArray.BoundAxisGroup.from_2d_size((2, 3), unit="nm") + group = AnnotatedArray.AxisGroup.from_2d_size((2, 3), unit="nm") array = AnnotatedArray.zeros_annotated_array((group,), dtype=numpy.float32) self.assertEqual((2, 3), array.data.shape) self.assertEqual(numpy.dtype(numpy.float32), array.header.dtype) - self.assertEqual((group,), array.descriptor.bound_axis_groups) + self.assertEqual((group,), array.descriptor.axis_groups) self.assertEqual(2, len(array.get_flat_axis_calibrations())) From 07a78abaf5984ee87b810ec3ba5ccea23e2514d9 Mon Sep 17 00:00:00 2001 From: Chris Meyer <34664+cmeyer@users.noreply.github.com> Date: Wed, 22 Jul 2026 16:31:36 -0700 Subject: [PATCH 10/16] WIP: value type. coordinate mapping. --- nion/data/AnnotatedArray.py | 107 +++++++++++++++++++++++++++++++----- 1 file changed, 94 insertions(+), 13 deletions(-) diff --git a/nion/data/AnnotatedArray.py b/nion/data/AnnotatedArray.py index 5f956c3..84844b2 100644 --- a/nion/data/AnnotatedArray.py +++ b/nion/data/AnnotatedArray.py @@ -22,6 +22,49 @@ DEFAULT_COORDINATE_MAPPING_KEY = "default" +class ValueType: + """Supported value types for array data.""" + SCALAR = "scalar" + COMPLEX = "complex" + RGB = "rgb" + RGBA = "rgba" + VECTOR = "vector" + + +def infer_value_type(dtype: numpy.dtype) -> str: + """Infer the value type from a numpy dtype. + + Args: + dtype: The numpy dtype to infer from. + + Returns: + A ValueType string (scalar, complex, rgb, rgba, or vector). + + Raises: + ValueError: If the dtype cannot be mapped to a supported value type. + """ + dtype = numpy.dtype(typing.cast(typing.Any, dtype)) + + # Complex types + if dtype.kind == 'c': + return ValueType.COMPLEX + + # Structured/record types for RGB and RGBA + if dtype.names is not None: + if len(dtype.names) == 3 and set(dtype.names) >= {'r', 'g', 'b'}: + return ValueType.RGB + elif len(dtype.names) == 4 and set(dtype.names) >= {'r', 'g', 'b', 'a'}: + return ValueType.RGBA + # Other structured types are treated as vectors + return ValueType.VECTOR + + # Numeric types (int, uint, float, bool) + if dtype.kind in ('b', 'i', 'u', 'f'): + return ValueType.SCALAR + + raise ValueError(f"Unsupported dtype {dtype} for value type inference") + + class Calibration(typing.Protocol): """Protocol converting between array indices and physical coordinates.""" @@ -112,24 +155,34 @@ def __post_init__(self) -> None: raise ValueError(f"Axis size must be positive, got {self.size}") +@dataclasses.dataclass(frozen=True) +class CoordinateMapping: + """A labeled per-axis coordinate mapping for an axis group.""" + + calibrations: tuple[Calibration, ...] = dataclasses.field(default_factory=tuple) + label: str | None = None + + def __post_init__(self) -> None: + object.__setattr__(self, "calibrations", tuple(self.calibrations)) + + @dataclasses.dataclass(frozen=True) class AxisGroup: """An ordered group of sized axes with optional coordinate mappings.""" axes: tuple[Axis, ...] = dataclasses.field(default_factory=tuple) coordinate_system_id: str | None = None - coordinate_mappings: typing.Mapping[str, tuple[Calibration, ...]] = dataclasses.field(default_factory=dict) + coordinate_mappings: typing.Mapping[str, CoordinateMapping] = dataclasses.field(default_factory=dict) primary_mapping_key: str | None = None def __post_init__(self) -> None: object.__setattr__(self, "axes", tuple(self.axes)) coordinate_mappings = dict(self.coordinate_mappings) - normalized_coordinate_mappings: dict[str, tuple[Calibration, ...]] = dict() + normalized_coordinate_mappings: dict[str, CoordinateMapping] = dict() for key, mapping in coordinate_mappings.items(): - normalized_mapping = tuple(mapping) - if len(normalized_mapping) != self.rank: - raise ValueError(f"coordinate mapping {key!r} rank {len(normalized_mapping)} does not match axis group rank {self.rank}") - normalized_coordinate_mappings[key] = normalized_mapping + if len(mapping.calibrations) != self.rank: + raise ValueError(f"coordinate mapping {key!r} rank {len(mapping.calibrations)} does not match axis group rank {self.rank}") + normalized_coordinate_mappings[key] = mapping object.__setattr__(self, "coordinate_mappings", types.MappingProxyType(normalized_coordinate_mappings)) if self.primary_mapping_key is not None and self.primary_mapping_key not in normalized_coordinate_mappings: @@ -144,7 +197,7 @@ def from_1d_size(size: int, *, label: str = "x", unit: str | None = None) -> Axi return AxisGroup( axes=(Axis(label, size),), coordinate_mappings={ - DEFAULT_COORDINATE_MAPPING_KEY: (calibration,), + DEFAULT_COORDINATE_MAPPING_KEY: CoordinateMapping(calibrations=(calibration,)), }, primary_mapping_key=DEFAULT_COORDINATE_MAPPING_KEY, ) @@ -161,7 +214,7 @@ def from_2d_size(size: tuple[int, int], *, labels: tuple[str, str] = ("x", "y"), Axis(y_label, size_y), ), coordinate_mappings={ - DEFAULT_COORDINATE_MAPPING_KEY: (calibration, calibration), + DEFAULT_COORDINATE_MAPPING_KEY: CoordinateMapping(calibrations=(calibration, calibration)), }, primary_mapping_key=DEFAULT_COORDINATE_MAPPING_KEY, ) @@ -178,13 +231,13 @@ def rank(self) -> int: def units(self) -> list[str]: if self.primary_mapping_key is None: return [] - return [calibration.unit for calibration in self.get_coordinate_mapping()] + return [calibration.unit for calibration in self.get_coordinate_mapping().calibrations] @property def mapping_keys(self) -> tuple[str, ...]: return tuple(self.coordinate_mappings.keys()) - def get_coordinate_mapping(self, key: str | None = None) -> tuple[Calibration, ...]: + def get_coordinate_mapping(self, key: str | None = None) -> CoordinateMapping: target_key = self.primary_mapping_key if key is None else key if target_key is None: raise KeyError("No primary coordinate mapping is designated") @@ -194,7 +247,7 @@ def get_coordinate_mapping(self, key: str | None = None) -> tuple[Calibration, . return mapping def get_calibration(self, axis: int, key: str | None = None) -> Calibration: - return self.get_coordinate_mapping(key)[axis] + return self.get_coordinate_mapping(key).calibrations[axis] @dataclasses.dataclass(frozen=True) @@ -203,6 +256,7 @@ class ArrayDescriptor: axis_groups: tuple[AxisGroup, ...] intensity_calibrations: CalibrationSet = dataclasses.field(default_factory=CalibrationSet) + value_type: str = ValueType.SCALAR def __post_init__(self) -> None: object.__setattr__(self, "axis_groups", tuple(self.axis_groups)) @@ -210,6 +264,9 @@ def __post_init__(self) -> None: raise ValueError("ArrayDescriptor requires at least one axis group") if any(axis_group.rank == 0 for axis_group in self.axis_groups[:-1]): raise ValueError("Only the final axis group may have rank 0") + valid_types = {ValueType.SCALAR, ValueType.COMPLEX, ValueType.RGB, ValueType.RGBA, ValueType.VECTOR} + if self.value_type not in valid_types: + raise ValueError(f"value_type must be one of {valid_types}, got {self.value_type!r}") @property def shape(self) -> tuple[int, ...]: @@ -239,6 +296,9 @@ def intensity_calibration_keys(self) -> tuple[str, ...]: def primary_intensity_calibration_key(self) -> str: return self.intensity_calibrations.primary_key + def with_value_type(self, value_type: str) -> ArrayDescriptor: + return dataclasses.replace(self, value_type=value_type) + @dataclasses.dataclass(frozen=True) class ExtensionRecord: @@ -341,6 +401,9 @@ class AnnotatedArray: def __post_init__(self) -> None: if self.descriptor.shape != self.data.shape: raise ValueError(f"Descriptor shape {self.descriptor.shape} does not match array shape {self.data.shape}") + inferred_value_type = infer_value_type(self.data.dtype) + if self.descriptor.value_type != inferred_value_type: + raise ValueError(f"Descriptor value_type {self.descriptor.value_type!r} does not match inferred value_type {inferred_value_type!r} from array dtype {self.data.dtype}") @property def header(self) -> ArrayHeader: @@ -350,6 +413,11 @@ def header(self) -> ArrayHeader: def from_header(cls, data: numpy.typing.NDArray[typing.Any], header: ArrayHeader) -> AnnotatedArray: if header.dtype != data.dtype: raise ValueError(f"Header dtype {header.dtype} does not match array dtype {data.dtype}") + if header.descriptor.shape != data.shape: + raise ValueError(f"Header shape {header.descriptor.shape} does not match array shape {data.shape}") + inferred_value_type = infer_value_type(data.dtype) + if header.descriptor.value_type != inferred_value_type: + raise ValueError(f"Header value_type {header.descriptor.value_type!r} does not match inferred value_type {inferred_value_type!r} from array dtype {data.dtype}") return cls(data, header.descriptor, header.metadata) def get_intensity_calibration(self, key: str | None = None) -> Calibration: @@ -360,5 +428,18 @@ def get_flat_axis_calibrations(self, key: str | None = None) -> list[Calibration def zeros_annotated_array(axis_groups: typing.Sequence[AxisGroup], dtype: numpy.typing.DTypeLike = numpy.float64) -> AnnotatedArray: - descriptor = ArrayDescriptor(axis_groups=tuple(axis_groups)) - return AnnotatedArray(data=numpy.zeros(descriptor.shape, dtype=dtype), descriptor=descriptor) + """Create an AnnotatedArray filled with zeros. + + The value_type is inferred from the provided dtype. + + Args: + axis_groups: Sequence of AxisGroup objects describing the array structure. + dtype: The numpy dtype for the array. Defaults to float64 (scalar). + + Returns: + An AnnotatedArray with the specified structure, filled with zeros. + """ + dtype_obj = numpy.dtype(typing.cast(typing.Any, dtype)) + value_type = infer_value_type(dtype_obj) + descriptor = ArrayDescriptor(axis_groups=tuple(axis_groups), value_type=value_type) + return AnnotatedArray(data=numpy.zeros(descriptor.shape, dtype=dtype_obj), descriptor=descriptor) From 51c0321a9ceff3e9f69611040c02aeb94860c88a Mon Sep 17 00:00:00 2001 From: Chris Meyer <34664+cmeyer@users.noreply.github.com> Date: Fri, 24 Jul 2026 15:34:28 -0700 Subject: [PATCH 11/16] WIP: clean up use of default coordinate mappings and intensity calibrations. --- nion/data/AnnotatedArray.py | 63 +++++++++++++-------------- nion/data/test/AnnotatedArray_test.py | 20 ++++----- 2 files changed, 40 insertions(+), 43 deletions(-) diff --git a/nion/data/AnnotatedArray.py b/nion/data/AnnotatedArray.py index 84844b2..c19bb5e 100644 --- a/nion/data/AnnotatedArray.py +++ b/nion/data/AnnotatedArray.py @@ -18,10 +18,6 @@ import tzlocal -DEFAULT_CALIBRATION_KEY = "default" -DEFAULT_COORDINATE_MAPPING_KEY = "default" - - class ValueType: """Supported value types for array data.""" SCALAR = "scalar" @@ -96,17 +92,17 @@ def to_index(self, coordinate: float) -> float: @dataclasses.dataclass(frozen=True) class CalibrationSet: - """Keyed calibrations with an explicit primary calibration key.""" + """Keyed calibrations with an optional primary calibration key.""" - calibrations: typing.Mapping[str, Calibration] = dataclasses.field(default_factory=lambda: {DEFAULT_CALIBRATION_KEY: AffineCalibration()}) - primary_key: str = DEFAULT_CALIBRATION_KEY + calibrations: typing.Mapping[str, Calibration] = dataclasses.field(default_factory=dict) + primary_key: str | None = None def __post_init__(self) -> None: calibrations = dict(self.calibrations) - if not calibrations: - raise ValueError("calibrations must not be empty") - if self.primary_key not in calibrations: + if self.primary_key is not None and self.primary_key not in calibrations: raise ValueError(f"primary_key {self.primary_key!r} is not present in calibrations") + if not calibrations and self.primary_key is not None: + raise ValueError("primary_key must be None when calibrations is empty") object.__setattr__(self, "calibrations", types.MappingProxyType(calibrations)) @property @@ -115,6 +111,8 @@ def calibration_keys(self) -> tuple[str, ...]: @property def primary_calibration(self) -> Calibration: + if self.primary_key is None: + raise KeyError("No primary calibration is designated") return self.calibrations[self.primary_key] def has_calibration(self, key: str) -> bool: @@ -122,13 +120,15 @@ def has_calibration(self, key: str) -> bool: def get_calibration(self, key: str | None = None) -> Calibration: target_key = self.primary_key if key is None else key + if target_key is None: + raise KeyError("No primary calibration is designated") calibration = self.calibrations.get(target_key) if calibration is None: raise KeyError(f"Unknown calibration {target_key!r}") return calibration @staticmethod - def from_calibration(calibration: Calibration, key: str = DEFAULT_CALIBRATION_KEY) -> CalibrationSet: + def from_calibration(calibration: Calibration, key: str) -> CalibrationSet: return CalibrationSet(calibrations={key: calibration}, primary_key=key) def with_primary_calibration(self, key: str) -> CalibrationSet: @@ -191,32 +191,20 @@ def __post_init__(self) -> None: raise ValueError("primary_mapping_key must be None when coordinate_mappings is empty") @staticmethod - def from_1d_size(size: int, *, label: str = "x", unit: str | None = None) -> AxisGroup: - """Create a 1D axis group with one default coordinate mapping.""" - calibration = AffineCalibration(unit=unit or "") - return AxisGroup( - axes=(Axis(label, size),), - coordinate_mappings={ - DEFAULT_COORDINATE_MAPPING_KEY: CoordinateMapping(calibrations=(calibration,)), - }, - primary_mapping_key=DEFAULT_COORDINATE_MAPPING_KEY, - ) + def from_1d_size(size: int, *, label: str = "x") -> AxisGroup: + """Create a 1D axis group with no coordinate mappings.""" + return AxisGroup(axes=(Axis(label, size),)) @staticmethod - def from_2d_size(size: tuple[int, int], *, labels: tuple[str, str] = ("x", "y"), unit: str | None = None) -> AxisGroup: - """Create a 2D axis group with one default coordinate mapping.""" + def from_2d_size(size: tuple[int, int], *, labels: tuple[str, str] = ("x", "y")) -> AxisGroup: + """Create a 2D axis group with no coordinate mappings.""" size_x, size_y = size x_label, y_label = labels - calibration = AffineCalibration(unit=unit or "") return AxisGroup( axes=( Axis(x_label, size_x), Axis(y_label, size_y), ), - coordinate_mappings={ - DEFAULT_COORDINATE_MAPPING_KEY: CoordinateMapping(calibrations=(calibration, calibration)), - }, - primary_mapping_key=DEFAULT_COORDINATE_MAPPING_KEY, ) @property @@ -237,6 +225,17 @@ def units(self) -> list[str]: def mapping_keys(self) -> tuple[str, ...]: return tuple(self.coordinate_mappings.keys()) + def with_coordinate_mapping(self, key: str, coordinate_mapping: CoordinateMapping) -> AxisGroup: + """Replace all coordinate mappings with one mapping and make it primary.""" + return dataclasses.replace(self, coordinate_mappings={key: coordinate_mapping}, primary_mapping_key=key) + + def with_added_coordinate_mapping(self, key: str, coordinate_mapping: CoordinateMapping, *, make_primary: bool = False) -> AxisGroup: + """Add or replace one keyed coordinate mapping.""" + coordinate_mappings = dict(self.coordinate_mappings) + coordinate_mappings[key] = coordinate_mapping + primary_mapping_key = key if make_primary else self.primary_mapping_key + return dataclasses.replace(self, coordinate_mappings=coordinate_mappings, primary_mapping_key=primary_mapping_key) + def get_coordinate_mapping(self, key: str | None = None) -> CoordinateMapping: target_key = self.primary_mapping_key if key is None else key if target_key is None: @@ -293,7 +292,7 @@ def intensity_calibration_keys(self) -> tuple[str, ...]: return self.intensity_calibrations.calibration_keys @property - def primary_intensity_calibration_key(self) -> str: + def primary_intensity_calibration_key(self) -> str | None: return self.intensity_calibrations.primary_key def with_value_type(self, value_type: str) -> ArrayDescriptor: @@ -306,15 +305,15 @@ class ExtensionRecord: extension_type_id: str schema_version: int - payload: bytes + payload: str def __post_init__(self) -> None: if not self.extension_type_id: raise ValueError("ExtensionRecord extension_type_id must not be empty") if self.schema_version < 1: raise ValueError("ExtensionRecord schema_version must be positive") - if not isinstance(self.payload, bytes): - raise TypeError("ExtensionRecord payload must be bytes") + if not isinstance(self.payload, str): + raise TypeError("ExtensionRecord payload must be str") @dataclasses.dataclass(frozen=True, init=False) diff --git a/nion/data/test/AnnotatedArray_test.py b/nion/data/test/AnnotatedArray_test.py index ee5cbb3..449c6a9 100644 --- a/nion/data/test/AnnotatedArray_test.py +++ b/nion/data/test/AnnotatedArray_test.py @@ -12,7 +12,7 @@ class TestAnnotatedArray(unittest.TestCase): def test_calibration_set_uses_explicit_calibration_accessors(self) -> None: primary = AnnotatedArray.AffineCalibration(unit="nm") alternate = AnnotatedArray.AffineCalibration(unit="rad") - calibrations = AnnotatedArray.CalibrationSet.from_calibration(primary).with_calibration("alternate", alternate) + calibrations = AnnotatedArray.CalibrationSet.from_calibration(primary, key="primary").with_calibration("alternate", alternate) self.assertTrue(calibrations.has_calibration("alternate")) self.assertIs(primary, calibrations.primary_calibration) @@ -37,7 +37,7 @@ def test_array_descriptor_requires_valid_axis_group_layout(self) -> None: AnnotatedArray.ArrayDescriptor((scalar_group, vector_group)) def test_array_metadata_controls_extension_access(self) -> None: - extension = AnnotatedArray.ExtensionRecord("org.nion.test", 1, b"value=42") + extension = AnnotatedArray.ExtensionRecord("org.nion.test", 1, "value=42") metadata = AnnotatedArray.ArrayMetadata(extensions=(extension,)) self.assertEqual("org.nion.test", extension.extension_type_id) @@ -47,7 +47,7 @@ def test_array_metadata_controls_extension_access(self) -> None: with self.assertRaises(KeyError): metadata.get_extension("org.nion.missing") - replacement = AnnotatedArray.ExtensionRecord("org.nion.test", 2, b"value=43") + replacement = AnnotatedArray.ExtensionRecord("org.nion.test", 2, "value=43") replaced_metadata = metadata.with_extension(replacement) self.assertEqual(2, replaced_metadata.get_extension("org.nion.test").schema_version) self.assertEqual(1, metadata.get_extension("org.nion.test").schema_version) @@ -55,13 +55,13 @@ def test_array_metadata_controls_extension_access(self) -> None: def test_array_metadata_validates_extensions(self) -> None: with self.assertRaisesRegex(ValueError, "must not be empty"): - AnnotatedArray.ExtensionRecord("", 1, b"") + AnnotatedArray.ExtensionRecord("", 1, "") with self.assertRaisesRegex(ValueError, "must be positive"): - AnnotatedArray.ExtensionRecord("org.nion.test", 0, b"") - with self.assertRaisesRegex(TypeError, "must be bytes"): + AnnotatedArray.ExtensionRecord("org.nion.test", 0, "") + with self.assertRaisesRegex(TypeError, "must be str"): AnnotatedArray.ExtensionRecord("org.nion.test", 1, bytearray()) # type: ignore[arg-type] - extension = AnnotatedArray.ExtensionRecord("org.nion.test", 1, b"") + extension = AnnotatedArray.ExtensionRecord("org.nion.test", 1, "") with self.assertRaisesRegex(ValueError, "Duplicate"): AnnotatedArray.ArrayMetadata(extensions=(extension, extension)) @@ -111,14 +111,12 @@ def test_annotated_array_validates_data_against_header(self) -> None: with self.assertRaisesRegex(ValueError, "dtype"): AnnotatedArray.AnnotatedArray.from_header(numpy.zeros((3,), dtype=numpy.float64), header) - def test_zeros_annotated_array_constructs_matching_header(self) -> None: - group = AnnotatedArray.AxisGroup.from_2d_size((2, 3), unit="nm") + def test_zeros_annotated_array_constructs_matching_shape_and_dtype(self) -> None: + group = AnnotatedArray.AxisGroup.from_2d_size((2, 3)) array = AnnotatedArray.zeros_annotated_array((group,), dtype=numpy.float32) self.assertEqual((2, 3), array.data.shape) self.assertEqual(numpy.dtype(numpy.float32), array.header.dtype) - self.assertEqual((group,), array.descriptor.axis_groups) - self.assertEqual(2, len(array.get_flat_axis_calibrations())) if __name__ == "__main__": From 6ede480cbdeea14a47349670c4dea5e48546b027 Mon Sep 17 00:00:00 2001 From: Chris Meyer <34664+cmeyer@users.noreply.github.com> Date: Fri, 24 Jul 2026 16:21:22 -0700 Subject: [PATCH 12/16] WIP: numpy array typing. and coordinate-calibrations instead of mapping. --- nion/data/AnnotatedArray.py | 108 +++++++++++++++----------- nion/data/test/AnnotatedArray_test.py | 45 +++++++++++ 2 files changed, 108 insertions(+), 45 deletions(-) diff --git a/nion/data/AnnotatedArray.py b/nion/data/AnnotatedArray.py index c19bb5e..881ab0e 100644 --- a/nion/data/AnnotatedArray.py +++ b/nion/data/AnnotatedArray.py @@ -61,6 +61,8 @@ def infer_value_type(dtype: numpy.dtype) -> str: raise ValueError(f"Unsupported dtype {dtype} for value type inference") + + class Calibration(typing.Protocol): """Protocol converting between array indices and physical coordinates.""" @@ -156,8 +158,8 @@ def __post_init__(self) -> None: @dataclasses.dataclass(frozen=True) -class CoordinateMapping: - """A labeled per-axis coordinate mapping for an axis group.""" +class CoordinateCalibration: + """A labeled per-axis coordinate calibration bundle for an axis group.""" calibrations: tuple[Calibration, ...] = dataclasses.field(default_factory=tuple) label: str | None = None @@ -168,36 +170,36 @@ def __post_init__(self) -> None: @dataclasses.dataclass(frozen=True) class AxisGroup: - """An ordered group of sized axes with optional coordinate mappings.""" + """An ordered group of sized axes with optional coordinate calibrations.""" axes: tuple[Axis, ...] = dataclasses.field(default_factory=tuple) coordinate_system_id: str | None = None - coordinate_mappings: typing.Mapping[str, CoordinateMapping] = dataclasses.field(default_factory=dict) - primary_mapping_key: str | None = None + coordinate_calibrations: typing.Mapping[str, CoordinateCalibration] = dataclasses.field(default_factory=dict) + primary_calibration_key: str | None = None def __post_init__(self) -> None: object.__setattr__(self, "axes", tuple(self.axes)) - coordinate_mappings = dict(self.coordinate_mappings) - normalized_coordinate_mappings: dict[str, CoordinateMapping] = dict() - for key, mapping in coordinate_mappings.items(): - if len(mapping.calibrations) != self.rank: - raise ValueError(f"coordinate mapping {key!r} rank {len(mapping.calibrations)} does not match axis group rank {self.rank}") - normalized_coordinate_mappings[key] = mapping - object.__setattr__(self, "coordinate_mappings", types.MappingProxyType(normalized_coordinate_mappings)) - - if self.primary_mapping_key is not None and self.primary_mapping_key not in normalized_coordinate_mappings: - raise ValueError(f"primary_mapping_key {self.primary_mapping_key!r} is not present in coordinate_mappings") - if not normalized_coordinate_mappings and self.primary_mapping_key is not None: - raise ValueError("primary_mapping_key must be None when coordinate_mappings is empty") + coordinate_calibrations = dict(self.coordinate_calibrations) + normalized_coordinate_calibrations: dict[str, CoordinateCalibration] = dict() + for key, coordinate_calibration in coordinate_calibrations.items(): + if len(coordinate_calibration.calibrations) != self.rank: + raise ValueError(f"coordinate calibration {key!r} rank {len(coordinate_calibration.calibrations)} does not match axis group rank {self.rank}") + normalized_coordinate_calibrations[key] = coordinate_calibration + object.__setattr__(self, "coordinate_calibrations", types.MappingProxyType(normalized_coordinate_calibrations)) + + if self.primary_calibration_key is not None and self.primary_calibration_key not in normalized_coordinate_calibrations: + raise ValueError(f"primary_calibration_key {self.primary_calibration_key!r} is not present in coordinate_calibrations") + if not normalized_coordinate_calibrations and self.primary_calibration_key is not None: + raise ValueError("primary_calibration_key must be None when coordinate_calibrations is empty") @staticmethod def from_1d_size(size: int, *, label: str = "x") -> AxisGroup: - """Create a 1D axis group with no coordinate mappings.""" + """Create a 1D axis group with no coordinate calibrations.""" return AxisGroup(axes=(Axis(label, size),)) @staticmethod def from_2d_size(size: tuple[int, int], *, labels: tuple[str, str] = ("x", "y")) -> AxisGroup: - """Create a 2D axis group with no coordinate mappings.""" + """Create a 2D axis group with no coordinate calibrations.""" size_x, size_y = size x_label, y_label = labels return AxisGroup( @@ -217,36 +219,36 @@ def rank(self) -> int: @property def units(self) -> list[str]: - if self.primary_mapping_key is None: + if self.primary_calibration_key is None: return [] - return [calibration.unit for calibration in self.get_coordinate_mapping().calibrations] + return [calibration.unit for calibration in self.get_coordinate_calibration().calibrations] @property - def mapping_keys(self) -> tuple[str, ...]: - return tuple(self.coordinate_mappings.keys()) - - def with_coordinate_mapping(self, key: str, coordinate_mapping: CoordinateMapping) -> AxisGroup: - """Replace all coordinate mappings with one mapping and make it primary.""" - return dataclasses.replace(self, coordinate_mappings={key: coordinate_mapping}, primary_mapping_key=key) - - def with_added_coordinate_mapping(self, key: str, coordinate_mapping: CoordinateMapping, *, make_primary: bool = False) -> AxisGroup: - """Add or replace one keyed coordinate mapping.""" - coordinate_mappings = dict(self.coordinate_mappings) - coordinate_mappings[key] = coordinate_mapping - primary_mapping_key = key if make_primary else self.primary_mapping_key - return dataclasses.replace(self, coordinate_mappings=coordinate_mappings, primary_mapping_key=primary_mapping_key) - - def get_coordinate_mapping(self, key: str | None = None) -> CoordinateMapping: - target_key = self.primary_mapping_key if key is None else key + def calibration_keys(self) -> tuple[str, ...]: + return tuple(self.coordinate_calibrations.keys()) + + def with_coordinate_calibration(self, key: str, coordinate_calibration: CoordinateCalibration) -> AxisGroup: + """Replace all coordinate calibrations with one calibration and make it primary.""" + return dataclasses.replace(self, coordinate_calibrations={key: coordinate_calibration}, primary_calibration_key=key) + + def with_added_coordinate_calibration(self, key: str, coordinate_calibration: CoordinateCalibration, *, make_primary: bool = False) -> AxisGroup: + """Add or replace one keyed coordinate calibration.""" + coordinate_calibrations = dict(self.coordinate_calibrations) + coordinate_calibrations[key] = coordinate_calibration + primary_calibration_key = key if make_primary else self.primary_calibration_key + return dataclasses.replace(self, coordinate_calibrations=coordinate_calibrations, primary_calibration_key=primary_calibration_key) + + def get_coordinate_calibration(self, key: str | None = None) -> CoordinateCalibration: + target_key = self.primary_calibration_key if key is None else key if target_key is None: - raise KeyError("No primary coordinate mapping is designated") - mapping = self.coordinate_mappings.get(target_key) - if mapping is None: - raise KeyError(f"Unknown coordinate mapping {target_key!r}") - return mapping + raise KeyError("No primary coordinate calibration is designated") + coordinate_calibration = self.coordinate_calibrations.get(target_key) + if coordinate_calibration is None: + raise KeyError(f"Unknown coordinate calibration {target_key!r}") + return coordinate_calibration def get_calibration(self, axis: int, key: str | None = None) -> Calibration: - return self.get_coordinate_mapping(key).calibrations[axis] + return self.get_coordinate_calibration(key).calibrations[axis] @dataclasses.dataclass(frozen=True) @@ -389,11 +391,23 @@ def shape(self) -> tuple[int, ...]: return self.descriptor.shape +class ArrayProtocol(typing.Protocol): + """Duck-array interface required by AnnotatedArray (satisfied by numpy.ndarray, h5py.Dataset, zarr.Array, etc.).""" + + @property + def shape(self) -> tuple[int, ...]: ... + + @property + def dtype(self) -> numpy.dtype[typing.Any]: ... + + def __array__(self, dtype: typing.Any = None, /, *, copy: bool | None = None) -> numpy.typing.NDArray[typing.Any]: ... + + @dataclasses.dataclass class AnnotatedArray: """A numpy array paired with its descriptor and contextual metadata.""" - data: numpy.typing.NDArray[typing.Any] + data: ArrayProtocol descriptor: ArrayDescriptor metadata: ArrayMetadata = dataclasses.field(default_factory=ArrayMetadata) @@ -404,12 +418,16 @@ def __post_init__(self) -> None: if self.descriptor.value_type != inferred_value_type: raise ValueError(f"Descriptor value_type {self.descriptor.value_type!r} does not match inferred value_type {inferred_value_type!r} from array dtype {self.data.dtype}") + def __array__(self, dtype: typing.Any = None, /, *, copy: bool | None = None) -> numpy.typing.NDArray[typing.Any]: + """Allow AnnotatedArray to be passed directly to numpy functions.""" + return numpy.asarray(self.data, dtype=dtype, copy=copy) + @property def header(self) -> ArrayHeader: return ArrayHeader(self.descriptor, self.data.dtype, self.metadata) @classmethod - def from_header(cls, data: numpy.typing.NDArray[typing.Any], header: ArrayHeader) -> AnnotatedArray: + def from_header(cls, data: ArrayProtocol, header: ArrayHeader) -> AnnotatedArray: if header.dtype != data.dtype: raise ValueError(f"Header dtype {header.dtype} does not match array dtype {data.dtype}") if header.descriptor.shape != data.shape: diff --git a/nion/data/test/AnnotatedArray_test.py b/nion/data/test/AnnotatedArray_test.py index 449c6a9..17b95f2 100644 --- a/nion/data/test/AnnotatedArray_test.py +++ b/nion/data/test/AnnotatedArray_test.py @@ -1,7 +1,10 @@ import datetime +import io +import typing import unittest import zoneinfo +import h5py import numpy from nion.data import AnnotatedArray @@ -118,6 +121,48 @@ def test_zeros_annotated_array_constructs_matching_shape_and_dtype(self) -> None self.assertEqual((2, 3), array.data.shape) self.assertEqual(numpy.dtype(numpy.float32), array.header.dtype) + def test_annotated_array_is_numpy_passable(self) -> None: + group = AnnotatedArray.AxisGroup.from_1d_size(4) + array = AnnotatedArray.zeros_annotated_array((group,), dtype=numpy.float64) + + # AnnotatedArray itself is directly usable with numpy functions via __array__ + self.assertEqual(0.0, float(numpy.sum(array))) + as_array = numpy.asarray(array) + self.assertIsInstance(as_array, numpy.ndarray) + self.assertEqual((4,), as_array.shape) + self.assertEqual(numpy.dtype(numpy.float64), as_array.dtype) + + def test_annotated_array_data_is_numpy_passable(self) -> None: + group = AnnotatedArray.AxisGroup.from_2d_size((2, 3)) + array = AnnotatedArray.zeros_annotated_array((group,), dtype=numpy.float32) + + # data satisfies ArrayProtocol including __array__, so it is directly usable with numpy + self.assertEqual(0.0, float(numpy.sum(array.data))) + result = numpy.asarray(array.data) + self.assertIsInstance(result, numpy.ndarray) + self.assertEqual((2, 3), result.shape) + self.assertEqual(numpy.dtype(numpy.float32), result.dtype) + + def test_annotated_array_accepts_h5py_dataset_as_data(self) -> None: + # h5py.Dataset satisfies ArrayProtocol: it exposes .shape, .dtype, and __array__. + # AnnotatedArray must accept it without forcing an eager copy into memory. + buf = io.BytesIO() + with h5py.File(buf, "w") as f: + ds = f.create_dataset("data", data=numpy.arange(6, dtype=numpy.float32).reshape(2, 3)) + group = AnnotatedArray.AxisGroup.from_2d_size((2, 3)) + annotated = AnnotatedArray.AnnotatedArray(ds, AnnotatedArray.ArrayDescriptor((group,))) + + # shape and dtype are read directly from the dataset without materialising it + self.assertEqual((2, 3), annotated.data.shape) + self.assertEqual(numpy.dtype(numpy.float32), annotated.data.dtype) + + # numpy functions work via __array__ (materialises on demand) + result = numpy.asarray(annotated) + self.assertIsInstance(result, numpy.ndarray) + self.assertEqual((2, 3), result.shape) + numpy.testing.assert_array_equal(result, numpy.arange(6, dtype=numpy.float32).reshape(2, 3)) + + if __name__ == "__main__": unittest.main() From 7f50ab83ad68ed90f7c4820c3374c21430d9ff4e Mon Sep 17 00:00:00 2001 From: Chris Meyer <34664+cmeyer@users.noreply.github.com> Date: Fri, 24 Jul 2026 16:27:15 -0700 Subject: [PATCH 13/16] WIP: expand creation routines. --- nion/data/AnnotatedArray.py | 32 ++++++++++++++++++++++----- nion/data/test/AnnotatedArray_test.py | 21 ++++++++++++++++++ 2 files changed, 48 insertions(+), 5 deletions(-) diff --git a/nion/data/AnnotatedArray.py b/nion/data/AnnotatedArray.py index 881ab0e..41337d5 100644 --- a/nion/data/AnnotatedArray.py +++ b/nion/data/AnnotatedArray.py @@ -193,13 +193,33 @@ def __post_init__(self) -> None: raise ValueError("primary_calibration_key must be None when coordinate_calibrations is empty") @staticmethod - def from_1d_size(size: int, *, label: str = "x") -> AxisGroup: - """Create a 1D axis group with no coordinate calibrations.""" - return AxisGroup(axes=(Axis(label, size),)) + def from_1d_size( + size: int, + *, + label: str = "x", + coordinate_calibrations: typing.Mapping[str, CoordinateCalibration] | None = None, + primary_calibration_key: str | None = None, + ) -> AxisGroup: + """Create a 1D axis group with optional coordinate calibrations.""" + calibrations: typing.Mapping[str, CoordinateCalibration] = coordinate_calibrations or {} + + return AxisGroup( + axes=(Axis(label, size),), + coordinate_calibrations=calibrations, + primary_calibration_key=primary_calibration_key, + ) @staticmethod - def from_2d_size(size: tuple[int, int], *, labels: tuple[str, str] = ("x", "y")) -> AxisGroup: - """Create a 2D axis group with no coordinate calibrations.""" + def from_2d_size( + size: tuple[int, int], + *, + labels: tuple[str, str] = ("x", "y"), + coordinate_calibrations: typing.Mapping[str, CoordinateCalibration] | None = None, + primary_calibration_key: str | None = None, + ) -> AxisGroup: + """Create a 2D axis group with optional coordinate calibrations.""" + calibrations: typing.Mapping[str, CoordinateCalibration] = coordinate_calibrations or {} + size_x, size_y = size x_label, y_label = labels return AxisGroup( @@ -207,6 +227,8 @@ def from_2d_size(size: tuple[int, int], *, labels: tuple[str, str] = ("x", "y")) Axis(x_label, size_x), Axis(y_label, size_y), ), + coordinate_calibrations=calibrations, + primary_calibration_key=primary_calibration_key, ) @property diff --git a/nion/data/test/AnnotatedArray_test.py b/nion/data/test/AnnotatedArray_test.py index 17b95f2..378631a 100644 --- a/nion/data/test/AnnotatedArray_test.py +++ b/nion/data/test/AnnotatedArray_test.py @@ -30,6 +30,27 @@ def test_array_descriptor_describes_shape_and_rank(self) -> None: self.assertEqual((3, 4, 5), descriptor.shape) self.assertEqual(3, descriptor.ndim) + def test_axis_group_size_factories_accept_optional_coordinate_calibrations(self) -> None: + single_calibration = AnnotatedArray.CoordinateCalibration(calibrations=(AnnotatedArray.AffineCalibration(unit="nm"),)) + group_1d = AnnotatedArray.AxisGroup.from_1d_size( + 3, + coordinate_calibrations={"spatial": single_calibration}, + primary_calibration_key="spatial", + ) + self.assertEqual(("spatial",), group_1d.calibration_keys) + self.assertEqual("spatial", group_1d.primary_calibration_key) + + map_calibration = AnnotatedArray.CoordinateCalibration( + calibrations=(AnnotatedArray.AffineCalibration(unit="nm"), AnnotatedArray.AffineCalibration(unit="nm")) + ) + group_2d = AnnotatedArray.AxisGroup.from_2d_size( + (2, 3), + coordinate_calibrations={"camera": map_calibration}, + primary_calibration_key="camera", + ) + self.assertEqual(("camera",), group_2d.calibration_keys) + self.assertEqual("camera", group_2d.primary_calibration_key) + def test_array_descriptor_requires_valid_axis_group_layout(self) -> None: with self.assertRaisesRegex(ValueError, "at least one"): AnnotatedArray.ArrayDescriptor(()) From 3e14f9e732019f78c8f476080ebf88f06760a9b3 Mon Sep 17 00:00:00 2001 From: Chris Meyer <34664+cmeyer@users.noreply.github.com> Date: Fri, 24 Jul 2026 17:09:16 -0700 Subject: [PATCH 14/16] WIP: add intensity fallback for no primary key. remove flat axis support. --- nion/data/AnnotatedArray.py | 4 ++-- nion/data/test/AnnotatedArray_test.py | 13 +++++++++++++ 2 files changed, 15 insertions(+), 2 deletions(-) diff --git a/nion/data/AnnotatedArray.py b/nion/data/AnnotatedArray.py index 41337d5..af58c2e 100644 --- a/nion/data/AnnotatedArray.py +++ b/nion/data/AnnotatedArray.py @@ -300,6 +300,8 @@ def ndim(self) -> int: return sum(axis_group.rank for axis_group in self.axis_groups) def get_intensity_calibration(self, key: str | None = None) -> Calibration: + if key is None and self.intensity_calibrations.primary_key is None: + return AffineCalibration() return self.intensity_calibrations.get_calibration(key) def with_primary_intensity_calibration(self, key: str) -> ArrayDescriptor: @@ -462,8 +464,6 @@ def from_header(cls, data: ArrayProtocol, header: ArrayHeader) -> AnnotatedArray def get_intensity_calibration(self, key: str | None = None) -> Calibration: return self.descriptor.get_intensity_calibration(key) - def get_flat_axis_calibrations(self, key: str | None = None) -> list[Calibration]: - return [axis_group.get_calibration(axis=i, key=key) for axis_group in self.descriptor.axis_groups for i in range(axis_group.rank)] def zeros_annotated_array(axis_groups: typing.Sequence[AxisGroup], dtype: numpy.typing.DTypeLike = numpy.float64) -> AnnotatedArray: diff --git a/nion/data/test/AnnotatedArray_test.py b/nion/data/test/AnnotatedArray_test.py index 378631a..b240302 100644 --- a/nion/data/test/AnnotatedArray_test.py +++ b/nion/data/test/AnnotatedArray_test.py @@ -30,6 +30,19 @@ def test_array_descriptor_describes_shape_and_rank(self) -> None: self.assertEqual((3, 4, 5), descriptor.shape) self.assertEqual(3, descriptor.ndim) + def test_array_descriptor_uses_identity_intensity_calibration_when_no_primary_is_designated(self) -> None: + descriptor = AnnotatedArray.ArrayDescriptor((AnnotatedArray.AxisGroup.from_1d_size(3),)) + + calibration = descriptor.get_intensity_calibration() + self.assertIsInstance(calibration, AnnotatedArray.AffineCalibration) + affine_calibration = typing.cast(AnnotatedArray.AffineCalibration, calibration) + self.assertEqual(1.0, affine_calibration.scale) + self.assertEqual(0.0, affine_calibration.offset) + self.assertEqual("", affine_calibration.unit) + + with self.assertRaises(KeyError): + descriptor.get_intensity_calibration("missing") + def test_axis_group_size_factories_accept_optional_coordinate_calibrations(self) -> None: single_calibration = AnnotatedArray.CoordinateCalibration(calibrations=(AnnotatedArray.AffineCalibration(unit="nm"),)) group_1d = AnnotatedArray.AxisGroup.from_1d_size( From bbd6eba0a55d878a29eff1458c8cdc7dcf39d6cc Mon Sep 17 00:00:00 2001 From: Chris Meyer <34664+cmeyer@users.noreply.github.com> Date: Fri, 24 Jul 2026 17:26:26 -0700 Subject: [PATCH 15/16] WIP: add tzlocal dependency; add timezone test. --- nion/data/test/AnnotatedArray_test.py | 9 +++++++++ pyproject.toml | 7 ++++--- test-requirements.txt | 1 + 3 files changed, 14 insertions(+), 3 deletions(-) diff --git a/nion/data/test/AnnotatedArray_test.py b/nion/data/test/AnnotatedArray_test.py index b240302..38448bf 100644 --- a/nion/data/test/AnnotatedArray_test.py +++ b/nion/data/test/AnnotatedArray_test.py @@ -123,6 +123,15 @@ def test_array_metadata_created_retains_iana_zone_and_offset(self) -> None: self.assertEqual("America/Los_Angeles", getattr(metadata.created.tzinfo, "key", None)) self.assertEqual(datetime.timedelta(hours=-7), metadata.created.utcoffset()) + def test_array_metadata_default_created_has_iana_timezone(self) -> None: + # tzlocal.get_localzone() returns a zoneinfo.ZoneInfo with a proper IANA key + # (e.g. "America/New_York"). A plain datetime.timezone fixed-offset object + # would not satisfy the isinstance check and would have no .key attribute. + metadata = AnnotatedArray.ArrayMetadata() + + self.assertIsInstance(metadata.created.tzinfo, zoneinfo.ZoneInfo) + self.assertIsNotNone(metadata.created.tzinfo.key) # type: ignore[union-attr] + def test_array_header_can_be_passed_without_data(self) -> None: descriptor = AnnotatedArray.ArrayDescriptor((AnnotatedArray.AxisGroup.from_2d_size((4, 5)),)) metadata = AnnotatedArray.ArrayMetadata(attributes={"note": "test"}) diff --git a/pyproject.toml b/pyproject.toml index 8e78a8d..4ea2571 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -21,9 +21,10 @@ classifiers = [ ] urls = { "Homepage" = "https://github.com/nion-software/niondata" } dependencies = [ - "nionutils >=4.14.0,<5.0.0", - "numpy >=2.0,<3.0", - "scipy" + "nionutils >=4.14.0,<5.0.0", + "numpy >=2.0,<3.0", + "scipy", + "tzlocal" ] [tool.setuptools.packages.find] diff --git a/test-requirements.txt b/test-requirements.txt index 2b670f8..ef8963d 100755 --- a/test-requirements.txt +++ b/test-requirements.txt @@ -8,3 +8,4 @@ git+https://github.com/nion-software/nionutils.git#egg=nionutils numpy scipy h5py +tzlocal From 7cae419a5e779f5db2a8d3a23c66736d0f48b629 Mon Sep 17 00:00:00 2001 From: Chris Meyer <34664+cmeyer@users.noreply.github.com> Date: Tue, 11 Aug 2026 16:19:12 -0700 Subject: [PATCH 16/16] WIP: Add annotated array to data and metadata conversion functions. --- nion/data/AnnotatedArray.py | 149 +++++++++++++++++++++++- nion/data/test/AnnotatedArray_test.py | 158 ++++++++++++++++++++++++++ 2 files changed, 306 insertions(+), 1 deletion(-) diff --git a/nion/data/AnnotatedArray.py b/nion/data/AnnotatedArray.py index af58c2e..dca12fa 100644 --- a/nion/data/AnnotatedArray.py +++ b/nion/data/AnnotatedArray.py @@ -13,9 +13,12 @@ import datetime import typing import types - import numpy import tzlocal +import zoneinfo + +from nion.data import Calibration as LegacyCalibration +from nion.data import DataAndMetadata class ValueType: @@ -482,3 +485,147 @@ def zeros_annotated_array(axis_groups: typing.Sequence[AxisGroup], dtype: numpy. value_type = infer_value_type(dtype_obj) descriptor = ArrayDescriptor(axis_groups=tuple(axis_groups), value_type=value_type) return AnnotatedArray(data=numpy.zeros(descriptor.shape, dtype=dtype_obj), descriptor=descriptor) + + +def _make_axis_group(shape: tuple[int, ...], dimensional_calibrations: tuple[typing.Any, ...]) -> AxisGroup: + axes = list[Axis]() + coordinate_calibrations = list[Calibration]() + for i, size in enumerate(shape): + calibration = dimensional_calibrations[i] if i < len(dimensional_calibrations) else None + if calibration is None: + affine_calibration = AffineCalibration() + else: + affine_calibration = AffineCalibration(calibration.scale, calibration.offset, calibration.units) + axes.append(Axis(str(i), size)) + coordinate_calibrations.append(affine_calibration) + return AxisGroup( + axes=tuple(axes), + coordinate_calibrations={"calibrated": CoordinateCalibration(calibrations=tuple(coordinate_calibrations))} if coordinate_calibrations else {}, + primary_calibration_key="calibrated" if coordinate_calibrations else None, + ) + + +def _parse_timezone_offset(timezone_offset: str) -> datetime.timezone | None: + if len(timezone_offset) != 5 or timezone_offset[0] not in ("+", "-"): + return None + if not timezone_offset[1:].isdigit(): + return None + hours = int(timezone_offset[1:3]) + minutes = int(timezone_offset[3:5]) + delta = datetime.timedelta(hours=hours, minutes=minutes) + if timezone_offset[0] == "-": + delta = -delta + return datetime.timezone(delta) + + +def _legacy_timestamp_to_created( + timestamp: datetime.datetime, + timezone_name: str | None, + timezone_offset: str | None, +) -> datetime.datetime: + utc_timestamp = timestamp if timestamp.tzinfo else timestamp.replace(tzinfo=datetime.timezone.utc) + + if timezone_name: + try: + return utc_timestamp.astimezone(zoneinfo.ZoneInfo(timezone_name)) + except Exception: + pass + + if timezone_offset: + timezone = _parse_timezone_offset(timezone_offset) + if timezone is not None: + return utc_timestamp.astimezone(timezone) + + return utc_timestamp.astimezone(datetime.timezone.utc) + + +def from_data_and_metadata(xdata: DataAndMetadata.DataAndMetadata) -> AnnotatedArray: + """Convert a legacy DataAndMetadata instance into an AnnotatedArray.""" + data = xdata.data + if data is None: + raise ValueError("DataAndMetadata input data is missing.") + + data_descriptor = xdata.data_descriptor + dimensional_calibrations = tuple(xdata.dimensional_calibrations) + shape = xdata.data_shape + + axis_groups = list[AxisGroup]() + calibration_index = 0 + shape_index = 0 + if data_descriptor.is_sequence: + axis_groups.append(_make_axis_group(shape[shape_index:shape_index + 1], dimensional_calibrations[calibration_index:calibration_index + 1])) + calibration_index += 1 + shape_index += 1 + if data_descriptor.collection_dimension_count: + count = data_descriptor.collection_dimension_count + axis_groups.append(_make_axis_group(shape[shape_index:shape_index + count], dimensional_calibrations[calibration_index:calibration_index + count])) + calibration_index += count + shape_index += count + datum_count = data_descriptor.datum_dimension_count + axis_groups.append(_make_axis_group(shape[shape_index:shape_index + datum_count], dimensional_calibrations[calibration_index:calibration_index + datum_count])) + + intensity_calibration = xdata.intensity_calibration + descriptor = ArrayDescriptor( + axis_groups=tuple(axis_groups), + intensity_calibrations=CalibrationSet.from_calibration( + AffineCalibration(intensity_calibration.scale, intensity_calibration.offset, intensity_calibration.units), + "calibrated", + ), + value_type=infer_value_type(data.dtype), + ) + + created = _legacy_timestamp_to_created(xdata.timestamp, xdata.timezone, xdata.timezone_offset) + metadata = ArrayMetadata(created=created, attributes=dict(xdata.metadata)) + return AnnotatedArray(data=data, descriptor=descriptor, metadata=metadata) + + +def to_data_and_metadata(annotated_array: AnnotatedArray) -> DataAndMetadata.DataAndMetadata: + """Convert an AnnotatedArray into a legacy DataAndMetadata instance.""" + axis_groups = annotated_array.descriptor.axis_groups + if len(axis_groups) == 1: + is_sequence = False + collection_dimension_count = 0 + datum_dimension_count = axis_groups[0].rank + elif len(axis_groups) == 2: + is_sequence = False + collection_dimension_count = axis_groups[0].rank + datum_dimension_count = axis_groups[1].rank + elif len(axis_groups) == 3 and axis_groups[0].rank == 1: + is_sequence = True + collection_dimension_count = axis_groups[1].rank + datum_dimension_count = axis_groups[2].rank + else: + raise ValueError("AnnotatedArray axis groups are incompatible with DataAndMetadata conversion.") + + dimensional_calibrations = list[LegacyCalibration.Calibration]() + for axis_group in axis_groups: + for axis_index in range(axis_group.rank): + calibration = axis_group.get_calibration(axis_index) if axis_group.primary_calibration_key else AffineCalibration() + if not isinstance(calibration, AffineCalibration): + raise ValueError("Only affine coordinate calibrations are supported for DataAndMetadata conversion.") + dimensional_calibrations.append(LegacyCalibration.Calibration(calibration.offset, calibration.scale, calibration.unit)) + + intensity_calibration = LegacyCalibration.Calibration() + if annotated_array.descriptor.primary_intensity_calibration_key: + calibration = annotated_array.get_intensity_calibration() + if not isinstance(calibration, AffineCalibration): + raise ValueError("Only affine intensity calibrations are supported for DataAndMetadata conversion.") + intensity_calibration = LegacyCalibration.Calibration(calibration.offset, calibration.scale, calibration.unit) + + created = annotated_array.metadata.created + timezone_name = typing.cast(str | None, getattr(created.tzinfo, "key", None)) or created.tzname() + timestamp = created.astimezone(datetime.timezone.utc).replace(tzinfo=None) + timezone_offset = created.strftime("%z") + + return DataAndMetadata.new_data_and_metadata( + data=numpy.asarray(annotated_array.data), + intensity_calibration=intensity_calibration, + dimensional_calibrations=tuple(dimensional_calibrations), + metadata=dict(annotated_array.metadata.attributes), + data_descriptor=DataAndMetadata.DataDescriptor(is_sequence, collection_dimension_count, datum_dimension_count), + timestamp=timestamp, + timezone=timezone_name, + timezone_offset=timezone_offset, + ) + + diff --git a/nion/data/test/AnnotatedArray_test.py b/nion/data/test/AnnotatedArray_test.py index 38448bf..d79d188 100644 --- a/nion/data/test/AnnotatedArray_test.py +++ b/nion/data/test/AnnotatedArray_test.py @@ -8,10 +8,27 @@ import numpy from nion.data import AnnotatedArray +from nion.data import Calibration +from nion.data import DataAndMetadata class TestAnnotatedArray(unittest.TestCase): + _descriptor_variants: tuple[tuple[bool, int, int], ...] = ( + (False, 0, 1), + (False, 0, 2), + (False, 1, 1), + (False, 1, 2), + (False, 2, 1), + (False, 2, 2), + (True, 0, 1), + (True, 0, 2), + (True, 1, 1), + (True, 1, 2), + (True, 2, 1), + (True, 2, 2), + ) + def test_calibration_set_uses_explicit_calibration_accessors(self) -> None: primary = AnnotatedArray.AffineCalibration(unit="nm") alternate = AnnotatedArray.AffineCalibration(unit="rad") @@ -205,6 +222,147 @@ def test_annotated_array_accepts_h5py_dataset_as_data(self) -> None: self.assertEqual((2, 3), result.shape) numpy.testing.assert_array_equal(result, numpy.arange(6, dtype=numpy.float32).reshape(2, 3)) + def test_from_data_and_metadata_covers_sequence_collection_and_datum_variations(self) -> None: + def flatten_affine_calibrations(annotated_array: AnnotatedArray.AnnotatedArray) -> tuple[AnnotatedArray.AffineCalibration, ...]: + calibrations = list[AnnotatedArray.AffineCalibration]() + for axis_group in annotated_array.descriptor.axis_groups: + for axis_index in range(axis_group.rank): + calibration = axis_group.get_calibration(axis_index) if axis_group.primary_calibration_key else AnnotatedArray.AffineCalibration() + calibrations.append(typing.cast(AnnotatedArray.AffineCalibration, calibration)) + return tuple(calibrations) + + for is_sequence, collection_rank, datum_rank in self._descriptor_variants: + with self.subTest(is_sequence=is_sequence, collection_rank=collection_rank, datum_rank=datum_rank): + total_rank = (1 if is_sequence else 0) + collection_rank + datum_rank + shape = tuple(range(2, 2 + total_rank)) + data = numpy.arange(int(numpy.prod(shape)), dtype=numpy.float32).reshape(shape) + dimensional_calibrations = tuple(Calibration.Calibration(offset=i + 0.25, scale=i + 1.5, units=f"u{i}") for i in range(total_rank)) + xdata = DataAndMetadata.new_data_and_metadata( + data=data, + intensity_calibration=Calibration.Calibration(offset=7.0, scale=3.0, units="counts"), + dimensional_calibrations=dimensional_calibrations, + metadata={"descriptor_variant": f"legacy-{int(is_sequence)}-{collection_rank}-{datum_rank}"}, + data_descriptor=DataAndMetadata.DataDescriptor(is_sequence, collection_rank, datum_rank), + timestamp=datetime.datetime(2026, 7, 16, 19, 0, 0), + timezone="America/Los_Angeles", + timezone_offset="-0700", + ) + + annotated = AnnotatedArray.from_data_and_metadata(xdata) + expected_axis_group_ranks = (1, datum_rank) if is_sequence and collection_rank == 0 else (1, collection_rank, datum_rank) if is_sequence else (datum_rank,) if collection_rank == 0 else (collection_rank, datum_rank) + self.assertEqual(expected_axis_group_ranks, tuple(axis_group.rank for axis_group in annotated.descriptor.axis_groups)) + self.assertEqual(xdata.data_shape, annotated.descriptor.shape) + self.assertEqual("counts", typing.cast(AnnotatedArray.AffineCalibration, annotated.get_intensity_calibration()).unit) + + flattened_annotated_calibrations = flatten_affine_calibrations(annotated) + self.assertEqual(len(xdata.dimensional_calibrations), len(flattened_annotated_calibrations)) + for i, (legacy_calibration, annotated_calibration) in enumerate(zip(xdata.dimensional_calibrations, flattened_annotated_calibrations)): + with self.subTest(calibration_index=i): + self.assertEqual(legacy_calibration.offset, annotated_calibration.offset) + self.assertEqual(legacy_calibration.scale, annotated_calibration.scale) + self.assertEqual(legacy_calibration.units, annotated_calibration.unit) + + self.assertEqual(xdata.metadata["descriptor_variant"], annotated.metadata.attributes["descriptor_variant"]) + + def test_to_data_and_metadata_covers_sequence_collection_and_datum_variations(self) -> None: + def flatten_affine_calibrations(annotated_array: AnnotatedArray.AnnotatedArray) -> tuple[AnnotatedArray.AffineCalibration, ...]: + calibrations = list[AnnotatedArray.AffineCalibration]() + for axis_group in annotated_array.descriptor.axis_groups: + for axis_index in range(axis_group.rank): + calibration = axis_group.get_calibration(axis_index) if axis_group.primary_calibration_key else AnnotatedArray.AffineCalibration() + calibrations.append(typing.cast(AnnotatedArray.AffineCalibration, calibration)) + return tuple(calibrations) + + for is_sequence, collection_rank, datum_rank in self._descriptor_variants: + with self.subTest(is_sequence=is_sequence, collection_rank=collection_rank, datum_rank=datum_rank): + if is_sequence and collection_rank == 0: + # (1, datum_rank) is ambiguous in legacy form and round-trips as non-sequence collection rank 1. + continue + + expected_axis_group_ranks = (1, collection_rank, datum_rank) if is_sequence else (datum_rank,) if collection_rank == 0 else (collection_rank, datum_rank) + dim_sizes = tuple(range(2, 2 + sum(expected_axis_group_ranks))) + + dim_index = 0 + axis_groups = list[AnnotatedArray.AxisGroup]() + for axis_group_rank in expected_axis_group_ranks: + axes = list[AnnotatedArray.Axis]() + calibrations = list[AnnotatedArray.AffineCalibration]() + for _ in range(axis_group_rank): + size = dim_sizes[dim_index] + calibration_index = dim_index + axes.append(AnnotatedArray.Axis(label=f"a{calibration_index}", size=size)) + calibrations.append(AnnotatedArray.AffineCalibration(offset=calibration_index + 0.5, scale=calibration_index + 1.25, unit=f"ua{calibration_index}")) + dim_index += 1 + axis_groups.append( + AnnotatedArray.AxisGroup( + axes=tuple(axes), + coordinate_calibrations={"calibrated": AnnotatedArray.CoordinateCalibration(calibrations=tuple(calibrations))}, + primary_calibration_key="calibrated", + ) + ) + + descriptor = AnnotatedArray.ArrayDescriptor( + axis_groups=tuple(axis_groups), + intensity_calibrations=AnnotatedArray.CalibrationSet.from_calibration( + AnnotatedArray.AffineCalibration(offset=4.0, scale=2.0, unit="counts"), + "calibrated", + ), + ) + metadata = AnnotatedArray.ArrayMetadata( + created=datetime.datetime(2026, 7, 16, 12, tzinfo=zoneinfo.ZoneInfo("America/Los_Angeles")), + attributes={"descriptor_variant": f"annotated-{int(is_sequence)}-{collection_rank}-{datum_rank}"}, + ) + annotated = AnnotatedArray.AnnotatedArray( + data=numpy.arange(int(numpy.prod(dim_sizes)), dtype=numpy.float32).reshape(dim_sizes), + descriptor=descriptor, + metadata=metadata, + ) + + xdata = AnnotatedArray.to_data_and_metadata(annotated) + self.assertEqual(is_sequence, xdata.is_sequence) + self.assertEqual(collection_rank, xdata.collection_dimension_count) + self.assertEqual(datum_rank, xdata.datum_dimension_count) + self.assertEqual(annotated.descriptor.shape, xdata.data_shape) + self.assertEqual("counts", xdata.intensity_calibration.units) + + flattened_annotated_calibrations = flatten_affine_calibrations(annotated) + self.assertEqual(len(flattened_annotated_calibrations), len(xdata.dimensional_calibrations)) + for i, (annotated_calibration, legacy_calibration) in enumerate(zip(flattened_annotated_calibrations, xdata.dimensional_calibrations)): + with self.subTest(calibration_index=i): + self.assertEqual(annotated_calibration.offset, legacy_calibration.offset) + self.assertEqual(annotated_calibration.scale, legacy_calibration.scale) + self.assertEqual(annotated_calibration.unit, legacy_calibration.units) + + round_tripped = AnnotatedArray.from_data_and_metadata(xdata) + self.assertEqual(tuple(axis_group.rank for axis_group in annotated.descriptor.axis_groups), tuple(axis_group.rank for axis_group in round_tripped.descriptor.axis_groups)) + self.assertEqual(annotated.descriptor.shape, round_tripped.descriptor.shape) + self.assertEqual(annotated.metadata.attributes["descriptor_variant"], round_tripped.metadata.attributes["descriptor_variant"]) + numpy.testing.assert_array_equal(numpy.asarray(annotated.data), numpy.asarray(round_tripped.data)) + + def test_data_and_metadata_timezone_round_trip_through_annotated_array(self) -> None: + xdata = DataAndMetadata.new_data_and_metadata( + data=numpy.arange(6, dtype=numpy.float32).reshape(2, 3), + intensity_calibration=Calibration.Calibration(offset=7.0, scale=3.0, units="counts"), + dimensional_calibrations=( + Calibration.Calibration(offset=0.25, scale=1.5, units="u0"), + Calibration.Calibration(offset=1.25, scale=2.5, units="u1"), + ), + metadata={"descriptor_variant": "legacy-0-1-1"}, + data_descriptor=DataAndMetadata.DataDescriptor(False, 1, 1), + timestamp=datetime.datetime(2026, 7, 16, 19, 0, 0), + timezone="America/Los_Angeles", + timezone_offset="-0700", + ) + + annotated = AnnotatedArray.from_data_and_metadata(xdata) + self.assertEqual("America/Los_Angeles", getattr(annotated.metadata.created.tzinfo, "key", None)) + self.assertEqual(datetime.timedelta(hours=-7), annotated.metadata.created.utcoffset()) + + round_tripped = AnnotatedArray.to_data_and_metadata(annotated) + self.assertEqual(xdata.timestamp, round_tripped.timestamp) + self.assertEqual("America/Los_Angeles", round_tripped.timezone) + self.assertEqual("-0700", round_tripped.timezone_offset) + if __name__ == "__main__":