r"""Read Mojito L1 files through the common :class:`mojito.MojitoL1` interface.
Reading files
-------------
Use :func:`open_l1` to open one or more existing Mojito L1 files. The function selects
the reader implementation from the file version and returns a context manager. Use
:func:`create_l1` to create a new versioned file. Always use readers within a ``with``
statement so that HDF5 resources are closed correctly.
.. code-block:: python
from mojito import open_l1
with open_l1("path/to/file.h5") as l1:
xyz = l1.xyz[:1000]
time = l1.tdi_time_grid.t(idx=slice(0, 1000))
The current implementation supports Mojito L1 version 1 files through
:class:`MojitoL1SingleFileV1` and version 2 files through :class:`MojitoL1SingleFileV2`.
The file version is read from the root ``version`` attribute when present; legacy files
without that attribute are treated as version ``1.0.0`` for backward compatibility.
Version dispatch is handled by :func:`open_single_file_l1`, which selects the matching
reader for versions in the supported range and raises :class:`ValueError` for
unsupported versions.
Available quantities
--------------------
The :class:`mojito.MojitoL1` protocol exposes metadata, TDI observables, light travel
times, spacecraft orbits, and noise estimates.
Each group has an associated time grid, and noise estimates also have a log-uniform
frequency grid. Dataset properties return lazy, array-like objects; data is read from
the HDF5 file when a slice is requested.
Incomplete files
----------------
Single-file readers do not validate that every group is present when the file is opened.
A property whose group or dataset is absent raises :class:`KeyError` when accessed. The
metadata properties also raise :class:`KeyError` when a required root attribute is
missing or has an unexpected type. The metadata setter validates the full dict and
raises :class:`ValueError` for missing or unexpected keys.
Note that some readers, when opened in write-capable mode, may create missing datasets
automatically when accessed, based on the current grid sizes.
Reading multiple files
----------------------
Pass a list of paths to :func:`open_l1` to create a read-only :class:`MojitoL1MultiFile`
reader:
.. code-block:: python
with open_l1(["file1.h5", "file2.h5"]) as l1:
xyz = l1.xyz[:1000]
flags = l1.xyz_flags[:1000]
TDI observables and noise covariance estimates are summed lazily across files. Quality
flags are combined with a logical OR. LTTs, orbit estimates, and their time grids are
taken from the first file after compatibility checks.
The multi-file reader checks compatibility while entering the context. Files must have
matching versions and all time and frequency grids must be identical. The first 200
samples of LTTs, LTT derivatives, spacecraft positions, and spacecraft velocities must
also match. If opening or compatibility checking fails, readers that were opened before
the failure are closed before the exception is re-raised.
Creating and modifying files
----------------------------
Use :func:`create_l1` to create a new file with an explicit format version.
.. code-block:: python
with create_l1("path/to/file.h5", "1.0.0", mode="w-") as l1:
l1.metadata = {
"pipeline_name": "example",
"lolipops_version": "1.0.0",
"laser_frequency": 281600000000000.0,
}
l1.tdi_time_grid = ... # Initialize the TDI time grid as needed
l1.xyz[:] = np.array(...) # Missing dataset are automatically created
Reference
---------
.. autofunction:: open_l1
.. autofunction:: create_l1
.. autofunction:: open_single_file_l1
.. autoclass:: MojitoL1SingleFileV1
:members:
.. autoclass:: MojitoL1SingleFileV2
:members:
.. autoclass:: MojitoL1MultiFile
:members:
"""
import json
import logging
from collections import defaultdict
from pathlib import Path
from typing import Any, ClassVar, Self, override
import h5py
import numpy as np
from lisaconstants.indexing import MOSAS, SPACECRAFT
from packaging.specifiers import SpecifierSet
from packaging.version import Version
from mojito import MojitoL1
from mojito.dataset import (
Dataset,
LazyBooleanOrDataset,
LazyMappedDataset,
LazyScaledDataset,
LazyStackedDataset,
LazySumDataset,
)
from mojito.grids import LogUniformFrequencyGrid, UniformTimeGrid
from mojito.utils import FileMode, H5File, is_write_mode
logger = logging.getLogger(__name__)
_XYZ_TO_AET = np.array(
[
[-1 / np.sqrt(2), 0, 1 / np.sqrt(2)],
[1 / np.sqrt(6), -2 / np.sqrt(6), 1 / np.sqrt(6)],
[1 / np.sqrt(3), 1 / np.sqrt(3), 1 / np.sqrt(3)],
]
)
"""Mixing matrix transforming XYZ into AET TDI combinations.
This is an array of shape (3, 3) that transforms the TDI Michelson combinations
XYZ into the AET combinations (in that order).
The matrix is orthonormal, so its inverse (AET to XYZ) is its transpose.
"""
def _is_number(value: Any) -> bool:
"""Check whether a decoded JSON value is a number (booleans excluded)."""
return isinstance(value, (int, float)) and not isinstance(value, bool)
def _is_integer(value: Any) -> bool:
"""Check whether a decoded JSON value is an integer (booleans excluded)."""
return isinstance(value, int) and not isinstance(value, bool)
class MojitoL1SingleFile(H5File, MojitoL1):
"""Base class for versioned single-file Mojito L1 readers.
This class cannot be instantiated directly. Concrete single-file reader
implementations must subclass this base class and define the supported versions and
creation version.
>>> class MojitoL1SingleFileV1(
... MojitoL1SingleFile,
... supported_versions=SpecifierSet(">=1.0,<2.0"),
... creation_version=Version("1.0.0"),
... ):
... ...
Parameters
----------
path
Path to the HDF5 file.
mode
File access mode.
"""
_readers: ClassVar[list[type["MojitoL1SingleFile"]]] = []
"""Concrete single-file readers registered for version dispatch."""
_supported_versions: ClassVar[SpecifierSet]
"""File format versions supported by a concrete reader subclass."""
_creation_version: ClassVar[Version]
"""File format version written when a reader creates a new file."""
def __init__(self, path: str | Path, *, mode: FileMode = "r") -> None:
"""Guard against direct instantiation of the base class."""
if type(self) is MojitoL1SingleFile:
raise TypeError("MojitoL1SingleFile cannot be instantiated directly")
super().__init__(path, mode=mode)
def __init_subclass__(
cls,
*,
supported_versions: SpecifierSet,
creation_version: Version,
**kwargs: Any,
) -> None:
"""Register a concrete reader and validate its version policy.
Parameters
----------
supported_versions
Versions of the file format that the reader can open.
creation_version
Format version written when creating a new file with this reader, using
:meth:`create`. It must be contained in ``supported_versions``.
**kwargs
Additional keyword arguments forwarded to ``__init_subclass__``.
Raises
------
ValueError
If ``creation_version`` is not supported by ``supported_versions``.
RuntimeError
If ``supported_versions`` overlaps the range of an existing reader.
"""
super().__init_subclass__(**kwargs)
# Make sure creation_version is within the supported_versions range
if creation_version not in supported_versions:
raise ValueError(
f"Creation version {creation_version} is not supported by "
f"{supported_versions}"
)
# Ensure the new reader does not overlap with existing readers
for reader in cls._readers:
if not supported_versions.is_disjoint( # type: ignore[attr-defined]
reader._supported_versions
):
raise RuntimeError(
f"Single-file readers {reader.__name__} and {cls.__name__} "
"have overlapping supported version ranges"
)
cls._supported_versions = supported_versions
cls._creation_version = creation_version
cls._readers.append(cls)
@staticmethod
def get_version(path: Path | str) -> Version:
"""Read the file version.
Mojito light L1 files from before version 2.0.0 may not contain a root
``version`` attribute. Such files are treated as version ``1.0.0`` for backward
compatibility.
Parameters
----------
path
Path to the HDF5 file.
Returns
-------
The version of the file.
"""
with h5py.File(str(path), "r") as file:
version_attr = file.attrs.get("version", "1.0.0")
return Version(str(version_attr))
@property
@override
def version(self) -> Version:
"""Version of the file format, as stored in the file.
Files without a root ``version`` attribute are treated as version ``1.0.0``
(see :meth:`get_version`).
"""
return Version(str(self._opened_file.attrs.get("version", "1.0.0")))
@classmethod
def create(cls, path: Path | str, *, mode: FileMode = "w-") -> Self:
"""Create a versioned empty file and return its reader.
Parameters
----------
path
Path to the file to be created.
mode
File mode for creation. Only write modes ("a", "w", and "w-") are supported.
Returns
-------
The reader for the newly created file.
"""
if not is_write_mode(mode):
raise ValueError(
"Only write modes ('a', 'w', 'w-') are supported for file creation"
)
with h5py.File(str(path), mode) as file:
# File can be opened in 'a' mode, so file might already exist with a version
if "version" not in file.attrs:
file.attrs["version"] = str(cls._creation_version)
return cls(path, mode="r+")
def __enter__(self) -> Self:
"""Open the file and validate its version.
Raises
------
ValueError
If the file version is not supported.
"""
super().__enter__()
try:
version = self.get_version(self.path)
if version not in self._supported_versions:
raise ValueError(f"Unsupported Mojito L1 file version: {version}")
except BaseException:
super().__exit__(None, None, None)
raise
return self
[docs]
class MojitoL1SingleFileV1(
MojitoL1SingleFile,
supported_versions=SpecifierSet(">=1.0,<2.0"),
creation_version=Version("1.0.0"),
):
"""Single-file Mojito L1 reader for Mojito light files (>=1.0,<2.0).
This implementation supports reading and writing.
Metadata
--------
In this specific format, metadata only accept three keys: "lolipops_version",
"pipeline_name", and "laser_frequency". These can be accessed and modified through
the attributes :attr:`lolipops_version`, :attr:`pipeline_name`, and
:attr:`laser_frequency`, as well as the usual :attr:`metadata`.
An error is raised if any of these required metadata keys are missing, or if one
attempts to set another key.
Time and frequency grids
------------------------
Time and frequency grids are required as they define the temporal and spectral
resolution of the datasets. They can be modified, but it is the user's
responsibility to ensure consistency with the datasets.
Datasets
--------
The datasets in this format are organized under specific HDF5 groups. Most datasets
are lazily stacked before they are returned to the user. TDI datasets are lazily
rescaled by the inverse laser frequency to convert from frequency units (as saved in
the HDF5 file) to Doppler units (as required by the :class:`MojitoL1` protocol).
Datasets are read-write: when a dataset is missing from the HDF5 file, the accessors
create an empty dataset with the expected shape and dtype based on the associated
time or frequency grid.
File structure
--------------
Groups end with ``/`` and attributes start with ``@``. Dataset shapes use the
sizes of the time grids (``N_tdi``, ``N_ltt``, ``N_orb``, ``N_noise``) and of the
frequency grid (``N_freq``). Units are those stored in the file. ``{ij}`` stands
for one dataset per MOSA, in the order of :data:`lisaconstants.indexing.MOSAS`, and
``{i}`` for one dataset per spacecraft, in the order of
:data:`lisaconstants.indexing.SPACECRAFT`. Entries marked "not read" exist in
Mojito light files but are ignored by this reader.
::
/
|-- @version str, optional ("1.0.0" if missing)
|-- @pipeline_name str
|-- @lolipops_version str
|-- @laser_frequency float64 [Hz]
|-- @modulation_frequencies (2,) float64 [Hz], not read
|-- tdis/
| |-- sampling/ @t0, @dt, @size (@duration, @fs not read)
| |-- eta_{ij} (N_tdi,) float64 [Hz]
| |-- X2, Y2, Z2 (N_tdi,) float64 [Hz]
| |-- A2, E2, T2 (N_tdi,) float64 [Hz]
| |-- eta_flags (N_tdi,) int8, created if missing
| `-- tdi_flags (N_tdi,) int8, XYZ/AET flags, created if missing
|-- ltts/
| |-- sampling/ @t0, @dt, @size (@duration, @fs not read)
| |-- ltt_{ij} (N_ltt,) float64 [s]
| `-- ltt_derivative_{ij} (N_ltt,) float64 [s/s]
|-- orbits/
| |-- sampling/ @t0, @dt, @size (@duration, @fs not read)
| |-- sc_position_{i} (N_orb, 3) float64 [m]
| `-- sc_velocity_{i} (N_orb, 3) float64 [m/s]
|-- noise_estimates/
| |-- sampling/ @t0, @dt, @size (@duration, @fs not read)
| |-- log_frequency_sampling/ @fmin, @fmax, @size
| |-- eta (N_noise, N_freq, 6, 6) complex128 [Hz^2/Hz]
| |-- XYZ (N_noise, N_freq, 3, 3) complex128 [Hz^2/Hz]
| `-- AET (N_noise, N_freq, 3, 3) complex128 [Hz^2/Hz]
|-- scet_wrt_tcb/ not read
| |-- sampling/ @t0, @dt, @size, @duration, @fs
| `-- scet_wrt_tcb_{i} (N_scet,) float64
`-- ttl/ not read
|-- ttl_coefficients/
| |-- sampling/ @t0, @dt, @size, @duration, @fs
| `-- c_{ij}_{eta,phi}_{rx,tx}
| (N_ttl,) float64
`-- ttl_contribution_to_tdis/
|-- eta_{ij} (N_ttl,) float64
|-- X2, Y2, Z2 (N_ttl,) float64
`-- A2, E2, T2 (N_ttl,) float64
TDI and LTT datasets also carry ``@index_start`` and ``@index_stop`` attributes,
which are not read.
"""
def _read_uniform_time_grid(self, group_name: str) -> UniformTimeGrid:
"""Read a uniform time grid from an HDF5 group.
Parameters
----------
group_name
Name of the HDF5 group containing the uniform time grid attributes.
Returns
-------
The uniform time grid read from the HDF5 group.
Raises
------
KeyError
If the specified group does not exist in the HDF5 file.
ValueError
If any of the required attributes are missing or cannot be converted to the
expected types.
"""
# Retrieve the group
group = self._opened_file.get(group_name, None)
if group is None:
raise KeyError(f"Group '{group_name}' not found")
# Read the required attributes from the group
dt_attr = group.attrs.get("dt", None)
t0_attr = group.attrs.get("t0", None)
size_attr = group.attrs.get("size", None)
if dt_attr is None or t0_attr is None or size_attr is None:
raise ValueError(f"Missing required attributes in group '{group_name}'")
# Check and cast types
if not isinstance(dt_attr, np.floating):
raise ValueError(f"Attribute 'dt' in group '{group_name}' must be a float")
if not isinstance(t0_attr, np.floating):
raise ValueError(f"Attribute 't0' in group '{group_name}' must be a float")
if not isinstance(size_attr, (int, np.integer)):
raise ValueError(f"Attribute 'size' in group '{group_name}' must be an int")
return UniformTimeGrid(
t0=float(t0_attr), dt=float(dt_attr), size=int(size_attr)
)
def _write_uniform_time_grid(self, group_name: str, grid: UniformTimeGrid) -> None:
"""Write a uniform time grid to an HDF5 group.
Parameters
----------
group_name
Name of the HDF5 group to write the uniform time grid attributes to.
grid
The uniform time grid to write.
"""
group = self._opened_file.require_group(group_name)
group.attrs["t0"] = grid.t0
group.attrs["dt"] = grid.dt
group.attrs["size"] = grid.size
def _read_log_uniform_frequency_grid(
self, group_name: str
) -> LogUniformFrequencyGrid:
"""Read a log-uniform frequency grid from an HDF5 group.
Parameters
----------
group_name
Name of the HDF5 group containing the log-uniform frequency grid attributes.
Returns
-------
The log-uniform frequency grid read from the HDF5 group.
Raises
------
KeyError
If the specified group does not exist in the HDF5 file.
ValueError
If any of the required attributes are missing or cannot be converted to the
expected types.
"""
# Retrieve the group
group = self._opened_file.get(group_name, None)
if group is None:
raise KeyError(f"Group '{group_name}' not found")
# Read the required attributes from the group
fmin = group.attrs.get("fmin", None)
fmax = group.attrs.get("fmax", None)
size_attr = group.attrs.get("size", None)
if fmin is None or fmax is None or size_attr is None:
raise ValueError(f"Missing required attributes in group '{group_name}'")
# Check and cast types
if not isinstance(fmin, np.number):
raise ValueError(
f"Attribute 'fmin' in group '{group_name}' must be a float"
)
if not isinstance(fmax, np.number):
raise ValueError(
f"Attribute 'fmax' in group '{group_name}' must be a float"
)
if not isinstance(size_attr, (int, np.integer)):
raise ValueError(f"Attribute 'size' in group '{group_name}' must be an int")
return LogUniformFrequencyGrid(
fmin=float(fmin), fmax=float(fmax), size=int(size_attr)
)
def _write_log_uniform_frequency_grid(
self, group_name: str, grid: LogUniformFrequencyGrid
) -> None:
"""Write a log-uniform frequency grid to an HDF5 group.
Parameters
----------
group_name
Name of HDF5 group to write the log-uniform frequency grid attributes to.
grid
The log-uniform frequency grid to write.
"""
group = self._opened_file.require_group(group_name)
group.attrs["fmin"] = grid.fmin
group.attrs["fmax"] = grid.fmax
group.attrs["size"] = grid.size
@property
def pipeline_name(self) -> str:
"""Pipeline name."""
name_attr = self._opened_file.attrs.get("pipeline_name", None)
if name_attr is None or not isinstance(name_attr, str):
raise KeyError("Pipeline name not found in the Mojito L1 file")
return name_attr
@pipeline_name.setter
def pipeline_name(self, value: str) -> None:
self._opened_file.attrs["pipeline_name"] = str(value)
@property
def laser_frequency(self) -> float:
"""Laser frequency [Hz].
This property is retrieved from metadata and is mandatory as it is used to
convert frequency data to Doppler units.
"""
freq_attr = self._opened_file.attrs.get("laser_frequency", None)
if freq_attr is None or not isinstance(freq_attr, np.floating):
raise KeyError("Laser frequency not found in the Mojito L1 file")
return float(freq_attr)
@laser_frequency.setter
def laser_frequency(self, value: float) -> None:
self._opened_file.attrs["laser_frequency"] = float(value)
@property
def lolipops_version(self) -> str:
"""Lolipops version."""
version_attr = self._opened_file.attrs.get("lolipops_version", None)
if version_attr is None or not isinstance(version_attr, str):
raise KeyError("Lolipops version not found in the Mojito L1 file")
return version_attr
@lolipops_version.setter
def lolipops_version(self, value: str) -> None:
self._opened_file.attrs["lolipops_version"] = str(value)
@property
@override
def metadata(self) -> dict[str, Any]:
return {
"pipeline_name": self.pipeline_name,
"lolipops_version": self.lolipops_version,
"laser_frequency": self.laser_frequency,
}
@metadata.setter
def metadata(self, value: dict[str, Any]) -> None:
required_keys = {"pipeline_name", "lolipops_version", "laser_frequency"}
missing_keys = required_keys - value.keys()
if missing_keys:
raise ValueError(f"Missing required metadata keys: {missing_keys}")
extra_keys = value.keys() - required_keys
if extra_keys:
raise ValueError(f"Unexpected metadata keys: {extra_keys}")
self.pipeline_name = value["pipeline_name"]
self.lolipops_version = value["lolipops_version"]
self.laser_frequency = value["laser_frequency"]
@property
@override
def tdi_time_grid(self) -> UniformTimeGrid:
return self._read_uniform_time_grid("tdis/sampling")
@tdi_time_grid.setter
def tdi_time_grid(self, value: UniformTimeGrid) -> None:
self._write_uniform_time_grid("tdis/sampling", value)
@property
@override
def eta(self) -> Dataset:
# Retrieve the datasets, convert them to Doppler units, and stack them
ds_hz = [
self._require_dataset(f"tdis/eta_{mosa}", shape=(self.tdi_time_grid.size,))
for mosa in MOSAS
]
ds_doppler = [LazyScaledDataset(ds, 1.0 / self.laser_frequency) for ds in ds_hz]
return LazyStackedDataset(ds_doppler)
@property
@override
def xyz(self) -> Dataset:
# Retrieve the datasets, convert them to Doppler units, and stack them
ds_names = ["X2", "Y2", "Z2"]
ds_hz = [
self._require_dataset(f"tdis/{name}", shape=(self.tdi_time_grid.size,))
for name in ds_names
]
ds_doppler = [LazyScaledDataset(ds, 1.0 / self.laser_frequency) for ds in ds_hz]
return LazyStackedDataset(ds_doppler)
@property
@override
def aet(self) -> Dataset:
# Retrieve the datasets, convert them to Doppler units, and stack them
ds_names = ["A2", "E2", "T2"]
ds_hz = [
self._require_dataset(f"tdis/{name}", shape=(self.tdi_time_grid.size,))
for name in ds_names
]
ds_doppler = [LazyScaledDataset(ds, 1.0 / self.laser_frequency) for ds in ds_hz]
return LazyStackedDataset(ds_doppler)
@property
@override
def eta_flags(self) -> Dataset:
return self._require_dataset(
"tdis/eta_flags", shape=(self.tdi_time_grid.size,), dtype=np.int8
)
@property
@override
def xyz_flags(self) -> Dataset:
return self._require_dataset(
"tdis/tdi_flags", shape=(self.tdi_time_grid.size,), dtype=np.int8
)
@property
@override
def aet_flags(self) -> Dataset:
# Same as XYZ flags
return self.xyz_flags
@property
@override
def ltt_time_grid(self) -> UniformTimeGrid:
return self._read_uniform_time_grid("ltts/sampling")
@ltt_time_grid.setter
def ltt_time_grid(self, value: UniformTimeGrid) -> None:
self._write_uniform_time_grid("ltts/sampling", value)
@property
@override
def ltts(self) -> Dataset:
# Retrieve the LTT datasets and stack them
datasets = [
self._require_dataset(f"ltts/ltt_{mosa}", shape=(self.ltt_time_grid.size,))
for mosa in MOSAS
]
return LazyStackedDataset(datasets)
@property
@override
def ltt_derivatives(self) -> Dataset:
# Retrieve the LTT datasets and stack them
datasets = [
self._require_dataset(
f"ltts/ltt_derivative_{mosa}", shape=(self.ltt_time_grid.size,)
)
for mosa in MOSAS
]
return LazyStackedDataset(datasets)
@property
@override
def orbits_time_grid(self) -> UniformTimeGrid:
return self._read_uniform_time_grid("orbits/sampling")
@orbits_time_grid.setter
def orbits_time_grid(self, value: UniformTimeGrid) -> None:
self._write_uniform_time_grid("orbits/sampling", value)
@property
@override
def spacecraft_positions(self) -> Dataset:
# Retrieve the position datasets and stack them
datasets = [
self._require_dataset(
f"orbits/sc_position_{sc}", shape=(self.orbits_time_grid.size, 3)
)
for sc in SPACECRAFT
]
return LazyStackedDataset(datasets, axis=1)
@property
@override
def spacecraft_velocities(self) -> Dataset:
# Retrieve the velocity datasets and stack them
datasets = [
self._require_dataset(
f"orbits/sc_velocity_{sc}", shape=(self.orbits_time_grid.size, 3)
)
for sc in SPACECRAFT
]
return LazyStackedDataset(datasets, axis=1)
@property
@override
def noise_estimates_time_grid(self) -> UniformTimeGrid:
return self._read_uniform_time_grid("noise_estimates/sampling")
@noise_estimates_time_grid.setter
def noise_estimates_time_grid(self, value: UniformTimeGrid) -> None:
self._write_uniform_time_grid("noise_estimates/sampling", value)
@property
@override
def noise_estimates_frequency_grid(self) -> LogUniformFrequencyGrid:
return self._read_log_uniform_frequency_grid(
"noise_estimates/log_frequency_sampling"
)
@noise_estimates_frequency_grid.setter
def noise_estimates_frequency_grid(self, value: LogUniformFrequencyGrid) -> None:
self._write_log_uniform_frequency_grid(
"noise_estimates/log_frequency_sampling", value
)
@property
@override
def eta_noise_estimates(self) -> Dataset:
# Retrieve dataset in Hz, and convert to Doppler units
ds_hz = self._require_dataset(
"noise_estimates/eta",
shape=(
self.noise_estimates_time_grid.size,
self.noise_estimates_frequency_grid.size,
6,
6,
),
dtype=np.complex128,
)
return LazyScaledDataset(ds_hz, 1.0 / self.laser_frequency**2)
@property
@override
def xyz_noise_estimates(self) -> Dataset:
# Retrieve dataset in Hz, and convert to Doppler units
ds_hz = self._require_dataset(
"noise_estimates/XYZ",
shape=(
self.noise_estimates_time_grid.size,
self.noise_estimates_frequency_grid.size,
3,
3,
),
dtype=np.complex128,
)
return LazyScaledDataset(ds_hz, 1.0 / self.laser_frequency**2)
@property
@override
def aet_noise_estimates(self) -> Dataset:
# Retrieve dataset in Hz, and convert to Doppler units
ds_hz = self._require_dataset(
"noise_estimates/AET",
shape=(
self.noise_estimates_time_grid.size,
self.noise_estimates_frequency_grid.size,
3,
3,
),
dtype=np.complex128,
)
return LazyScaledDataset(ds_hz, 1.0 / self.laser_frequency**2)
[docs]
class MojitoL1SingleFileV2(
MojitoL1SingleFile,
supported_versions=SpecifierSet(">=2.0,<3.0"),
creation_version=Version("2.0.0"),
):
"""Single-file Mojito L1 reader for Mojito medium and heavy files (>=2.0,<3.0).
This implementation supports reading and writing.
Metadata
--------
Metadata are stored as a dictionary of key-value pairs. Keys must be strings and
values must be JSON-serializable, as the dictionary is stored as a JSON string in
the ``metadata`` attribute of the root group.
These key-value pairs can be used as queryables in the GlobalIndex.
Time and frequency grids
------------------------
Time and frequency grids are stored as attributes of the corresponding HDF5 group
(see :attr:`UNIFORM_TIME_GRID_GROUP` and :attr:`LOG_UNIFORM_FREQUENCY_GRID_GROUP`),
encoded as JSON strings.
They define the temporal and spectral resolution of the datasets. They can be
modified, but it is the user's responsibility to ensure consistency with the
datasets.
Datasets
--------
The datasets in this format are stored directly as HDF5 datasets at the file root.
TDI and noise-estimate datasets are expressed in Doppler units (dimensionless), so
no conversion is needed in this reader.
Datasets are read-write: when a dataset is missing from the HDF5 file, the accessors
create an empty dataset with the expected shape and dtype based on the associated
time or frequency grid.
AET quantities are not stored in the file. :attr:`aet` and
:attr:`aet_noise_estimates` are computed on the fly from :attr:`xyz` and
:attr:`xyz_noise_estimates`, and writing to them writes the corresponding XYZ
values. Only whole rows (slices along the first axis) can be written to them; for
example, ``aet[:, 0] = ...`` raises a :class:`ValueError`. :attr:`aet_flags` is the
same as :attr:`xyz_flags`.
File structure
--------------
Groups end with ``/`` and attributes start with ``@``. Dataset shapes use the
sizes of the time grids (``N_tdi``, ``N_ltt``, ``N_orb``, ``N_noise``) and of the
frequency grid (``N_freq``). Units are those stored in the file. MOSA axes follow
:data:`lisaconstants.indexing.MOSAS` and spacecraft axes follow
:data:`lisaconstants.indexing.SPACECRAFT`.
::
/
|-- @version str, file format version, e.g. "2.0.0"
|-- @metadata str, JSON-encoded dict
|-- uniform_time_grids/
| |-- @tdi str, JSON {"t0", "dt", "size"}
| |-- @ltts str, JSON {"t0", "dt", "size"}
| |-- @orbits str, JSON {"t0", "dt", "size"}
| `-- @noise_estimates str, JSON {"t0", "dt", "size"}
|-- log_uniform_frequency_grids/
| `-- @noise_estimates str, JSON {"fmin", "fmax", "size"}
|-- eta (N_tdi, 6) float64 [dimensionless]
|-- xyz (N_tdi, 3) float64 [dimensionless]
|-- eta_flags (N_tdi,) int8
|-- xyz_flags (N_tdi,) int8
|-- ltts (N_ltt, 6) float64 [s]
|-- ltt_derivatives (N_ltt, 6) float64 [s/s]
|-- spacecraft_positions (N_orb, 3, 3) float64 [m]
|-- spacecraft_velocities (N_orb, 3, 3) float64 [m/s]
|-- eta_noise_estimates (N_noise, N_freq, 6, 6) complex128 [/Hz]
`-- xyz_noise_estimates (N_noise, N_freq, 3, 3) complex128 [/Hz]
"""
UNIFORM_TIME_GRID_GROUP = "uniform_time_grids"
"""HDF5 group for uniform time grids."""
LOG_UNIFORM_FREQUENCY_GRID_GROUP = "log_uniform_frequency_grids"
"""HDF5 group for log-uniform frequency grids."""
def _read_grid_attributes(
self, group_name: str, name: str, kind: str, keys: tuple[str, ...]
) -> dict[str, Any]:
"""Read and decode the JSON-encoded attributes of a grid.
Parameters
----------
group_name
Name of the HDF5 group holding the grid attributes.
name
Name identifying the grid, used as attribute name in the group.
kind
Kind of grid ("time" or "frequency"), used in error messages.
keys
Keys that must be present (and not null) in the decoded JSON object.
Returns
-------
The decoded grid attributes.
Raises
------
KeyError
If the group or the grid with the specified name does not exist.
ValueError
If the grid attributes are malformed or a required key is missing.
"""
group = self._opened_file.get(group_name)
json_data = None if group is None else group.attrs.get(name, None)
if json_data is None:
raise KeyError(f"{kind.capitalize()} grid '{name}' not found")
if not isinstance(json_data, str):
raise ValueError(
f"{kind.capitalize()} grid attributes for '{name}' must be a JSON "
"string"
)
try:
data = json.loads(json_data)
except json.JSONDecodeError as e:
raise ValueError(f"Malformed {kind} grid attributes for '{name}'") from e
if not isinstance(data, dict):
raise ValueError(f"Malformed {kind} grid attributes for '{name}'")
if any(data.get(key) is None for key in keys):
raise ValueError(f"Missing required attributes for {kind} grid '{name}'")
return data
def _write_grid_attributes(
self, group_name: str, name: str, kind: str, data: dict[str, Any]
) -> None:
"""Encode grid attributes as a JSON string and store them in a group.
Parameters
----------
group_name
Name of the HDF5 group holding the grid attributes, created if missing.
name
Name identifying the grid, used as attribute name in the group.
kind
Kind of grid ("time" or "frequency"), used in error messages.
data
Grid attributes to encode.
Raises
------
ValueError
If the grid attributes cannot be encoded as standard JSON (e.g. NaN).
"""
try:
json_data = json.dumps(data, allow_nan=False)
except ValueError as e:
raise ValueError(f"Cannot encode {kind} grid '{name}' as JSON: {e}") from e
self._opened_file.require_group(group_name).attrs[name] = json_data
def _read_uniform_time_grid(self, name: str) -> UniformTimeGrid:
"""Read a uniform time grid.
The time grid properties (t0, dt, size) are read from the HDF5 file's
:attr:`UNIFORM_TIME_GRID_GROUP` group attribute with the given name and decoded
from JSON. They are used to construct a :class:`UniformTimeGrid` object.
Parameters
----------
name
Name identifying the uniform time grid.
Returns
-------
The uniform time grid.
Raises
------
KeyError
If a time grid with the specified name does not exist in the HDF5 file.
ValueError
If the time grid attributes are malformed or missing.
"""
data = self._read_grid_attributes(
self.UNIFORM_TIME_GRID_GROUP, name, "time", ("t0", "dt", "size")
)
for key in ("t0", "dt"):
if not _is_number(data[key]):
raise ValueError(
f"Attribute '{key}' for time grid '{name}' must be a number"
)
if not _is_integer(data["size"]):
raise ValueError(
f"Attribute 'size' for time grid '{name}' must be an integer"
)
return UniformTimeGrid(
t0=float(data["t0"]), dt=float(data["dt"]), size=int(data["size"])
)
def _write_uniform_time_grid(self, name: str, grid: UniformTimeGrid) -> None:
"""Write a uniform time grid.
The time grid properties (t0, dt, size) are encoded in a JSON string and stored
as the file :attr:`UNIFORM_TIME_GRID_GROUP` group's attribute with the given
name.
Parameters
----------
name
Name identifying the uniform time grid.
grid
The uniform time grid to write.
Raises
------
ValueError
If the grid properties cannot be encoded as standard JSON (e.g. NaN).
"""
data = {"t0": grid.t0, "dt": grid.dt, "size": grid.size}
self._write_grid_attributes(self.UNIFORM_TIME_GRID_GROUP, name, "time", data)
def _read_log_uniform_frequency_grid(self, name: str) -> LogUniformFrequencyGrid:
"""Read a log-uniform frequency grid.
The frequency grid properties (fmin, fmax, size) are read from the HDF5 file's
:attr:`LOG_UNIFORM_FREQUENCY_GRID_GROUP` group attribute with the given name and
decoded from JSON. They are used to construct a :class:`LogUniformFrequencyGrid`
object.
Parameters
----------
name
Name identifying the log-uniform frequency grid.
Returns
-------
The log-uniform frequency grid.
Raises
------
KeyError
If a frequency grid with the specified name does not exist in the HDF5 file.
ValueError
If the frequency grid attributes are malformed or missing.
"""
data = self._read_grid_attributes(
self.LOG_UNIFORM_FREQUENCY_GRID_GROUP,
name,
"frequency",
("fmin", "fmax", "size"),
)
for key in ("fmin", "fmax"):
if not _is_number(data[key]):
raise ValueError(
f"Attribute '{key}' for frequency grid '{name}' must be a number"
)
if not _is_integer(data["size"]):
raise ValueError(
f"Attribute 'size' for frequency grid '{name}' must be an integer"
)
return LogUniformFrequencyGrid(
fmin=float(data["fmin"]), fmax=float(data["fmax"]), size=int(data["size"])
)
def _write_log_uniform_frequency_grid(
self, name: str, grid: LogUniformFrequencyGrid
) -> None:
"""Write a log-uniform frequency grid.
The frequency grid properties (fmin, fmax, size) are encoded in a JSON string
and stored as the file :attr:`LOG_UNIFORM_FREQUENCY_GRID_GROUP` group's
attribute with the given name.
Parameters
----------
name
Name identifying the log-uniform frequency grid.
grid
The log-uniform frequency grid to write.
Raises
------
ValueError
If the grid properties cannot be encoded as standard JSON (e.g. NaN).
"""
data = {"fmin": grid.fmin, "fmax": grid.fmax, "size": grid.size}
self._write_grid_attributes(
self.LOG_UNIFORM_FREQUENCY_GRID_GROUP, name, "frequency", data
)
@property
@override
def metadata(self) -> dict[str, Any]:
json_data = self._opened_file.attrs.get("metadata", "{}")
if not isinstance(json_data, str):
raise ValueError("Invalid metadata format in HDF5 file")
try:
metadata = json.loads(json_data)
except json.JSONDecodeError as e:
raise ValueError("Malformed metadata attributes") from e
if not isinstance(metadata, dict):
raise ValueError("Invalid metadata format in HDF5 file")
return metadata
@metadata.setter
def metadata(self, value: dict[str, Any]) -> None:
if not isinstance(value, dict):
raise ValueError("Metadata must be a dictionary")
if not all(isinstance(key, str) for key in value):
raise ValueError("Metadata keys must be strings")
try:
json_data = json.dumps(value, allow_nan=False)
except (TypeError, ValueError) as e:
raise ValueError(f"Metadata values must be JSON-serializable: {e}") from e
self._opened_file.attrs["metadata"] = json_data
@property
@override
def tdi_time_grid(self) -> UniformTimeGrid:
return self._read_uniform_time_grid("tdi")
@tdi_time_grid.setter
def tdi_time_grid(self, value: UniformTimeGrid) -> None:
self._write_uniform_time_grid("tdi", value)
@property
@override
def eta(self) -> Dataset:
return self._require_dataset("eta", shape=(self.tdi_time_grid.size, 6))
@property
@override
def xyz(self) -> Dataset:
return self._require_dataset("xyz", shape=(self.tdi_time_grid.size, 3))
@property
@override
def aet(self) -> Dataset:
# Retrieve the XYZ datasets and form A, E and T
return LazyMappedDataset(
self.xyz,
func=lambda x: x @ _XYZ_TO_AET.T,
# _XYZ_TO_AET is orthonormal, so its inverse is its transpose
inverse_func=lambda x: x @ _XYZ_TO_AET,
is_multivariate=True,
)
@property
@override
def eta_flags(self) -> Dataset:
return self._require_dataset(
"eta_flags", shape=(self.tdi_time_grid.size,), dtype=np.int8
)
@property
@override
def xyz_flags(self) -> Dataset:
return self._require_dataset(
"xyz_flags", shape=(self.tdi_time_grid.size,), dtype=np.int8
)
@property
@override
def aet_flags(self) -> Dataset:
# Same as XYZ flags
return self.xyz_flags
@property
@override
def ltt_time_grid(self) -> UniformTimeGrid:
return self._read_uniform_time_grid("ltts")
@ltt_time_grid.setter
def ltt_time_grid(self, value: UniformTimeGrid) -> None:
self._write_uniform_time_grid("ltts", value)
@property
@override
def ltts(self) -> Dataset:
return self._require_dataset("ltts", shape=(self.ltt_time_grid.size, 6))
@property
@override
def ltt_derivatives(self) -> Dataset:
return self._require_dataset(
"ltt_derivatives", shape=(self.ltt_time_grid.size, 6)
)
@property
@override
def orbits_time_grid(self) -> UniformTimeGrid:
return self._read_uniform_time_grid("orbits")
@orbits_time_grid.setter
def orbits_time_grid(self, value: UniformTimeGrid) -> None:
self._write_uniform_time_grid("orbits", value)
@property
@override
def spacecraft_positions(self) -> Dataset:
return self._require_dataset(
"spacecraft_positions", shape=(self.orbits_time_grid.size, 3, 3)
)
@property
@override
def spacecraft_velocities(self) -> Dataset:
return self._require_dataset(
"spacecraft_velocities", shape=(self.orbits_time_grid.size, 3, 3)
)
@property
@override
def noise_estimates_time_grid(self) -> UniformTimeGrid:
return self._read_uniform_time_grid("noise_estimates")
@noise_estimates_time_grid.setter
def noise_estimates_time_grid(self, value: UniformTimeGrid) -> None:
self._write_uniform_time_grid("noise_estimates", value)
@property
@override
def noise_estimates_frequency_grid(self) -> LogUniformFrequencyGrid:
return self._read_log_uniform_frequency_grid("noise_estimates")
@noise_estimates_frequency_grid.setter
def noise_estimates_frequency_grid(self, value: LogUniformFrequencyGrid) -> None:
self._write_log_uniform_frequency_grid("noise_estimates", value)
@property
@override
def eta_noise_estimates(self) -> Dataset:
return self._require_dataset(
"eta_noise_estimates",
shape=(
self.noise_estimates_time_grid.size,
self.noise_estimates_frequency_grid.size,
6,
6,
),
dtype=np.complex128,
)
@property
@override
def xyz_noise_estimates(self) -> Dataset:
return self._require_dataset(
"xyz_noise_estimates",
shape=(
self.noise_estimates_time_grid.size,
self.noise_estimates_frequency_grid.size,
3,
3,
),
dtype=np.complex128,
)
@property
@override
def aet_noise_estimates(self) -> Dataset:
# Transform the XYZ covariance matrices S into AET as M S M^T, using the same
# double-precision contraction as the simulation pipeline. Since M is
# orthonormal, the inverse transformation is M^T S M.
return LazyMappedDataset(
self.xyz_noise_estimates,
func=lambda s: np.einsum("ik,...kl,jl->...ij", _XYZ_TO_AET, s, _XYZ_TO_AET),
inverse_func=lambda s: np.einsum(
"ki,...kl,lj->...ij", _XYZ_TO_AET, s, _XYZ_TO_AET
),
is_multivariate=True,
)
[docs]
def open_single_file_l1(path: Path | str, *, mode: FileMode) -> MojitoL1:
"""Select the appropriate single-file Mojito L1 class based on file.
The version is read from the file and matched against the supported version ranges
of the concrete readers registered on :class:`MojitoL1SingleFile`. This function
only opens existing files; use :func:`create_l1` to create a new file.
Parameters
----------
path
Path to the Mojito L1 file.
mode
File mode to use when opening the Mojito L1 file.
Returns
-------
The appropriate Mojito L1 class for the given file.
Raises
------
ValueError
If the file version is unsupported or cannot be determined.
"""
if is_write_mode(mode):
raise ValueError("Use create_l1 to create a new Mojito L1 file")
# Read the file version
version = MojitoL1SingleFile.get_version(path)
# Select the appropriate class based on the version
for reader in MojitoL1SingleFile._readers:
if version in reader._supported_versions:
return reader(path, mode=mode)
raise ValueError(f"Unsupported Mojito L1 file version: {version}")
[docs]
def create_l1(
path: Path | str, version: Version | str, *, mode: FileMode = "w"
) -> MojitoL1:
"""Create a new Mojito L1 file.
Parameters
----------
path
Path to the file to create.
version
File format version to write and use for reader selection.
mode
File mode for creation. Write modes "a", "w", and "w-" are supported.
Returns
-------
The reader for the newly created file.
Raises
------
ValueError
If no registered reader supports ``version``.
"""
file_version = version if isinstance(version, Version) else Version(version)
for reader in MojitoL1SingleFile._readers:
if file_version in reader._supported_versions:
return reader.create(path, mode=mode)
raise ValueError(f"Unsupported Mojito L1 file version: {file_version}")
[docs]
class MojitoL1MultiFile(MojitoL1):
"""Multi-file Mojito L1 reader that handles proper dataset combination.
This implementation is read-only.
Metadata
--------
Contains the combined metadata from all files, grouped by key into lists.
Time and frequency grids
------------------------
Time and frequency grids are required to be consistent across all files.
Datasets
--------
The datasets from all files are combined using the appropriate aggregation method.
Typically, measurements, such as TDI observables, are summed across all files while
data quality flags are typically combined using logical operations (e.g., OR).
Aggregation operations are applied lazily, meaning that the actual combination of
datasets occurs only when they are accessed, and on the specific slices requested.
Parameters
----------
path
Path to the directory containing the Mojito L1 files.
Raises
------
ValueError
If no paths are provided for the Mojito L1 multi-file reader.
"""
def __init__(self, paths: list[Path | str]) -> None:
self.paths = [str(path) for path in paths]
self._readers: list[MojitoL1] | None = None
if not self.paths:
raise ValueError("No paths provided for Mojito L1 multi-file reader")
def __enter__(self) -> Self:
"""Open all single-file Mojito L1 readers and check they are compatible."""
self._readers = []
try: # try-except necessary to handle failures during a reader initialization
for path in self.paths:
reader = open_single_file_l1(path, mode="r")
self._readers.append(reader)
reader.__enter__()
self._check_compatibility()
except BaseException as exc:
for reader in self._readers:
reader.__exit__(type(exc), exc, exc.__traceback__)
self._readers = None
raise
return self
def __exit__(self, exc_type, exc_val, exc_tb) -> None:
"""Exit the context manager and close all single-file Mojito L1 readers."""
if self._readers is not None:
for reader in self._readers:
reader.__exit__(exc_type, exc_val, exc_tb)
self._readers = None
@property
def _opened_readers(self) -> list[MojitoL1]:
"""Ensure that all single-file Mojito L1 readers are opened and return them.
Raises
------
RuntimeError
If the multi-file reader has not been opened yet.
Returns
-------
The list of opened single-file Mojito L1 readers.
"""
if self._readers is None:
raise RuntimeError("Mojito L1 multi-file reader is not opened")
return self._readers
def _check_compatibility(self) -> None:
"""Check that all single-file Mojito L1 readers are compatible.
In particular, we check that
- All files have the same version
- All files have identical time and frequency grids (all of them)
- All files have the same first 200 samples of LTT and LTT derivatives
- All files have the same spacecraft positions and velocities
Note that we never compare entire arrays, but only compare the first 200 samples
as a cheap but indicative check.
"""
# Check they are of the same version
versions = {reader.version for reader in self._opened_readers}
if len(versions) > 1:
raise ValueError("Incompatible Mojito L1 file versions")
# Check time grids are identical
tdi_time_grids = {reader.tdi_time_grid for reader in self._opened_readers}
if len(tdi_time_grids) > 1:
raise ValueError("Incompatible TDI time grids")
ltt_time_grids = {reader.ltt_time_grid for reader in self._opened_readers}
if len(ltt_time_grids) > 1:
raise ValueError("Incompatible LTT time grids")
orbits_time_grids = {reader.orbits_time_grid for reader in self._opened_readers}
if len(orbits_time_grids) > 1:
raise ValueError("Incompatible orbits time grids")
noise_estimates_time_grids = {
reader.noise_estimates_time_grid for reader in self._opened_readers
}
if len(noise_estimates_time_grids) > 1:
raise ValueError("Incompatible noise estimates time grids")
noise_estimates_frequency_grids = {
reader.noise_estimates_frequency_grid for reader in self._opened_readers
}
if len(noise_estimates_frequency_grids) > 1:
raise ValueError("Incompatible noise estimates frequency grids")
# Check the first 200 samples of LTT and LTT derivatives are identical
ltt_baseline = self._opened_readers[0].ltts[:200]
for reader in self._opened_readers[1:]:
if not np.allclose(ltt_baseline, reader.ltts[:200]):
raise ValueError("Incompatible LTTs")
ltt_derivative_baseline = self._opened_readers[0].ltt_derivatives[:200]
for reader in self._opened_readers[1:]:
if not np.allclose(ltt_derivative_baseline, reader.ltt_derivatives[:200]):
raise ValueError("Incompatible LTT derivatives")
# Check the first 200 samples of spacecraft positions and velocities
positions_baseline = self._opened_readers[0].spacecraft_positions[:200]
for reader in self._opened_readers[1:]:
if not np.allclose(positions_baseline, reader.spacecraft_positions[:200]):
raise ValueError("Incompatible spacecraft positions")
velocities_baseline = self._opened_readers[0].spacecraft_velocities[:200]
for reader in self._opened_readers[1:]:
if not np.allclose(velocities_baseline, reader.spacecraft_velocities[:200]):
raise ValueError("Incompatible spacecraft velocities")
@property
@override
def version(self) -> Version:
# We know all files have the same version
return self._opened_readers[0].version
@property
@override
def metadata(self) -> dict[str, Any]:
"""Combine metadata from all files.
When multiple files define the same metadata key, a list of values is created
for that key, in the same order as the files are provided.
"""
combined_metadata: dict[str, Any] = defaultdict(list)
for reader in self._opened_readers:
for key, value in reader.metadata.items():
combined_metadata[key].append(value)
return dict(combined_metadata)
@property
@override
def tdi_time_grid(self) -> UniformTimeGrid:
# We know all files have the same TDI time grid
return self._opened_readers[0].tdi_time_grid
@property
@override
def eta(self) -> Dataset:
return LazySumDataset([reader.eta for reader in self._opened_readers])
@property
@override
def xyz(self) -> Dataset:
return LazySumDataset([reader.xyz for reader in self._opened_readers])
@property
@override
def aet(self) -> Dataset:
return LazySumDataset([reader.aet for reader in self._opened_readers])
@property
@override
def eta_flags(self) -> Dataset:
return LazyBooleanOrDataset(
[reader.eta_flags for reader in self._opened_readers],
dtype=np.int8,
)
@property
@override
def xyz_flags(self) -> Dataset:
return LazyBooleanOrDataset(
[reader.xyz_flags for reader in self._opened_readers],
dtype=np.int8,
)
@property
@override
def aet_flags(self) -> Dataset:
return LazyBooleanOrDataset(
[reader.aet_flags for reader in self._opened_readers],
dtype=np.int8,
)
@property
@override
def ltt_time_grid(self) -> UniformTimeGrid:
# We know all files have the same LTT time grid
return self._opened_readers[0].ltt_time_grid
@property
@override
def ltts(self) -> Dataset:
# We know all files have the same LTTs
return self._opened_readers[0].ltts
@property
@override
def ltt_derivatives(self) -> Dataset:
# We know all files have the same LTT derivatives
return self._opened_readers[0].ltt_derivatives
@property
@override
def orbits_time_grid(self) -> UniformTimeGrid:
# We know all files have the same orbits time grid
return self._opened_readers[0].orbits_time_grid
@property
@override
def spacecraft_positions(self) -> Dataset:
# We know all files have the same spacecraft positions
return self._opened_readers[0].spacecraft_positions
@property
@override
def spacecraft_velocities(self) -> Dataset:
# We know all files have the same spacecraft velocities
return self._opened_readers[0].spacecraft_velocities
@property
@override
def noise_estimates_time_grid(self) -> UniformTimeGrid:
# We know all files have the same noise estimate time grid
return self._opened_readers[0].noise_estimates_time_grid
@property
@override
def noise_estimates_frequency_grid(self) -> LogUniformFrequencyGrid:
# We know all files have the same noise estimate frequency grid
return self._opened_readers[0].noise_estimates_frequency_grid
@property
@override
def eta_noise_estimates(self) -> Dataset:
return LazySumDataset(
[reader.eta_noise_estimates for reader in self._opened_readers]
)
@property
@override
def xyz_noise_estimates(self) -> Dataset:
return LazySumDataset(
[reader.xyz_noise_estimates for reader in self._opened_readers]
)
@property
@override
def aet_noise_estimates(self) -> Dataset:
return LazySumDataset(
[reader.aet_noise_estimates for reader in self._opened_readers]
)
def open_multiple_files_l1(paths: list[Path | str]) -> MojitoL1MultiFile:
"""Open multiple Mojito L1 files as a combined reader.
Parameters
----------
paths
List of paths to the Mojito L1 files.
Returns
-------
A combined reader for the provided Mojito L1 files.
"""
return MojitoL1MultiFile(paths)
[docs]
def open_l1(paths: Path | str | list[Path | str], *, mode: FileMode = "r") -> MojitoL1:
"""Open one or more Mojito L1 files.
If multiple paths are provided, the L1 datasets are combined appropriately.
Typically, TDI and noise estimate datasets are summed across the provided files;
data quality flagged are combined using a logical OR operation.
Parameters
----------
paths
Path or list of paths to the Mojito L1 files.
mode
File mode to use when opening the files.
Returns
-------
MojitoL1
The appropriate Mojito L1 class for the given file(s).
Raises
------
ValueError
If multiple files are provided and the mode is not "r".
ValueError
If the provided paths are invalid.
"""
if isinstance(paths, (Path, str)):
return open_single_file_l1(paths, mode=mode)
elif isinstance(paths, list):
if mode != "r":
raise ValueError("Only read mode is supported for multiple files")
return open_multiple_files_l1(paths)
else:
raise ValueError("Invalid paths, provide a Path, str, or a list")