"""zea data file (HDF5)."""
import contextlib
import difflib
import enum
import inspect
import re
from collections.abc import Mapping
from dataclasses import dataclass, replace
from datetime import datetime, timezone
from pathlib import Path, PurePath
from typing import TYPE_CHECKING, Any, List, Tuple, Union, cast
import h5py
import numpy as np
import zea
from zea import log
from zea.data.legacy_file import legacy_data, legacy_probe, legacy_scan
from zea.data.spec import (
DEFAULT_CHUNK_AXES,
DEFAULT_COMPRESSION,
DataSpec,
FileSpec,
MetadataSpec,
MetricsSpec,
ProbeSpec,
ScanSpec,
)
from zea.internal.checks import _DATA_TYPES, _NON_IMAGE_DATA_TYPES
from zea.internal.preset_utils import HF_PREFIX, _hf_resolve_path, _hf_stream_open
from zea.internal.utils import deprecated
_ZEA_ISSUES_URL = "https://github.com/tue-bmd/zea/issues"
_SNAKE_CASE_RE = re.compile(r"^[a-z_][a-z0-9_]*$")
def _suggest_snake_case(name: str) -> str:
"""Best-effort snake_case suggestion for an error message (not applied automatically)."""
suggestion = re.sub(r"[^0-9a-zA-Z_]+", "_", name).strip("_").lower()
return suggestion or "_"
# h5py.File's own named parameters (everything but its **kwds catch-all). Kept for streamed
# hf:// opens: h5py validates unknown kwargs itself for a plain path, but for a file-like
# object it routes them to the 'fileobj' driver, which silently ignores anything it does not
# recognise (see _set_fapl_fileobj) instead of raising — so a typo like ``version=`` for
# ``revision=`` would otherwise vanish rather than error.
_H5PY_FILE_KWARGS = frozenset(inspect.signature(h5py.File.__init__).parameters) - {
"self",
"name",
"mode",
"kwds",
}
if TYPE_CHECKING:
# ``Self`` is in ``typing`` only from 3.11; import lazily to keep the
# 3.10 floor runtime-clean.
from typing_extensions import Self
from zea.parameters import Parameters
from zea.probes import Probe
[docs]
@dataclass
class CustomElement:
"""Class to store **custom** dataset elements with a name, data, description and unit in the
zea format."""
# The name of the dataset. This will be the key in the group.
name: str
# The data to store in the dataset.
data: np.ndarray
description: str
unit: str
# The group name to store the dataset under. This can be a nested group, e.g.
# "lens/profiles"
group_name: str = ""
def _load_dataset_element_from_group(file, path: str) -> CustomElement:
"""Loads a specific dataset element from a group.
Args:
file (h5py.File): The HDF5 file object.
path (str): The full path to the dataset element.
e.g., "non_standard_elements/lens/lens_profile"
Returns:
CustomElement: The loaded dataset element.
"""
dataset = file[path]
description = dataset.attrs.get("description", "")
unit = dataset.attrs.get("unit", "")
data = dataset[()]
path_parts = path.split("/")
return CustomElement(
name=path_parts[-1],
data=data,
description=description,
unit=unit,
group_name="/".join(path_parts[1:-1]),
)
def _load_custom_elements_from_group(file, path: str) -> List[CustomElement]:
"""Recursively loads additional dataset elements from a group."""
elements = []
for name, item in file[path].items():
if isinstance(item, h5py.Dataset):
elements.append(_load_dataset_element_from_group(file, f"{path}/{name}"))
elif isinstance(item, h5py.Group):
elements.extend(_load_custom_elements_from_group(file, f"{path}/{name}"))
return elements
def _validate_custom_element_naming(element: CustomElement, index: int) -> None:
"""Raise ``ValueError`` if ``element.name`` or any ``group_name`` segment isn't snake_case.
Only called when *saving* new custom elements (:class:`~zea.data.spec.FileSpec`); existing
files with non-conforming names can still be read via :attr:`File.custom`.
"""
if not _SNAKE_CASE_RE.match(element.name):
raise ValueError(
f"custom[{index}].name {element.name!r} is not snake_case "
f"(expected lowercase letters, digits, underscores; e.g. "
f"{_suggest_snake_case(element.name)!r}). This is only enforced when "
f"saving — existing files with non-conforming names can still be read."
)
if element.group_name.startswith("/"):
raise ValueError(
f"custom[{index}].group_name {element.group_name!r} is not snake_case "
f"(group_name must be a relative path and must not start with '/'). This is "
f"only enforced when saving — existing files with non-conforming names can "
f"still be read."
)
for segment in filter(None, element.group_name.split("/")):
if not _SNAKE_CASE_RE.match(segment):
raise ValueError(
f"custom[{index}].group_name segment {segment!r} (in "
f"{element.group_name!r}) is not snake_case (e.g. "
f"{_suggest_snake_case(segment)!r}). This is only enforced when "
f"saving — existing files with non-conforming names can still be read."
)
def _custom_elements_relative(elements: List[CustomElement], prefix: str) -> List[CustomElement]:
"""Elements under ``prefix``, with ``group_name`` rewritten relative to it.
Used to build a nested :class:`CustomElements` view, so that slicing by the leading path
segment composes correctly at every nesting level.
"""
result = []
for element in elements:
if element.group_name == prefix:
result.append(replace(element, group_name=""))
elif element.group_name.startswith(prefix + "/"):
result.append(replace(element, group_name=element.group_name[len(prefix) + 1 :]))
return result
[docs]
class CustomElements:
"""A :class:`list` of :class:`CustomElement` with name/group based dict- and dot-access.
Behaves like ``list[CustomElement]`` — iteration, ``len()``, positional indexing/slicing,
and equality with a plain list all work as before. On top of that, elements can be looked
up by name, and nested ``group_name`` groups can be walked into with dict- or dot-style
access::
f.custom["lens_correction"] # by name
f.custom.lens_correction # same, as an attribute
f.custom["lens"]["profile"] # into a nested group_name
f.custom.lens.profile # same
f.custom["lens/profile"] # slash-path shortcut for the above
Dot access only works for names that are valid Python identifiers (snake_case is
recommended, and enforced when *saving* new files — see :class:`CustomElement`); names
with spaces or other characters are still reachable via ``[...]`` bracket access.
Note:
This wraps rather than subclasses :class:`list`, so a custom element named ``"keys"``
shadows that method for attribute access — use ``f.custom["keys"]`` in that case.
"""
__slots__ = ("_elements",)
def __init__(self, elements):
object.__setattr__(self, "_elements", list(elements))
def __len__(self) -> int:
return len(self._elements)
def __iter__(self):
return iter(self._elements)
def __bool__(self) -> bool:
return bool(self._elements)
def __eq__(self, other) -> bool:
if isinstance(other, CustomElements):
return self._elements == other._elements
if isinstance(other, list):
return self._elements == other
return NotImplemented
def __repr__(self) -> str:
return repr(self._elements)
def __getitem__(self, key):
if isinstance(key, (int, slice)):
return self._elements[key]
if isinstance(key, str):
if "/" in key:
result = self
for part in key.split("/"):
result = result[part]
return result
return self._lookup(key)
raise TypeError(f"CustomElements indices must be int, slice or str, not {type(key)}")
def __getattr__(self, name: str):
try:
return self._lookup(name)
except KeyError as e:
raise AttributeError(str(e)) from e
def _lookup(self, name: str):
for element in self._elements:
if element.group_name == "" and element.name == name:
return element
group = _custom_elements_relative(self._elements, name)
if group:
return CustomElements(group)
raise KeyError(
f"No custom element or group named {name!r}. Available: {sorted(self.keys())}"
)
[docs]
def keys(self):
"""Top-level names: leaf element names plus immediate sub-group names."""
return {
element.name if element.group_name == "" else element.group_name.split("/")[0]
for element in self._elements
}
class _StringDataset:
"""Thin wrapper around an h5py string Dataset that auto-decodes bytes on read.
h5py returns variable-length or fixed-length string datasets as ``bytes``
objects. This wrapper intercepts ``__getitem__`` and converts any bytes
elements to ``str`` so callers always receive a ``numpy.ndarray`` with
``dtype=numpy.str_``.
"""
__slots__ = ("_ds",)
def __init__(self, ds: h5py.Dataset):
self._ds = ds
def __getitem__(self, key):
data = self._ds[key]
if isinstance(data, np.ndarray):
decoded = np.array(
[v.decode() if isinstance(v, bytes) else str(v) for v in data.flat],
dtype=np.str_,
).reshape(data.shape)
return decoded
return data.decode() if isinstance(data, bytes) else str(data)
def __getattr__(self, name: str):
return getattr(self._ds, name)
def __len__(self):
return len(self._ds)
def __repr__(self):
return f"<StringDataset shape={self._ds.shape} dtype=str>"
[docs]
class ChunkedDataset:
"""An ``h5py.Dataset`` whose reads go through :mod:`zea.data.chunk_reader`.
Behaves exactly like the dataset it wraps, except that slicing fetches the chunk byte
ranges itself and decodes them concurrently; reads the fast path does not understand fall
back to h5py, so the result is always what h5py would have returned.
Handed out by :class:`_GroupProxy`, so ``file.data.raw_data[0:8]`` is already on the fast
path. ``file["data/raw_data"]`` still returns the bare ``h5py.Dataset``.
"""
__slots__ = ("_dset", "_fetcher")
def __init__(self, dset: h5py.Dataset, fetcher):
self._dset = dset
self._fetcher = fetcher
def __getitem__(self, selection):
from zea.data.chunk_reader import read
progress = getattr(self._fetcher, "progress", False)
return read(self._dset, selection, self._fetcher, progress)
def __setitem__(self, selection, value):
# Writes never take the read fast path; forward straight to the underlying dataset
# so ``file.data.raw_data[...] = value`` keeps behaving like a bare h5py.Dataset.
self._dset[selection] = value
def __getattr__(self, name: str):
return getattr(self._dset, name)
def __array__(self, dtype=None, copy=None):
values = self[...]
return values if dtype is None else values.astype(dtype)
def __len__(self):
return len(self._dset)
def __iter__(self):
for index in range(len(self._dset)):
yield self[index]
def __repr__(self):
return f"<ChunkedDataset {self._dset.name} shape={self.shape} dtype={self.dtype}>"
def _format_selection(selection) -> str:
"""Formats a __getitem__ argument as it would appear in source, e.g. slice(None) -> ':'."""
def _format_one(part) -> str:
if isinstance(part, slice):
start = "" if part.start is None else part.start
stop = "" if part.stop is None else part.stop
if part.step is None:
return f"{start}:{stop}"
return f"{start}:{stop}:{part.step}"
if part is Ellipsis:
return "..."
return repr(part)
if isinstance(selection, tuple):
return ", ".join(_format_one(part) for part in selection)
return _format_one(selection)
class _GroupProxy:
"""Lazy proxy for an h5py.Group that exposes children as attributes.
Datasets are returned as :class:`ChunkedDataset` – they slice like an
``h5py.Dataset`` (nothing is loaded until you index them), but the read goes
through :mod:`zea.data.chunk_reader`, which fetches the chunks concurrently::
with File(path) as f:
# returns ChunkedDataset – no data loaded yet
f.data.raw_data
# slicing triggers the actual read, just like plain h5py
f.data.raw_data[:, :n_tx]
# nested groups work too
f.data.image.values[0]
String datasets are automatically wrapped in :class:`_StringDataset` so
slicing always returns ``numpy.ndarray`` with ``dtype=numpy.str_`` rather
than raw ``bytes``.
"""
__slots__ = ("_group", "_fetcher")
def __init__(self, group: h5py.Group, fetcher=None):
self._group = group
self._fetcher = fetcher
def __getattr__(self, name: str):
try:
child = self._group[name]
except KeyError:
raise AttributeError(
f"No key '{name}' in group '{self._group.name}'. "
f"Available keys: {list(self._group.keys())}"
)
if isinstance(child, h5py.Group):
return _GroupProxy(child, self._fetcher)
if isinstance(child, h5py.Dataset) and h5py.check_string_dtype(child.dtype):
return _StringDataset(child)
return ChunkedDataset(child, self._fetcher)
def __dir__(self):
return list(self._group.keys())
def __repr__(self):
return repr(self._group)
def keys(self):
"""Return the keys of the underlying group."""
return self._group.keys()
def __contains__(self, key):
return key in self._group
def __iter__(self):
return iter(self._group)
def __getitem__(self, selection):
attr_path = "f." + self._group.name.strip("/").replace("/", ".")
hint = ""
if "values" in self._group:
hint = f" Did you mean '{attr_path}.values[{_format_selection(selection)}]'?"
raise TypeError(
f"'{attr_path}' is a group, not a dataset, so it can't be sliced directly."
f"{hint} Available keys: {list(self._group.keys())}"
)
[docs]
def assert_key(file: h5py.File, key: str):
"""Asserts key is in a h5py.File."""
if key not in file.keys():
raise KeyError(f"{key} not found in file")
if TYPE_CHECKING:
from typing import Iterator
class _NdArrayDataset:
"""TYPE_CHECKING stub for h5py.Dataset that exposes np.ndarray on indexing.
At runtime these are plain ``h5py.Dataset`` objects; this stub exists so
the type checker knows that ``dataset[()]`` / ``dataset[0]`` returns
``np.ndarray`` rather than ``Unknown`` (h5py ships no PEP 561 stubs).
"""
shape: tuple[int, ...]
dtype: np.dtype
ndim: int
size: int
def __getitem__(self, args: object) -> np.ndarray: ...
def __len__(self) -> int: ...
def __iter__(self) -> Iterator[np.ndarray]: ...
class _SpatialMapProxy(_GroupProxy):
"""TYPE_CHECKING view of an HDF5 spatial-map group (values + optional metadata).
Exposed via ``File.data.<map_name>`` (e.g. ``f.data.image``,
``f.data.segmentation``). At runtime these are plain ``_GroupProxy``
objects; this class exists solely so the IDE resolves leaf datasets.
"""
values: _NdArrayDataset
coordinates: _NdArrayDataset
labels: _StringDataset
description: _NdArrayDataset
unit: _NdArrayDataset
class _DataProxy(_GroupProxy):
"""TYPE_CHECKING view of the HDF5 ``data/`` group in a :class:`File`.
All known :class:`~zea.data.spec.DataSpec` fields are declared here so
that ``f.data.raw_data``, ``f.data.segmentation.values``, etc. resolve
to concrete types. At runtime ``File.data`` returns a plain
``_GroupProxy``; this class exists only for the IDE/type checker.
"""
raw_data: _NdArrayDataset
aligned_data: _SpatialMapProxy
beamformed_data: _SpatialMapProxy
envelope_data: _SpatialMapProxy
image: _SpatialMapProxy
segmentation: _SpatialMapProxy
sos_map: _SpatialMapProxy
attenuation_map: _SpatialMapProxy
strain_percentage_map: _SpatialMapProxy
shear_wave_elastography_map: _SpatialMapProxy
tissue_doppler: _SpatialMapProxy
color_doppler: _SpatialMapProxy
[docs]
class Track:
"""A single acquisition track within a :class:`File`.
Provides the same ``.data``, ``.scan`` and ``.load_parameters()`` interface
as :class:`File` but scoped to one ``tracks/track_N`` group. Obtain
instances through :attr:`File.tracks` rather than constructing this class
directly.
Example::
with File("multi_track.hdf5") as f:
for track in f.tracks:
raw = track.data.raw_data[:]
parameters = track.load_parameters()
"""
__slots__ = ("_index", "_group", "_timestamps", "_label", "_probe", "_fetcher")
# Declared for type checkers only: the values live in the slots above and
# are populated in __init__ via ``object.__setattr__`` (Track is immutable).
_index: int
_group: "h5py.Group"
_timestamps: "np.ndarray | None"
_label: "str | None"
_probe: "dict | None"
_fetcher: Any
def __init__(
self,
index: int,
group: "h5py.Group",
timestamps: "np.ndarray | None" = None,
label: "str | None" = None,
probe: "dict | None" = None,
fetcher: Any = None,
):
object.__setattr__(self, "_index", index)
object.__setattr__(self, "_group", group)
object.__setattr__(self, "_timestamps", timestamps)
object.__setattr__(self, "_label", label)
object.__setattr__(self, "_probe", probe)
object.__setattr__(self, "_fetcher", fetcher)
@property
def data(self) -> "_DataProxy":
"""Lazy proxy for this track's ``data`` group."""
if "data" not in self._group:
raise KeyError(
f"Track {self._index} has no 'data' group. "
f"Available keys: {list(self._group.keys())}"
)
return cast("_DataProxy", _GroupProxy(self._group["data"], self._fetcher))
@property
def scan(self) -> "ScanSpec":
"""Return the validated :class:`~zea.data.spec.ScanSpec` for this track.
This is the bare scan group as a spec object. For a full, derivable
parameter object (merged probe + scan) use :meth:`load_parameters`.
"""
if "scan" not in self._group:
raise KeyError(
f"Track {self._index} has no 'scan' group. "
f"Available keys: {list(self._group.keys())}"
)
scan_dict = load_dict_from_hdf5_group(self._group["scan"])
return ScanSpec(**scan_dict)
@property
def n_frames(self) -> int:
"""Number of frames."""
return _shape_from_data_group(
self._group["data"],
index=0,
name="n_frames",
requires_raw=False,
)
@property
def n_tx(self) -> int:
"""Number of transmit events."""
return _shape_from_data_group(
self._group["data"],
index=1,
name="n_tx",
requires_raw=True,
)
@property
def n_ax(self) -> int:
"""Number of axial samples."""
return _shape_from_data_group(
self._group["data"],
index=2,
name="n_ax",
requires_raw=True,
)
@property
def n_el(self) -> int:
"""Number of elements."""
return _shape_from_data_group(
self._group["data"],
index=3,
name="n_el",
requires_raw=True,
)
[docs]
def load_parameters(self, **overrides) -> "Parameters":
"""Load this track's parameters (merged probe + scan) as :class:`~zea.Parameters`.
Each track shares the same probe but has its own scan, so the returned
object has the same shape as :meth:`File.load_parameters` for a
single-track file.
Args:
**overrides: Override any parameter.
Returns:
Parameters: Initialised parameters object for this track.
"""
from zea.parameters import Parameters
if "scan" not in self._group:
raise KeyError(
f"Track {self._index} has no 'scan' group. "
f"Available keys: {list(self._group.keys())}"
)
scan = self.scan
scan_dict = scan.to_dict()
other_dict = {
"n_ax": self.n_ax,
"n_el": scan.n_el,
"n_tx": scan.n_tx,
}
merged_dict = {**(self._probe or {}), **scan_dict, **other_dict, **overrides}
return Parameters(**merged_dict)
@property
def label(self) -> "str | None":
"""Human-readable name for this track (e.g. ``'focused'`` or ``'planewave'``).
Returns ``None`` for single-track files or legacy files written without a label.
Use :attr:`File.track_labels` to print all labels in acquisition order and
:meth:`File.get_track` to retrieve a track by name.
"""
return self._label
@property
def timestamps(self) -> "np.ndarray | None":
"""Global transmit timestamps for this track, shape ``(n_frames, n_tx)``.
Timestamps are pre-computed when the :class:`Track` is created via
:attr:`File.tracks`. Returns ``None`` if the file has no
``track_schedule`` or any track is missing ``time_to_next_transmit``.
"""
return self._timestamps
def __repr__(self) -> str:
label_part = f' "{self._label}"' if self._label is not None else ""
keys = list(self._group.get("data", {}).keys())
return f"<Track[{self._index}]{label_part} data={keys}>"
[docs]
def load_dict_from_hdf5_group(group: "h5py.Group") -> dict:
"""Recursively load the contents of an HDF5 group into a plain dict.
Datasets are returned as numpy arrays or scalars; nested groups are
converted recursively. String datasets are decoded to ``np.str_``.
Args:
group: An open :class:`h5py.Group` (or :class:`h5py.File`).
Returns:
dict: Nested dictionary mirroring the group structure.
"""
ans = {}
for key, item in group.items():
if isinstance(item, h5py.Dataset):
if h5py.check_string_dtype(item.dtype) is not None:
val = item.asstr()[()]
if isinstance(val, np.ndarray) and val.dtype == object:
val = val.astype(np.str_)
ans[key] = val
else:
ans[key] = item[()]
elif isinstance(item, h5py.Group):
ans[key] = load_dict_from_hdf5_group(item)
return ans
def _get_data_array_shape(data_group: "h5py.Group") -> "tuple[tuple | None, bool]":
"""Return the shape one of the data arrays in *data_group*.
Checks flat datasets first (e.g., ``raw_data``).
Then looks for the first spatial map group (e.g., ``image``) containing a ``values``
dataset and returns its shape.
Returns ``None`` if neither is found.
"""
flat_datasets = [k for k in data_group if isinstance(data_group[k], h5py.Dataset)]
spatial_map_groups = [k for k in data_group if isinstance(data_group[k], h5py.Group)]
has_raw_data = False
# first check for one of the recognized data arrays
for key in _DATA_TYPES:
if key in flat_datasets:
if key == "raw_data":
has_raw_data = True
return data_group[key].shape, has_raw_data
if key in spatial_map_groups:
group = data_group[key]
if "values" in group and isinstance(group["values"], h5py.Dataset):
return group["values"].shape, has_raw_data
# if none of the arrays are found under their expected names,
# look for any other dataset
for key in flat_datasets:
return data_group[key].shape, has_raw_data
for key in spatial_map_groups:
group = data_group[key]
if "values" in group and isinstance(group["values"], h5py.Dataset):
return group["values"].shape, has_raw_data
return None, has_raw_data
def _shape_from_data_group(
data_group: "h5py.Group", index: int, name: str = "", requires_raw: bool = False
) -> int:
shape, has_raw_data = _get_data_array_shape(data_group)
if requires_raw and not has_raw_data:
raise TypeError(f"`{name}` is only available if the file contains a `raw_data` dataset")
if shape is None:
raise TypeError(
f"Cannot determine `{name}`, no recognized data arrays found in the data group."
) from None
return shape[index]
def _compute_all_track_timestamps(
schedule: "np.ndarray",
tracks: "list[Track]",
) -> "list[np.ndarray | None]":
"""Compute and return timestamps for every track given a track schedule.
Walks the schedule once, keeping a single scalar cumulative timestamp and
a per-track ``[frame_idx, tx_idx]`` counter to index into each track's
``time_to_next_transmit`` array. The schedule must cover exactly
``sum(n_frames_t * n_tx_t)`` events across all tracks.
Args:
schedule: ``int32`` array mapping each global transmit event to a track
index, shape ``(n_total_tx,)``.
tracks: :class:`Track` list (without timestamps yet assigned).
Returns:
list: One ``np.ndarray`` of shape ``(n_frames_t, n_tx_t)`` per track,
or a list of ``None`` values if timestamps cannot be computed.
"""
n_tracks = len(tracks)
t2nts: list[np.ndarray] = []
# pre-load time_to_next_transmit for each track,
# validating that it's present and has the right shape
for track in tracks:
t2nt = track.scan.time_to_next_transmit
if t2nt is None:
log.warning(
f"Track {track._index} has no 'time_to_next_transmit';"
" cannot compute track timestamps."
)
return [None] * n_tracks
t2nts.append(np.asarray(t2nt, dtype=np.float32))
n_frames_per_track = [t2nt.shape[0] for t2nt in t2nts]
n_tx_per_frame_per_track = [t2nt.shape[1] for t2nt in t2nts]
# results will be stored here as we walk the schedule
timestamp_matrices_per_track: "list[np.ndarray | None]" = [
np.zeros_like(t2nt) for t2nt in t2nts
]
# counters to keep track of where we are in each track's
# timestamp matrix.
track_counters = [[0, 0] for _ in tracks] # [frame_idx, tx_idx] per track
cumulative_timestamp = 0.0
# walk through the schedule, filling in the timestamp matrices as we go
for track_idx in schedule:
frame_idx, tx_idx = track_counters[track_idx]
timestamp_matrices_per_track[track_idx][frame_idx, tx_idx] = cumulative_timestamp
cumulative_timestamp += float(t2nts[track_idx][frame_idx, tx_idx])
# update the counters keeping track of where we are in this track's timestamp matrix
tx_idx += 1
if tx_idx >= n_tx_per_frame_per_track[track_idx]:
tx_idx = 0
frame_idx += 1
track_counters[track_idx] = [frame_idx, tx_idx]
for i, (frame_idx, _) in enumerate(track_counters):
assert frame_idx == n_frames_per_track[i], (
f"There was a mismatch between the track_schedule and the number of frames and "
f"transmits in track {i}. "
f"Please ensure that the track_schedule correctly maps to the global number of "
f"transmit events."
)
return timestamp_matrices_per_track
def _parse_version(v: str) -> tuple[int, ...]:
return tuple(int(p) for p in v.split(".")[:3] if p.isdigit())
def _warn_if_legacy_file(file: "File") -> None:
"""Warn if *file* has no zea_version or was written before v0.1.0."""
version = file.attrs.get("zea_version", None)
if version is None or _parse_version(version) < (0, 1, 0):
legacy_version = version if version is not None else "<0.1.0"
log.warning_once(
f"This ``zea.File`` '{file.filename}' was created with a legacy version of "
f"zea ({legacy_version}), while you are using zea v{zea.__version__}. "
"It may behave in unexpected ways. Install an earlier version of zea<0.1.0 for full "
"compatibility or re-save the file with zea v0.1.0 or later (e.g. via File.create).",
key=file.filename,
)
def _validate_custom_key_naming(data: dict, metadata: dict) -> None:
"""Raise ``ValueError`` for non-snake_case custom keys in data/metadata dicts.
Only called when *saving* (from :meth:`File.create`); existing files with non-conforming
custom keys can still be read.
"""
for label, mapping, schema in (
("data", data, DataSpec.SCHEMA),
("metadata", metadata, MetadataSpec.SCHEMA),
):
for key in mapping:
if key not in schema and not _SNAKE_CASE_RE.match(key):
raise ValueError(
f"Custom {label!r} key {key!r} is not snake_case (expected lowercase "
f"letters, digits, underscores; e.g. {_suggest_snake_case(key)!r}). "
f"This is only enforced when saving — existing files with "
f"non-conforming keys can still be read."
)
def _warn_custom_keys(data: dict, metadata: dict):
"""Warn about custom keys in data/metadata dicts when saving."""
known_map_keys = [k for k, v in DataSpec.SCHEMA.items() if "spec" in v]
custom_maps = [k for k in data if k not in DataSpec.SCHEMA]
if custom_maps:
parts = [
f"Custom key(s) added to 'data' and validated as generic Map specs: "
f"{', '.join(sorted(custom_maps))}."
]
for key in sorted(custom_maps):
close = difflib.get_close_matches(key, known_map_keys, n=1, cutoff=0.6)
if close:
parts.append(
f" '{key}' closely resembles the built-in field '{close[0]}' — "
f"did you mean '{close[0]}'?"
)
parts.append(f"Supported data fields: {', '.join(known_map_keys)}.")
parts.append(
f"Think one of your keys should be a recognized field? Open an issue: {_ZEA_ISSUES_URL}"
)
log.warning("\n".join(parts))
known_signal_keys = [k for k, v in MetadataSpec.SCHEMA.items() if "spec" in v]
custom_signals = [k for k in metadata if k not in MetadataSpec.SCHEMA]
if custom_signals:
parts = [
f"Custom key(s) added to 'metadata' and validated as generic SignalND specs: "
f"{', '.join(sorted(custom_signals))}."
]
for key in sorted(custom_signals):
close = difflib.get_close_matches(key, known_signal_keys, n=1, cutoff=0.6)
if close:
parts.append(
f" '{key}' closely resembles the built-in field '{close[0]}' — "
f"did you mean '{close[0]}'?"
)
parts.append(f"Supported metadata fields: {', '.join(known_signal_keys)}.")
parts.append(
f"Think one of your keys should be a recognized field? Open an issue: {_ZEA_ISSUES_URL}"
)
log.warning("\n".join(parts))
[docs]
class File(h5py.File):
"""File handler for ``zea`` formatted ultrasound files. Extends the h5py.File class.
Has functionality to read and write zea files, as well as to access the data, parameters
and metadata in the file.
In contrast to `h5py.File`, this class can read and decompress chunks concurrently
for some codecs (e.g. the zea default). If it cannot use the fast path, it will fallback to
the standard `h5py.File` behavior, and will throw a relevant warning to point the user
to the issue. See :mod:`zea.data.chunk_reader` for details.
"""
def __init__(self, name, mode="r", *args, **kwargs):
"""Initialize the file.
Args:
name (str, Path, HFPath): The path to the file.
Can be a string or a Path object. Additionally can be a string with
the prefix 'hf://', in which case it will be resolved to a
huggingface path.
mode (str, optional): The mode to open the file in. Defaults to "r".
stream (bool, optional): Only used when ``name`` starts with ``hf://``.
When ``True`` (the default for ``hf://`` paths), the file is read
lazily over HTTP range requests via
:class:`~huggingface_hub.HfFileSystem` — only the byte ranges
actually accessed (e.g. the frames you slice) are fetched, instead
of downloading the whole file. Requires read mode (``"r"``). Pass
``stream=False`` to download the full file to the local cache
(the previous behaviour), which is required for writing.
revision (str, optional): HuggingFace revision (branch, tag, or commit hash)
to read from. Only used when ``name`` starts with ``hf://``.
Defaults to the repository default branch. Example: ``revision="v0.1.0"``.
repo_type (str, optional): HuggingFace repository type. Only used when
``name`` starts with ``hf://``. Defaults to ``"dataset"``.
cache_dir (str or Path, optional): Local cache directory for downloaded
HuggingFace files. Only used when ``name`` starts with ``hf://`` and
``stream=False``.
block_size (int, optional): Block size in bytes for streaming reads.
zea files store each frame as many small chunks, so a larger block
coalesces more chunk reads into a single HTTP request — faster for
whole-frame reads, at the cost of more over-fetch for sparse reads.
Only used when ``name`` starts with ``hf://`` and ``stream=True``.
cache_type (str, optional): fsspec cache strategy for streaming reads
(default ``"blockcache"``). Only used when ``name`` starts with
``hf://`` and ``stream=True``.
progress (bool or callable, optional): Report progress while streaming chunks.
Also settable after opening (``file.progress = True``). Defaults to ``True``
for streamed files, ``False`` otherwise. Reads that fall back to h5py
(an lzf file, a strided selection) report no per-chunk progress.
cache (bool, optional): Cache fetched/streamed chunks on disk so a repeated read is
served locally instead of re-downloaded (default ``True``, streamed files only; see
:mod:`zea.data.chunk_cache`).
*args: Additional arguments to pass to h5py.File.
**kwargs: Additional keyword arguments to pass to h5py.File.
"""
# First check if the file is an HDF5 file
assert str(name).endswith(".hdf5") or str(name).endswith(".h5"), (
"File must be an HDF5 file with .hdf5 or .h5 extension."
)
# Extract HF-only kwargs so they never reach h5py
hf_kwargs: dict[str, Any] = {}
for key in ("revision", "repo_type", "cache_dir", "block_size", "cache_type"):
if key in kwargs:
hf_kwargs[key] = kwargs.pop(key)
is_hf = str(name).startswith(HF_PREFIX)
# Streaming (HTTP range requests) is the default for hf:// paths: only the
# bytes actually read are fetched, rather than downloading the whole file.
stream = kwargs.pop("stream", None)
if stream is None:
stream = is_hf
elif stream and not is_hf:
raise ValueError("stream=True is only supported for 'hf://' paths.")
# Progress reporting for concurrent chunk reads. Defaults to True for streamed
# (hf://) files, where a bar communicates real network wait; a local file's
# concurrent reads are fast enough (pread, no round trips) that a bar is just
# noise by default.
progress = kwargs.pop("progress", stream)
# Cache streamed chunks on disk (see zea.data.chunk_cache). On by default, like the
# HF hub cache; ``cache=False`` (or ZEA_CHUNK_CACHE=0) re-fetches every time.
cache_chunks = kwargs.pop("cache", True)
# File object opened for streaming; kept so we can close it in close().
stream_fileobj = None
# Original source path, retained for streamed files whose ``filename`` is
# just a placeholder for the underlying file object (see ``path``).
source_name = None
# Resolve huggingface path
if is_hf and stream:
if mode != "r":
raise ValueError(
"Streaming hf:// files is only supported in read mode ('r'). "
"Pass stream=False to download the file for other modes."
)
# A file-like object routes through h5py's 'fileobj' driver, which silently drops
# any kwarg it does not recognise (see _H5PY_FILE_KWARGS) rather than raising —
# so check now, before that swallows a typo like version= for revision=.
unknown = set(kwargs) - _H5PY_FILE_KWARGS
if unknown:
raise TypeError(
f"'{next(iter(unknown))}' is an invalid keyword argument for zea.File."
)
# cache_dir only applies to full downloads; it is irrelevant here.
hf_kwargs.pop("cache_dir", None)
source_name = str(name)
stream_fileobj = _hf_stream_open(str(name), **hf_kwargs)
# Serve the HDF5 metadata h5py walks on open from the cache too, not just the
# data chunks: it is the same file, and re-opening otherwise re-fetches it every
# time (~0.4 s, and ~3 s whenever HF's CDN stalls on a cold object).
if cache_chunks:
from zea.data.chunk_cache import CachedFile, cache_for
chunk_cache = cache_for(getattr(stream_fileobj, "details", None) or {})
if chunk_cache is not None:
stream_fileobj = CachedFile(stream_fileobj, chunk_cache)
name = stream_fileobj
elif is_hf:
# Full download (previous behaviour); block_size/cache_type are
# streaming-only.
hf_kwargs.pop("block_size", None)
hf_kwargs.pop("cache_type", None)
name = _hf_resolve_path(str(name), **hf_kwargs)
# Disable locking for read mode by default. Locking is meaningless for an
# in-memory/streamed file object, so only apply it to real filesystem paths.
if stream_fileobj is None and "locking" not in kwargs and mode == "r":
# If the file is opened in read mode, disable locking
kwargs["locking"] = False
# Initialize the h5py.File
try:
super().__init__(name, mode, *args, **kwargs)
except Exception:
if stream_fileobj is not None:
stream_fileobj.close()
raise
self._stream_fileobj = stream_fileobj
self._source_name = source_name
self._hf_kwargs = hf_kwargs
# Source of raw chunk bytes for the concurrent read path, built on first use (it
# opens a file descriptor / an HTTP session, which a write-only user never needs).
self._fetcher: Any = None
self._fetcher_built = False
self._progress = progress
self._cache_chunks = cache_chunks
# Warn when opening an existing file that pre-dates zea v0.1.0
if mode in ("r", "r+"):
_warn_if_legacy_file(self)
@property
def progress(self):
"""Whether chunk reads report progress (see :func:`zea.data.chunk_reader.read`).
``True`` draws a tqdm bar over the compressed bytes of each read; a callable is
invoked with each chunk's size as it lands. Settable after opening::
with File("hf://...") as file:
file.progress = True
file.data.raw_data[:5]
Reads that fall back to h5py (an lzf file, a strided selection) report no per-chunk
progress — h5py fetches the whole selection in one opaque call.
"""
return self._progress
@progress.setter
def progress(self, value):
self._progress = value
if self._fetcher_built and self._fetcher is not None:
self._fetcher.progress = value
@property
def _chunk_fetcher(self):
"""Fetcher backing :class:`ChunkedDataset` reads, or ``None`` when unavailable.
``None`` means every dataset of this file falls back to plain h5py reads — which is
correct, just not concurrent.
"""
if not self._fetcher_built:
from zea.data.chunk_reader import fetcher_for
self._fetcher_built = True
try:
self._fetcher = fetcher_for(self)
if self._fetcher is not None:
self._fetcher.progress = self._progress
except Exception as exc: # noqa: BLE001 — a fast path is never worth an error
log.debug(f"Concurrent chunk reads unavailable for '{self.path}': {exc}")
self._fetcher = None
return self._fetcher
def __enter__(self) -> "Self":
"""Enter the context manager, returning this :class:`File` instance.
Overrides ``h5py.File.__enter__`` purely to narrow the return type so
that ``with File(...) as f:`` binds ``f`` to :class:`File` (preserving
access to zea-specific properties like :attr:`data`, :attr:`metadata`
and :meth:`load_parameters`) rather than the base ``h5py`` type.
"""
return self
[docs]
def close(self):
"""Close the file, also releasing the streamed HF file object if any.
``h5py`` does not close file-like objects it did not open itself, so when
this :class:`File` was opened from an ``hf://`` path with ``stream=True``
we close the underlying fsspec file object here.
"""
try:
super().close()
finally:
fetcher = getattr(self, "_fetcher", None)
if fetcher is not None:
fetcher.close()
self._fetcher = None
stream_fileobj = getattr(self, "_stream_fileobj", None)
if stream_fileobj is not None:
stream_fileobj.close()
self._stream_fileobj = None
def __contains__(self, key):
"""Check whether *key* exists in the file.
Extends the h5py default to also match legacy short-form keys
(``"scan"``, ``"data"``) against the tracks layout
(``tracks/track_0/scan``, ``tracks/track_0/data``), including
sub-paths like ``"data/segmentation"``.
"""
if super().__contains__(key):
return True
# Handle both "data" and "data/..." paths — only remap for single-track files.
parts = key.split("/", 1)
if parts[0] in ("scan", "data"):
if super().__contains__("tracks"):
if len(self["tracks"]) == 1:
remapped = f"tracks/track_0/{key}"
return super().__contains__(remapped)
else:
log.warning(
f"Multiple tracks found; Try accessing '{key}' on a specific track instead."
)
return False
return False
def __getitem__(self, name):
"""Open an object in the file.
Extends the h5py default to redirect ``"data"`` and ``"scan"`` (and
sub-paths like ``"data/segmentation"``) to the tracks layout for
single-track new-format files. Multi-track files raise :exc:`AttributeError`.
"""
parts = name.split("/", 1)
if parts[0] in ("data", "scan") and not super().__contains__(name):
n = self._n_tracks
if n > 1:
raise AttributeError(
f"This file has {n} tracks; use file.tracks to access each one."
)
if n == 1:
return super().__getitem__(f"tracks/track_0/{name}")
return super().__getitem__(name)
@property
def path(self):
"""Return the path of the file.
For files opened by streaming an ``hf://`` path, ``self.filename`` is a
placeholder for the underlying file object rather than a real path, so we
return the original ``hf://`` source path instead.
"""
source = getattr(self, "_source_name", None)
if source is not None:
# ``Path("hf://org/repo/file")`` collapses the ``//`` to ``hf:/``, mangling the
# URI, so return it verbatim. ``name``/``stem`` wrap it in PurePath, which still
# extracts the basename correctly.
if str(source).startswith(HF_PREFIX):
return str(source)
return Path(source)
return Path(self.filename)
@property
def zea_version(self) -> str | None:
"""Return the zea version that wrote this file, or ``None`` for legacy files.
Files created with zea v0.1.0 and later store a ``zea_version``
root attribute. Files written before zea v0.1.0 return ``None``.
"""
return self.attrs.get("zea_version", None)
@property
def _n_tracks(self) -> int:
"""Return the number of tracks stored in this file.
Returns 0 for files with neither a ``tracks/`` group nor a root-level
``data/`` group, 1 for flat-layout files (no tracks group), and the actual
track count for files written with the multi-track layout.
"""
if not self.id.valid:
raise ValueError(
"File is closed. Use 'with File(...) as f:' or call f.close() "
"explicitly after you're done."
)
if "tracks" not in self:
return 1 if (super().__contains__("data") or super().__contains__("scan")) else 0
tracks_group = self["tracks"]
count = 0
while f"track_{count}" in tracks_group:
count += 1
return count
@property
def tracks(self) -> "list[Track]":
"""Return a list of :class:`Track` objects, one per track.
Each track exposes ``.data`` (a :class:`GroupProxy`), ``.scan`` (a
:class:`~zea.data.spec.ScanSpec`) and ``.load_parameters()`` (a
:class:`~zea.Parameters` factory method) for that specific track.
Raises:
AttributeError: For flat-layout files that have no
``tracks/`` group — use :attr:`data` and :meth:`scan`
directly for those.
Example::
with File("multi_track.hdf5") as f:
for track in f.tracks:
raw = track.data.raw_data[:]
parameters = track.load_parameters()
"""
if "tracks" not in self:
raise AttributeError(
"This file uses the flat layout (no 'tracks' group). "
"Access data and parameters directly with file.data and file.load_parameters()."
)
tracks_group = self["tracks"]
# Load file-level probe once so every track's scan can supplement its
# scan parameters with probe_geometry, element_width, etc.
probe_dict: "dict | None" = None
if super().__contains__("probe"):
probe_dict = load_dict_from_hdf5_group(self["probe"])
tracks: list[Track] = []
i = 0
while f"track_{i}" in tracks_group:
track_group = tracks_group[f"track_{i}"]
label = None
if "label" in track_group:
raw = track_group["label"][()]
label = raw.decode() if isinstance(raw, bytes) else str(raw)
tracks.append(
Track(i, track_group, label=label, probe=probe_dict, fetcher=self._chunk_fetcher)
)
i += 1
schedule = self.track_schedule
if schedule is None and len(tracks) == 1:
# For single-track files without an explicit schedule every transmit
# event belongs to track 0, so synthesise the schedule on the fly.
try:
n_events = tracks[0].n_frames * tracks[0].n_tx
schedule = np.zeros(n_events, dtype=np.int32)
except (TypeError, KeyError):
schedule = None
if schedule is not None:
all_timestamps = _compute_all_track_timestamps(schedule, tracks)
for track, ts in zip(tracks, all_timestamps):
object.__setattr__(track, "_timestamps", ts)
else:
log.warning_once(
"`track_schedule` was not found in the file; cannot compute track timestamps."
)
return tracks
@property
def track_labels(self) -> "list[str | None]":
"""Labels of all tracks in acquisition order.
Returns a list with one entry per track. Each entry is the label
string stored on that track, or ``None`` for unlabelled tracks (e.g.
single-track or legacy files). The list order matches
:attr:`tracks`, so unpacking ``f.tracks`` in the same order as
``f.track_labels`` is always safe.
Example::
with File("acquisition.hdf5") as f:
print(f.track_labels) # ['focused', 'planewave']
focused, planewave = f.tracks # safe — same order
"""
if "tracks" not in self:
return []
tracks_group = self["tracks"]
labels = []
i = 0
while f"track_{i}" in tracks_group:
tg = tracks_group[f"track_{i}"]
if "label" in tg:
raw = tg["label"][()]
labels.append(raw.decode() if isinstance(raw, bytes) else str(raw))
else:
labels.append(None)
i += 1
return labels
[docs]
def get_track(self, label: str) -> "Track":
"""Return the track with the given label.
Args:
label: The exact label string assigned to the desired track.
Returns:
Track: The matching :class:`Track` object.
Raises:
KeyError: If no track with that label exists, with a message
listing the available labels so the error is self-diagnosing.
Example::
with File("acquisition.hdf5") as f:
focused = f.get_track("focused")
raw = focused.data.raw_data[:]
"""
for track in self.tracks:
if track.label == label:
return track
available = [t.label for t in self.tracks]
raise KeyError(
f"No track with label {label!r}. Available labels (in acquisition order): {available}"
)
@property
def track_schedule(self) -> "np.ndarray | None":
"""Track index for each global transmit event, shape ``(n_total_tx,)``.
Returns an ``int32`` array that maps every transmit event (in
acquisition order) to the track it belongs to, or ``None`` if no
``track_schedule`` dataset was stored in this file.
Example::
with File("multi_track.hdf5") as f:
sched = f.track_schedule # e.g. array([0, 1, 0, 1, ...])
"""
if "track_schedule" not in self:
return None
return self["track_schedule"][()].astype(np.int32)
@property
def timestamps(self) -> "np.ndarray | None":
"""Global transmit timestamps in acquisition order, shape ``(n_total_tx,)``.
Returns a 1-D ``float32`` array whose ``i``-th entry is the
start time of the ``i``-th transmit event across **all** tracks, in the
order defined by :attr:`track_schedule`. This is the flattened view
complementary to the per-track ``(n_frames, n_tx)`` matrices returned
by :attr:`Track.timestamps`.
Returns ``None`` when timestamps cannot be computed (missing
``time_to_next_transmit`` on any track, or the file uses the legacy
flat layout with no tracks group).
Example::
with File("multi_track.hdf5") as f:
ts = f.timestamps # shape (n_total_tx,)
"""
try:
tracks = self.tracks
except AttributeError:
return None
if not tracks or any(t.timestamps is None for t in tracks):
return None
schedule = self.track_schedule
if schedule is None and len(tracks) == 1:
n_events = tracks[0].n_frames * tracks[0].n_tx
schedule = np.zeros(n_events, dtype=np.int32)
if schedule is None:
return None
ts_matrices = [np.asarray(t.timestamps, dtype=np.float32) for t in tracks]
n_tx_per_track = [m.shape[1] for m in ts_matrices]
track_counters = [[0, 0] for _ in tracks]
global_timestamps = np.empty(len(schedule), dtype=np.float32)
for event_idx, track_idx in enumerate(schedule):
frame_idx, tx_idx = track_counters[track_idx]
global_timestamps[event_idx] = ts_matrices[track_idx][frame_idx, tx_idx]
tx_idx += 1
if tx_idx >= n_tx_per_track[track_idx]:
tx_idx = 0
frame_idx += 1
track_counters[track_idx] = [frame_idx, tx_idx]
return global_timestamps
@property
def _scan_h5_group(self) -> "h5py.Group | None":
"""Return the HDF5 scan group for single-track or flat-layout files.
Track format (single track): ``tracks/track_0/scan/``
Flat layout (no tracks group): ``scan/`` at root
Returns ``None`` when neither is present.
Raises:
AttributeError: For multi-track files — use ``file.tracks[i]`` instead.
"""
n = self._n_tracks
if n > 1:
raise AttributeError(
f"This file has {n} tracks. "
"Use file.tracks[i].scan to access a specific track's scan parameters."
)
if "tracks" in self:
track0 = self["tracks"].get("track_0")
if track0 is not None and "scan" in track0:
return track0["scan"]
if super().__contains__("scan"):
return self["scan"]
return None
[docs]
@classmethod
def create(
cls,
path,
data: dict | None = None,
scan: dict | None = None,
tracks: list | None = None,
track_schedule: "np.ndarray | None" = None,
metadata: dict | None = None,
metrics: dict | None = None,
probe_name: str | None = None,
probe: "ProbeSpec | dict | None" = None,
us_machine: str | None = None,
description: str | None = None,
acquisition_time: str | None = None,
custom=None,
compression: "str | Mapping | None" = DEFAULT_COMPRESSION,
chunk_axes: "tuple[str, ...] | None" = DEFAULT_CHUNK_AXES,
overwrite: bool = False,
ignore_warnings: bool = False,
warn_missing_optional_fields: bool = True,
):
"""Create a new zea HDF5 file from data, scan, and optional metadata.
All inputs are validated against the :class:`~zea.data.spec.FileSpec`
schema (dtypes, shapes, dimension consistency) **before** anything is
written to disk.
For single-track files, supply ``data`` and ``scan``. For multi-track
files, supply ``tracks`` (a list of dicts with ``"data"`` and ``"scan"``
keys, or :class:`~zea.data.spec.TrackSpec` objects) and optionally
``track_schedule``.
Args:
path: Destination file path.
data: Data dict accepted by :class:`~zea.data.spec.DataSpec`.
Mutually exclusive with ``tracks``.
scan: Scan-parameter dict accepted by :class:`~zea.data.spec.ScanSpec`.
Mutually exclusive with ``tracks``.
tracks: List of track dicts (each with ``"data"`` and ``"scan"``
keys) accepted by :class:`~zea.data.spec.TrackSpec` objects.
Mutually exclusive with ``data``/``scan``.
track_schedule: Optional int32 array of length ``n_total_tx``
indicating which track each global transmit belongs to.
Only used with ``tracks``.
metadata: Optional metadata dict accepted by
:class:`~zea.data.spec.MetadataSpec`.
metrics: Optional metrics dict accepted by
:class:`~zea.data.spec.MetricsSpec`.
probe_name: Removed — use ``probe={'name': ...}`` instead.
probe: Probe specification as a :class:`~zea.probes.Probe` object or a
plain dict accepted by :class:`~zea.data.spec.ProbeSpec`.
us_machine: Name of the ultrasound machine.
description: Free-text description of the acquisition.
acquisition_time: UTC acquisition timestamp as an ISO 8601 string
(e.g. ``"2026-06-12T14:30:00+00:00"``). When *None* (default) no
timestamp is recorded. To capture the current moment, pass
``datetime.now(timezone.utc).isoformat()`` (requires
``from datetime import datetime, timezone``). Note: recording
timestamps for human subjects may constitute Protected Health
Information (PHI) under HIPAA and similar regulations.
custom: Optional list of :class:`CustomElement` objects holding data that
does not fit the zea format. They are stored in a ``custom`` group and
read back via :attr:`File.custom`.
compression: HDF5 compression filter (default Blosc zstd+shuffle). May be
a filter name (``"lzf"``, ``"gzip"``) or a mapping of ``create_dataset``
kwargs, e.g. an ``hdf5plugin`` filter object (``hdf5plugin.Blosc(...)``)
or ``{"compression": "gzip", "compression_opts": 4}``. Files written
with ``hdf5plugin`` filters require ``import hdf5plugin`` to be read
back (zea imports it automatically).
chunk_axes: Dimension names to chunk with HDF5 chunk size 1 (default
``("n_frames",)``); every other axis is stored at full extent,
so partial and streamed (``hf://``) reads fetch only the frames/
transmits requested. Axes not present on a field are ignored (an
``image`` without ``n_tx`` is chunked on ``n_frames`` only). Set to a
single axis (e.g. ``("n_frames",)``) for one full frame per chunk, or
to ``None``/``()`` for contiguous storage (use with
``compression=None``). See :meth:`zea.data.spec.Spec._resolve_chunks`.
overwrite: If *False* (default), raise if the file exists.
ignore_warnings: If *True*, suppress all warnings emitted while
creating the file (missing optional metadata fields, custom keys,
PHI timestamp warning, etc.). Defaults to *False*. Takes precedence
over ``warn_missing_optional_fields``. Note that some rarely-used
metadata fields (e.g. ``voice_narration``, ``ecg``) are never
warned about regardless of this flag.
warn_missing_optional_fields: If *True* (default), warn about missing
optional fields specifically (a subset of ``ignore_warnings``).
``ignore_warnings=True`` with this set to *False* raises
``ValueError`` as redundant.
Returns:
None. The validated file is written to ``path``; open it with
``File(path)`` to read it back.
Single-track example:
.. doctest::
>>> from datetime import datetime, timezone
>>> import numpy as np
>>> from zea import File
>>> n_frames, n_tx, n_ax, n_el = 2, 4, 64, 8
>>> raw = np.zeros((n_frames, n_tx, n_ax, n_el, 1), dtype=np.float32)
>>> probe_geometry = np.zeros((n_el, 3), dtype=np.float32)
>>> scan = {
... "sampling_frequency": np.float32(40e6),
... "center_frequency": np.float32(5e6),
... "demodulation_frequency": np.float32(5e6),
... "initial_times": np.zeros(n_tx, dtype=np.float32),
... "t0_delays": np.zeros((n_tx, n_el), dtype=np.float32),
... "tx_apodizations": np.ones((n_tx, n_el), dtype=np.float32),
... "focus_distances": np.full(n_tx, np.inf, dtype=np.float32),
... "transmit_origins": np.zeros((n_tx, 3), dtype=np.float32),
... "polar_angles": np.zeros(n_tx, dtype=np.float32),
... "time_to_next_transmit": np.ones((n_frames, n_tx), dtype=np.float32) * 1e-4,
... }
>>> File.create(
... "example.hdf5",
... data={"raw_data": raw},
... scan=scan,
... probe={"name": "verasonics_l11_4v", "probe_geometry": probe_geometry},
... acquisition_time=datetime.now(timezone.utc).isoformat(),
... overwrite=True,
... )
.. testcleanup::
import os
os.unlink("example.hdf5")
"""
if tracks is not None and (data is not None or scan is not None):
raise ValueError("Provide either 'tracks' or 'data'/'scan', not both.")
if ignore_warnings and not warn_missing_optional_fields:
raise ValueError(
"warn_missing_optional_fields=False has no effect when ignore_warnings=True: "
"ignore_warnings already suppresses every warning, including missing-optional-"
"field warnings. Pass only one of the two."
)
path = Path(path)
if path.exists() and not overwrite:
raise FileExistsError(f"File already exists: {path}")
if probe_name is not None:
raise TypeError(
"probe_name is no longer supported. "
"Use probe={'name': ...} to specify the probe name."
)
if probe is not None and not isinstance(probe, (dict, ProbeSpec)):
raise TypeError(f"probe must be a Probe object or a dict, got {type(probe).__name__}.")
kwargs: dict = {}
if tracks is not None:
kwargs["tracks"] = tracks
if track_schedule is not None:
kwargs["track_schedule"] = track_schedule
else:
if data is None:
raise ValueError("Either 'data' or 'tracks' must be provided.")
kwargs["data"] = data
if scan:
kwargs["scan"] = scan
if metadata is not None:
kwargs["metadata"] = metadata
if metrics is not None:
kwargs["metrics"] = metrics
if probe is not None:
kwargs["probe"] = probe
if us_machine is not None:
kwargs["us_machine"] = us_machine
if description is not None:
kwargs["description"] = description
if acquisition_time is not None:
kwargs["acquisition_time"] = acquisition_time
if custom is not None:
kwargs["custom"] = custom
if tracks is not None:
for track in tracks:
track_data = (
track.get("data")
if isinstance(track, Mapping)
else getattr(track, "data", None)
)
if isinstance(track_data, Mapping):
_validate_custom_key_naming(track_data, {})
_validate_custom_key_naming({}, kwargs.get("metadata", {}))
else:
_validate_custom_key_naming(kwargs.get("data", {}), kwargs.get("metadata", {}))
warn_ctx = log.suppress_warnings() if ignore_warnings else contextlib.nullcontext()
with warn_ctx:
_warn_custom_keys(kwargs.get("data", {}), kwargs.get("metadata", {}))
spec = FileSpec(**kwargs)
spec.save(
str(path),
compression=compression,
chunk_axes=chunk_axes,
warn_missing_optional_fields=warn_missing_optional_fields,
)
@property
def data(self) -> "_DataProxy":
"""Lazy proxy for the ``data`` group of a single-track file.
Supports both the new ``tracks/track_0/data/`` layout and the
flat ``data/`` layout (files without a tracks group).
Returns a :class:`GroupProxy` so individual datasets can be accessed
as attributes without loading everything into RAM::
with File(path) as f:
f.data.raw_data[:, :n_tx] # read a slice
f.data.image.values[0] # nested group access
Raises:
AttributeError: When the file contains more than one track.
Use :attr:`tracks` to iterate over individual tracks.
"""
n = self._n_tracks
if n > 1:
raise AttributeError(
f"This file has {n} tracks. "
"Use file.tracks to get a list of tracks and access each "
"track's data individually: file.tracks[i].data"
)
# New-format single track
if "tracks" in self:
track0 = self["tracks"].get("track_0")
if track0 is not None and "data" in track0:
return cast("_DataProxy", _GroupProxy(track0["data"], self._chunk_fetcher))
# Flat layout (no tracks group): root-level data/ group
if super().__contains__("data"):
return cast("_DataProxy", _GroupProxy(self["data"], self._chunk_fetcher))
raise KeyError("No 'data' group found in this file.")
@property
def _is_legacy_file(self) -> bool:
return _is_legacy_file(self)
@property
def custom(self) -> "CustomElements":
"""Custom data elements.
Returned as a :class:`CustomElements` collection: it behaves like
``list[CustomElement]`` (iteration, ``len()``, positional indexing, equality with a
plain list), but also supports name- and group-based access, e.g.
``f.custom.lens_correction`` or ``f.custom["lens"]["profile"]`` for a
``group_name="lens"`` element named ``"profile"``.
"""
if self._is_legacy_file:
if "non_standard_elements" not in self:
return CustomElements([])
return CustomElements(_load_custom_elements_from_group(self, "non_standard_elements"))
if "custom" not in self:
return CustomElements([])
return CustomElements(_load_custom_elements_from_group(self, "custom"))
@property
def name(self):
"""Return the name of the file."""
# ``path`` may be a plain string (an ``hf://`` URI); PurePath handles both.
return PurePath(str(self.path)).name
@property
def stem(self):
"""Return the stem of the file."""
return PurePath(str(self.path)).stem
def _get_single_track_data_group(self) -> "h5py.Group":
"""Return the data group for single-track or flat-layout files."""
if "tracks" in self:
return self["tracks/track_0/data"]
return self["data"]
@property
def n_frames(self) -> int:
"""Number of frames."""
if self._n_tracks > 1:
raise AttributeError(
f"This file has {self._n_tracks} tracks. Use file.tracks[i].n_frames."
)
return _shape_from_data_group(
self._get_single_track_data_group(),
index=0,
name="n_frames",
requires_raw=False,
)
@property
def n_tx(self) -> int:
"""Number of transmit events."""
if self._n_tracks > 1:
raise AttributeError(f"This file has {self._n_tracks} tracks. Use file.tracks[i].n_tx.")
return _shape_from_data_group(
self._get_single_track_data_group(),
index=1,
name="n_tx",
requires_raw=True,
)
@property
def n_ax(self) -> int:
"""Number of axial samples."""
if self._n_tracks > 1:
raise AttributeError(f"This file has {self._n_tracks} tracks. Use file.tracks[i].n_ax.")
return _shape_from_data_group(
self._get_single_track_data_group(),
index=2,
name="n_ax",
requires_raw=True,
)
@property
def n_el(self) -> int:
"""Number of elements."""
if self._n_tracks > 1:
raise AttributeError(f"This file has {self._n_tracks} tracks. Use file.tracks[i].n_el.")
return _shape_from_data_group(
self._get_single_track_data_group(),
index=3,
name="n_el",
requires_raw=True,
)
[docs]
def shape(self, key) -> tuple:
"""Return shape of some key."""
key = self.format_key(key)
return self[key].shape
[docs]
def to_iterator(self, key):
"""Convert the data to an iterator over all frames."""
key = self.format_key(key)
for frame_idx in range(self.n_frames):
yield self[key][frame_idx]
[docs]
@staticmethod
def key_to_data_type(key):
"""Convert the key to a data type."""
data_type = key.split("/")[-1]
return data_type
[docs]
def load_transmits(self, key, selected_transmits):
"""Load raw_data or aligned_data for a given list of transmits.
Args:
key (str): The type of data to load. Options are 'raw_data' and 'aligned_data'.
selected_transmits (list, np.ndarray): The transmits to load.
"""
key = self.format_key(key)
data_type = self.key_to_data_type(key)
assert data_type in ["raw_data", "aligned_data"], (
f"Cannot load transmits for {data_type}. Only raw_data and aligned_data are supported."
)
# First axis: all frames, second axis: selected transmits
indices = (slice(None), np.array(selected_transmits))
return self[key][indices]
[docs]
@deprecated(replacement="File.data.<key> with h5py slice indexing")
def load_data(
self,
data_type,
indices: Tuple[Union[list, slice, int], ...] | List[int] | int | slice | None = None,
) -> np.ndarray:
"""Load data from the file.
.. deprecated::
Use ``file.data.<key>`` with standard h5py slice indexing instead::
with File(path) as f:
raw = f.data.raw_data[:] # all frames
raw = f.data.raw_data[0] # first frame
raw = f.data.raw_data[0, [0, 2]] # frame 0, transmits 0 and 2
.. include:: ../common/file_indexing.rst
Args:
data_type (str): The type of data to load. Options are 'raw_data', 'aligned_data',
'beamformed_data', 'envelope_data', 'image' and 'image_sc'.
indices (optional): The indices to load. Defaults to ``None`` in
which case all data is loaded.
"""
key = self.format_key(data_type)
if indices is None or (isinstance(indices, str) and indices == "all"):
indices = slice(None)
data = self[key]
try:
data = data[indices]
except (OSError, IndexError) as exc:
raise ValueError(
f"Invalid indices {indices} for key {key}. {key} has shape {data.shape}."
) from exc
return data
@property
def probe_name(self):
"""Reads the probe name from the data file and returns it."""
# Priority: 'probe_name' attr → 'probe' group name → legacy 'probe' attr
if "probe_name" in self.attrs:
return self.attrs["probe_name"]
# Check the structured probe group for a 'name' dataset.
if "probe" in self and "name" in self["probe"]:
return self["probe"]["name"].asstr()[()]
if "probe" in self.attrs:
return self.attrs["probe"]
raise AttributeError(
"Probe name not found in file attributes. "
"Make sure you are using a zea file. "
f"Found attributes: {list(self.attrs)}"
)
@property
def us_machine(self):
"""Reads the ultrasound machine name from the data file and returns it."""
assert "us_machine" in self.attrs, (
"Ultrasound machine name not found in file attributes. "
"Make sure you are using a zea file. "
f"Found attributes: {list(self.attrs)}"
)
us_machine = self.attrs["us_machine"]
return us_machine
@property
def description(self):
"""Reads the description from the data file and returns it."""
assert "description" in self.attrs, (
"Description not found in file attributes. "
"Make sure you are using a zea file. "
f"Found attributes: {list(self.attrs)}"
)
description = self.attrs["description"]
return description
@property
def acquisition_time(self) -> datetime | None:
"""Return the acquisition timestamp as a timezone-aware UTC :class:`datetime`.
Returns *None* when no timestamp was stored (e.g. human-subject files
saved without an explicit ``acquisition_time``).
"""
if "acquisition_time" not in self.attrs:
return None
raw = self.attrs["acquisition_time"]
if isinstance(raw, bytes):
raw = raw.decode()
dt = datetime.fromisoformat(raw)
if dt.tzinfo is None:
dt = dt.replace(tzinfo=timezone.utc)
return dt
[docs]
def get_scan_parameters(self):
"""Returns a dictionary of parameters to initialize a scan
object that comes with the file (stored inside datafile).
If there are no scan parameters in the hdf5 file, returns
an empty dictionary.
Returns:
dict: The scan parameters.
"""
scan_group = self._scan_h5_group
if scan_group is None:
log.warning("Could not find scan parameters in file.")
return {}
scan_parameters = load_dict_from_hdf5_group(scan_group)
return scan_parameters
@property
def scan(self) -> "ScanSpec | None":
"""Return the validated :class:`~zea.data.spec.ScanSpec` for this file.
This is the bare scan group as a spec object. For a full, derivable
parameter object (merged probe + scan, with caching and derived
properties) use :meth:`load_parameters`.
Raises:
AttributeError: When the file contains more than one track.
Use :attr:`tracks` and access ``.scan`` on each track instead.
.. doctest::
>>> from zea import File
>>> path = (
... "hf://zeahub/picmus/database/experiments/contrast_speckle/"
... "contrast_speckle_expe_dataset_iq/contrast_speckle_expe_dataset_iq.hdf5"
... )
>>> with File(path, mode="r") as f:
... scan = f.scan
>>> type(scan).__name__
'ScanSpec'
"""
n = self._n_tracks
if n > 1:
raise AttributeError(
f"This file has {n} tracks. "
"Use file.tracks to get a list of tracks and access .scan on each: "
"file.tracks[i].scan"
)
scan_dict = self.get_scan_parameters()
if _is_legacy_file(self):
scan_dict = legacy_scan(scan_dict)
return ScanSpec(**scan_dict) if len(scan_dict) > 0 else None
[docs]
def load_parameters(self, **overrides) -> "Parameters":
"""Load the acquisition parameters (merged probe + scan) from the file.
Reads both the ``scan`` and ``probe`` groups and merges them into a
single :class:`~zea.Parameters` object that owns derivation,
caching, and lazy loading of derived quantities. The probe and scan
groups live at the same level and have non-overlapping field names, so
merging is a plain dict union.
Args:
**overrides: Override any parameter from the file. Custom
(non-spec) keys are stored as passthrough parameters.
Returns:
Parameters: The merged, derivable parameters object.
Raises:
AttributeError: When the file contains more than one track.
Use :attr:`tracks` and call ``.load_parameters()`` on each track.
.. doctest::
>>> from zea import File
>>> path = (
... "hf://zeahub/picmus/database/experiments/contrast_speckle/"
... "contrast_speckle_expe_dataset_iq/contrast_speckle_expe_dataset_iq.hdf5"
... )
>>> with File(path) as f:
... parameters = f.load_parameters()
>>> type(parameters).__name__
'Parameters'
"""
from zea.parameters import Parameters
n = self._n_tracks
if n > 1:
raise AttributeError(
f"This file has {n} tracks. "
"Use file.tracks to get a list of tracks and call "
"file.tracks[i].load_parameters() on each."
)
scan = self.scan
probe = self.probe
probe_dict = probe.get_parameters()
if scan is not None:
scan_dict = scan.to_dict()
other_dict = {
"n_ax": self.n_ax,
"n_el": scan.n_el,
"n_tx": scan.n_tx,
}
else:
scan_dict = {}
other_dict = {}
merged_dict = {**probe_dict, **scan_dict, **other_dict, **overrides}
# skip None values
merged_dict = {k: v for k, v in merged_dict.items() if v is not None}
return Parameters(**merged_dict)
@property
def probe(self) -> "Probe":
"""Returns a Probe object initialized with the parameters from the file.
Returns:
Probe: The probe object.
Example:
.. doctest::
>>> from zea import File
>>> path = (
... "hf://zeahub/picmus/database/experiments/contrast_speckle/"
... "contrast_speckle_expe_dataset_iq/contrast_speckle_expe_dataset_iq.hdf5"
... )
>>> with File(path) as f:
... probe = f.probe
>>> probe.name
'verasonics_l11_4v'
"""
from zea.probes import Probe
if "probe" in self.keys():
probe_dict = self.recursively_load_dict_contents_from_group("probe")
elif _is_legacy_file(self):
scan_dict = self.get_scan_parameters()
probe_dict = legacy_probe(scan_dict)
probe_dict["name"] = self.probe_name
else:
# Image-only datasets carry no probe group; an empty Probe lets
# load_parameters merge in nothing rather than failing.
probe_dict = {}
return Probe(**probe_dict)
@property
def metadata(self) -> MetadataSpec:
"""Return a validated :class:`~zea.data.spec.MetadataSpec` object from the file.
Returns:
MetadataSpec: The validated metadata spec.
Raises:
KeyError: If the file has no ``metadata`` group.
Example:
.. doctest::
>>> from zea import File
>>> path = (
... "hf://zeahub/picmus/database/experiments/contrast_speckle/"
... "contrast_speckle_expe_dataset_iq/contrast_speckle_expe_dataset_iq.hdf5"
... )
>>> with File(path) as f:
... meta = f.metadata
... print(meta.subject.type)
phantom
"""
if "metadata" not in self:
raise KeyError("No 'metadata' group in this file.")
raw = load_dict_from_hdf5_group(self["metadata"])
return MetadataSpec(**raw)
@property
def metrics(self) -> MetricsSpec:
"""Return a validated :class:`~zea.data.spec.MetricsSpec` object from the file.
Returns:
MetricsSpec: The validated metrics spec.
Raises:
KeyError: If the file has no ``metrics`` group.
Example::
>>> with File("my_file.hdf5") as f: # doctest: +SKIP
... met = f.metrics
... print(met.coherence_factor.shape)
"""
if "metrics" not in self:
raise KeyError("No 'metrics' group in this file.")
raw = load_dict_from_hdf5_group(self["metrics"])
return MetricsSpec(**raw)
[docs]
def recursively_load_dict_contents_from_group(self, path: str) -> dict:
"""Load dict from contents of group.
.. deprecated::
Use the module-level :func:`load_dict_from_hdf5_group` function instead,
passing an :class:`h5py.Group` directly.
Args:
path (str): path to group
Returns:
dict: dictionary with contents of group
"""
return load_dict_from_hdf5_group(self[path])
[docs]
def has_key(self, key: str) -> bool:
"""Check if the file has a specific key.
Args:
key (str): The key to check.
Returns:
bool: True if the key exists, False otherwise.
"""
try:
key = self.format_key(key)
except AssertionError:
return False
return True
[docs]
@classmethod
def get_shape(cls, path: str, key: str) -> tuple:
"""Get the shape of a key in a file.
Args:
path (str): The path to the file.
key (str): The key to get the shape of.
Returns:
tuple: The shape of the key.
"""
with cls(path, mode="r") as file:
return file.shape(key)
[docs]
def validate(self):
"""Lightweight structural validation — no array data is loaded into RAM.
Checks that the file has a ``data`` group and that all keys within it
are recognised zea data types. For legacy files (before zea v0.1.0)
a minimal key-name check is performed. For files created with
zea v0.1.0 and later (via :meth:`File.create`) the keys are checked
against the :class:`~zea.data.spec.DataSpec` schema.
Use :meth:`validate_spec` for a **full** validation that loads all data
and checks dtypes, shapes, and cross-field dimension consistency.
Returns:
dict: ``{"status": "success"}`` on success.
Raises:
AssertionError: If the file is missing required groups or contains
unrecognised data keys.
"""
try:
return validate_file(file=self)
except Exception as e:
log.error(f"File {self.path} is not a valid zea file.\n{e}\n")
raise
def _to_file_spec(self) -> FileSpec:
"""Load the whole file into a validated :class:`~zea.data.spec.FileSpec`.
Unlike the lazy :attr:`data` / :attr:`scan` accessors, every dataset is
read into memory here. Both the multi-track ``tracks/track_N/`` layout
and the flat ``data/`` + ``scan/`` layout (files without a tracks group) are supported.
Returns:
FileSpec: A fully validated spec object, with all arrays in RAM.
"""
kwargs: dict = {"tracks": self._load_tracks(), "probe": self.probe}
if self.custom:
kwargs["custom"] = self.custom
if self.track_schedule is not None:
kwargs["track_schedule"] = self.track_schedule
if "metadata" in self:
kwargs["metadata"] = self.metadata
if "metrics" in self:
kwargs["metrics"] = self.metrics
if "us_machine" in self.attrs:
kwargs["us_machine"] = self.attrs["us_machine"]
if "description" in self.attrs:
kwargs["description"] = self.attrs["description"]
if "acquisition_time" in self.attrs:
kwargs["acquisition_time"] = self.attrs["acquisition_time"]
return FileSpec(**kwargs)
def _load_tracks(self) -> "list[dict]":
"""Read every track's ``data`` and ``scan`` fully into a list of dicts.
Each dict is shaped for :class:`~zea.data.spec.TrackSpec`. Flat-layout
files (no tracks group) are returned as a single, unlabelled track.
"""
# Flat layout: one unlabelled track at the file root.
if "tracks" not in self:
track: dict = {}
if super().__contains__("data"):
data = load_dict_from_hdf5_group(self["data"])
track["data"] = legacy_data(data) if _is_legacy_file(self) else data
if self.scan is not None:
track["scan"] = self.scan
return [track]
# Multi-track layout: tracks/track_N/{data,scan,label}.
tracks = []
for track in self.tracks:
track_dict: dict = {"label": track.label}
if "data" in track._group:
track_dict["data"] = load_dict_from_hdf5_group(track._group["data"])
if "scan" in track._group:
track_dict["scan"] = track.scan
# Preserve transmit-only tracks (data=None); without this the rebuilt TrackSpec
# would default transmit_only=False and reject the missing data.
if "transmit_only" in track._group:
track_dict["transmit_only"] = bool(track._group["transmit_only"][()])
tracks.append(track_dict)
return tracks
[docs]
def validate_spec(self) -> FileSpec:
"""Full schema validation — loads all data into RAM.
Reads every dataset in the file and runs dtype, shape, and
cross-dimension consistency checks as defined by :class:`~zea.data.spec.FileSpec`.
Use this to confirm a file is fully spec-compliant before sharing or
processing it.
For a fast, zero-IO structural check use :meth:`validate` instead.
.. note::
This method only works on files created with zea v0.1.0 and later.
Files written before zea v0.1.0 should be re-saved through
:meth:`File.create`.
Returns:
FileSpec: The fully validated spec object, with all data accessible
as typed attributes (e.g. ``spec.data.raw_data``, ``spec.scan.n_tx``).
Raises:
TypeError, ValueError: If the file does not conform to the spec.
.. doctest::
>>> with File("my_file.hdf5") as f: # doctest: +SKIP
... spec = f.validate_spec()
... print(spec.scan.n_tx)
"""
return self._to_file_spec()
def __repr__(self):
name = Path(self.filename).name
try:
n = self._n_tracks
if n > 1:
labels = self.track_labels
label_str = ", ".join(
f'"{label}"' if label else str(i) for i, label in enumerate(labels)
)
track_info = f"{n} tracks: {label_str}"
else:
track_info = f"{n} track"
except Exception:
track_info = ""
mode_info = f"mode {self.mode}"
return f'<File "{name}" ({mode_info}, {track_info})>'
[docs]
def copy_key(self, key: str, dst: "File"):
"""Copy a specific key to another file.
Will always copy the attributes and the scan data if it exists. Will warn if the key is
not in this file or if the key already exists in the destination file.
Args:
key (str): The key to copy.
dst (File): The destination file to copy the key to.
"""
key = self.format_key(key)
# Copy the key if it does not already exist in the destination file
if key in dst:
log.warning(f"Skipping key '{key}' because it already exists in dst file {dst.path}.")
elif key in self:
self.copy(key, dst, name=key)
else:
log.warning(f"Key '{key}' not found in src file {self.path}. Skipping copy.")
# Copy attributes from src to dst
for attr_key, attr_value in self.attrs.items():
dst[key].attrs[attr_key] = attr_value
# Copy scan data if requested
if "scan" in self and "scan" not in dst:
# Use the actual HDF5 path (not our overridden key) for h5py.copy.
# The group is guaranteed to exist here because ``"scan" in self``.
scan_group = self._scan_h5_group
assert scan_group is not None
scan_path = scan_group.name.lstrip("/")
self.copy(scan_path, dst, name="scan")
[docs]
def summary(self):
"""Print a human-readable tree of the file's contents.
Scalars and small datasets are printed with their actual value; large
datasets show their shape and dtype instead. Unit and description
metadata is shown inline where available.
"""
_print_hdf5_attrs(self)
[docs]
@deprecated(replacement="File(...) with file.load_parameters() and file.data.<type>[...]")
def load_file(
path,
data_type="raw_data",
indices: Tuple[Union[list, slice, int], ...] | List[int] | int | slice | None = None,
scan_kwargs: dict | None = None,
) -> Tuple[np.ndarray, "Parameters"]:
"""Loads a zea data files (h5py file).
Returns the data together with a parameters object containing the parameters
of the acquisition. Probe information is available via ``parameters.to_probe_dict()``
or ``File.probe``.
Additionally, it can load a specific subset of frames / transmits.
.. include:: ../common/file_indexing.rst
Args:
path (str, pathlike): The path to the hdf5 file.
data_type (str, optional): The type of data to load. Defaults to
'raw_data'. Other options are 'aligned_data', 'beamformed_data',
'envelope_data', 'image' and 'image_sc'.
indices (optional): The indices to load. Defaults to None in
which case all frames are loaded.
scan_kwargs (Config, dict, optional): Additional keyword arguments
to pass to :meth:`File.load_parameters`. These will override the
parameters from the file if they are present. Defaults to None.
Returns:
(np.ndarray): The raw data of shape (n_frames, n_tx, n_ax, n_el, n_ch).
(Parameters): A parameters object containing the parameters of the acquisition.
"""
# Define the additional keyword parameters from the scan object
if scan_kwargs is None:
scan_kwargs = {}
with File(path, mode="r") as file:
# Load the desired frames from the file
_key = file.format_key(data_type)
_indices = indices if indices is not None else slice(None)
item = file[_key]
if isinstance(item, h5py.Group):
data = item["values"][_indices]
else:
data = item[_indices]
# extract transmits from indices
# we only have to do this when the data has a n_tx dimension
# in that case we also have update scan parameters to match
# the number of selected transmits
if data_type in ["raw_data", "aligned_data"]:
if isinstance(indices, tuple) and len(indices) > 1:
scan_kwargs["selected_transmits"] = indices[1]
parameters = file.load_parameters(**scan_kwargs)
return data, parameters
# A dataset with this many elements or fewer has its actual values printed;
# bigger datasets only show their shape/dtype to avoid dumping large arrays.
_MAX_INLINE_ELEMENTS = 8
# Placeholder unit/description values written by DataSpec for fields that have
# no curated metadata (see `_DEFAULT_FIELD_UNIT`/`_DEFAULT_FIELD_DESCRIPTION`
# in `zea.data.spec`). These carry no information, so the summary hides them.
_EMPTY_UNITS = {"", "-", "–"}
def _decode_hdf5_scalar(value):
"""Decode a raw scalar read from HDF5 (bytes/np.generic) to a native Python value."""
if isinstance(value, np.generic):
value = value.item()
if isinstance(value, bytes):
value = value.decode()
return value
def _format_hdf5_scalar(value):
"""Colorize a single decoded scalar value for display."""
if isinstance(value, str):
return log.cyan(f'"{value}"')
if isinstance(value, (bool, np.bool_)):
return log.cyan(str(value))
if isinstance(value, float):
if value != value: # NaN
text = "nan"
elif value.is_integer() and abs(value) < 1e15:
text = str(int(value))
else:
text = f"{value:.6g}"
return log.cyan(text)
return log.cyan(str(value))
def _format_hdf5_value(value):
"""Colorize an HDF5 attribute or (small) dataset value for display."""
if isinstance(value, np.ndarray):
if value.shape == ():
return _format_hdf5_scalar(_decode_hdf5_scalar(value[()]))
items = [_decode_hdf5_scalar(item) for item in value.tolist()]
return log.cyan(str(items))
return _format_hdf5_scalar(_decode_hdf5_scalar(value))
def _format_hdf5_attr_line(key, value):
"""Format a plain HDF5 attribute (e.g. a file- or group-level attribute) as one line."""
return f"{log.bold(key)}: {_format_hdf5_value(value)}"
def _format_hdf5_dataset_line(key, dataset):
"""Format a leaf HDF5 dataset as one line: its value (or shape) plus unit/description."""
if 0 < dataset.size <= _MAX_INLINE_ELEMENTS:
body = f"{log.bold(key)}: {_format_hdf5_value(dataset[()])}"
else:
shape_str = str(dataset.shape).replace(",)", ")")
if h5py.check_string_dtype(dataset.dtype) is not None:
dtype_str = "str"
else:
dtype_str = dataset.dtype.name
body = f"{log.bold(key)} " + log.dim(f"(shape={shape_str}, dtype={dtype_str})")
unit = _decode_hdf5_scalar(dataset.attrs.get("unit", ""))
description = _decode_hdf5_scalar(dataset.attrs.get("description", ""))
extras = []
if unit not in _EMPTY_UNITS:
extras.append(log.dim(f"[{unit}]"))
if description:
extras.append(log.dim(description))
if extras:
body += " " + " ".join(extras)
return body
def _print_hdf5_attrs(hdf5_obj, prefix=""):
"""Recursively prints the contents of an HDF5 file in a human-readable tree.
Scalars and small datasets (up to ``_MAX_INLINE_ELEMENTS`` elements) are printed
with their actual value; larger datasets only show their shape and dtype. Unit and
description metadata is shown inline, and hidden when unset.
Args:
hdf5_obj (h5py.File, h5py.Group): HDF5 object to print.
prefix (str, optional): Prefix to print before each line. This
parameter is used in internal recursion and should not be supplied
by the user.
"""
assert isinstance(hdf5_obj, (h5py.File, h5py.Group)), (
"ERROR: hdf5_obj must be a File or Group object"
)
if isinstance(hdf5_obj, h5py.File):
name = "root" if hdf5_obj.name == "/" else hdf5_obj.name
print(prefix + log.bold(name) + "/")
# Collect all children (attributes first, then sub-groups/datasets) into a
# single ordered list so ``is_last`` is computed across the full set and the
# tree connectors line up correctly.
entries = [("attr", key, val) for key, val in hdf5_obj.attrs.items()]
entries += [("item", key, None) for key in hdf5_obj.keys()]
for i, (kind, key, val) in enumerate(entries):
is_last = i == len(entries) - 1
marker = "└── " if is_last else "├── "
child_prefix = prefix + (" " if is_last else "│ ")
if kind == "attr":
print(prefix + marker + _format_hdf5_attr_line(key, val))
continue
child = hdf5_obj[key]
if isinstance(child, h5py.Dataset):
print(prefix + marker + _format_hdf5_dataset_line(key, child))
elif isinstance(child, h5py.Group):
print(prefix + marker + log.bold(key) + "/")
_print_hdf5_attrs(child, child_prefix)
else:
# e.g. a committed h5py.Datatype — not a Dataset or Group, nothing to recurse into.
print(prefix + marker + log.bold(key))
[docs]
def validate_file(path: str | None = None, file: "File | None" = None):
"""Validate the structure and data of a zea HDF5 file.
For files created with zea v0.1.0 and later this runs the full
:class:`~zea.data.spec.FileSpec` schema validation (dtypes, shapes, and
dimension consistency). Legacy files (before zea v0.1.0) are detected by the
presence of scalar dataset ``scan/n_frames``; for those only a lightweight
structural ``data`` group check is performed.
Provide either *path* or *file*, but not both.
Args:
path (str | pathlike): Path to the HDF5 file.
file (File): An already-open :class:`File` instance.
Returns:
dict: ``{"status": "success"}`` on success.
Raises:
AssertionError: If the file is missing the ``data`` group.
TypeError, ValueError: If spec validation fails on files created with zea v0.1.0 and later.
"""
assert (path is not None) ^ (file is not None), (
"Provide either the path or the file, but not both."
)
if path is not None:
with File(path, "r") as _file:
_validate_file_impl(_file)
else:
assert file is not None # guaranteed by the xor assertion above
_validate_file_impl(file)
return {"status": "success"}
def _is_legacy_file(file: File) -> bool:
"""Return ``True`` when *file* pre-dates the dataspec format.
Files created with zea v0.1.0 and later always store a
``zea_version`` root attribute. Files that lack it were produced by
the legacy data format path and are treated as legacy.
"""
return "zea_version" not in file.attrs
def _validate_file_impl(file: File) -> None:
"""Lightweight structural validation — no array data is loaded.
Checks that:
- a ``data`` group is present — either at ``tracks/track_N/data``
or at the root ``data`` group (legacy)
- for legacy files, every key in ``data`` is a recognised zea data type
- for files created with zea v0.1.0 and later, every key in ``data``
is in :class:`~zea.data.spec.DataSpec`\'s schema
"""
# Collect all data groups to validate
data_groups: list[tuple[str, h5py.Group]] = []
if super(File, file).__contains__("tracks"):
# New multi-track format: tracks/track_N/data
tracks_group = file["tracks"]
for track_key in tracks_group.keys():
track_grp = tracks_group[track_key]
assert "data" in track_grp, f"Track group '{track_key}' is missing a 'data' subgroup."
assert isinstance(track_grp["data"], h5py.Group), (
f"'{track_key}/data' is not a group - this may not be a zea file."
)
data_groups.append((f"tracks/{track_key}/data", track_grp["data"]))
elif super(File, file).__contains__("data"):
# Legacy root-level data group
assert isinstance(file["data"], h5py.Group), (
"'data' is not a group - this may not be a zea file."
)
data_groups.append(("data", file["data"]))
assert data_groups, (
"'data' group not found in file. "
"Expected either tracks/track_N/data or a root 'data' group."
)
for group_path, data_group in data_groups:
if _is_legacy_file(file):
# For legacy files: accepted keys are the flat _DATA_TYPES list.
has_raw = any(k in data_group for k in _NON_IMAGE_DATA_TYPES)
if has_raw:
assert "scan" in file, "Legacy file is missing the 'scan' group."
for key in data_group.keys():
assert key in _DATA_TYPES, (
f"'{group_path}/{key}' is not a recognised zea data type."
)
else:
# For new-format files: flat datasets must be known DataSpec keys.
# HDF5 Groups are Map specs (either a named type or a custom map)
# and are always accepted; validate() is a structural check only.
known = set(DataSpec.SCHEMA.keys())
known_flat = {k for k, v in DataSpec.SCHEMA.items() if "spec" not in v}
for key in data_group.keys():
if isinstance(data_group[key], h5py.Group):
# Named map or custom map — accepted without further checks here.
continue
assert key in known_flat, (
f"'{group_path}/{key}' is not in the DataSpec schema. "
f"Known keys: {sorted(known)}"
)