IO API

read_nwb(path, mode='a', return_io=False)

read_nwb(path: str, mode: str = 'a') -> NWBFile

read_nwb(
    path: str, mode: str = "a", return_io: bool = True
) -> tuple[NWBFile, NWBHDF5IO]

Read an NWB file.

Parameters:

• path (str) –

  Path to the NWB file.

• mode (str, default: 'a' ) –

  File read mode (i.e. read/write/append). Defaults to "a".

• return_io (bool, default: False ) –

  Return IO object alongside the NWB file object. Defaults to False.

Returns:

• NWBFile ( NWBFile | tuple[NWBFile, NWBHDF5IO] ) –

  NWB file object.

• NWBHDF5IO ( NWBFile | tuple[NWBFile, NWBHDF5IO] ) –

  IO object (if return_io=True).

Source code in src/mesoscopy/io.py

def read_nwb(path: str, mode: str = "a", return_io: bool = False) -> NWBFile | tuple[NWBFile, NWBHDF5IO]:
    """Read an NWB file.

    Args:
        path (str): Path to the NWB file.
        mode (str, optional): File read mode (i.e. read/write/append). Defaults to "a".
        return_io (bool, optional): Return IO object alongside the NWB file object. Defaults to False.

    Returns:
        NWBFile: NWB file object.
        NWBHDF5IO: IO object (if return_io=True).
    """
    from pynwb import NWBHDF5IO

    io = NWBHDF5IO(path, mode=mode)
    nwbfile = io.read()
    if return_io:
        return nwbfile, io
    return nwbfile

write_nwb(path, nwbfile, mode='w', io=None, **kwargs)

Write an NWB file.

Parameters:

• path (str) –

  Path to the NWB file.

• nwbfile (NWBFile) –

  NWB file object.

• mode (str, default: 'w' ) –

  File write mode (i.e. write/append). Defaults to "w".

• io (NWBHDF5IO, default: None ) –

  An already open IO object to write through. When given, the file is not reopened and path and mode are ignored. Defaults to None.

• **kwargs (Any, default: {} ) –

  Parameters passed to NWBHDF5IO.write.

Source code in src/mesoscopy/io.py

def write_nwb(path: str, nwbfile: NWBFile, mode: str = "w", io: NWBHDF5IO = None, **kwargs: typing.Any) -> None:
    """Write an NWB file.

    Args:
        path (str): Path to the NWB file.
        nwbfile (NWBFile): NWB file object.
        mode (str, optional): File write mode (i.e. write/append). Defaults to "w".
        io (NWBHDF5IO, optional): An already open IO object to write through. When given, the file
            is not reopened and `path` and `mode` are ignored. Defaults to None.
        **kwargs (typing.Any): Parameters passed to NWBHDF5IO.write.
    """
    from pynwb import NWBHDF5IO

    if io:
        return io.write(nwbfile, **kwargs)
    with NWBHDF5IO(path, mode=mode) as io:
        return io.write(nwbfile, **kwargs)

read_h5(path)

Read an HDF5 file.

Parameters:

• path (str) –

  Path to the HDF5 file.

Returns:

• File –

  h5py.File: HDF5 file object.

Source code in src/mesoscopy/io.py

def read_h5(path: str) -> h5py.File:
    """Read an HDF5 file.

    Args:
        path (str): Path to the HDF5 file.

    Returns:
        h5py.File: HDF5 file object.
    """
    return h5py.File(path, "r")

write_h5(path, data, compression='lzf', attributes={})

Write a dictionary to an HDF5 file.

Parameters:

• path (str) –

  Path to the HDF5 file.

• data (dict) –

  Dictionary containing datasets to write in {'dataset_name': data_array} format.

• compression (str, default: 'lzf' ) –

  Compression method for the datasets. Defaults to "lzf".

• attributes (dict, default: {} ) –

  Attributes to write to the HDF5 file. Defaults to {}.

Returns:

• str ( str ) –

  Path to the written HDF5 file.

Example

data = {"dataset1": np.array([1, 2, 3]), "dataset2": np.array([[1, 2], [3, 4]])} write_h5("output.h5", data)

Source code in src/mesoscopy/io.py

def write_h5(path: str, data: dict, compression: str = "lzf", attributes: dict = {}) -> str:
    """Write a dictionary to an HDF5 file.

    Args:
        path (str): Path to the HDF5 file.
        data (dict): Dictionary containing datasets to write in {'dataset_name': data_array} format.
        compression (str, optional): Compression method for the datasets. Defaults to "lzf".
        attributes (dict, optional): Attributes to write to the HDF5 file. Defaults to {}.

    Returns:
        str: Path to the written HDF5 file.

    Example:
        >>> data = {"dataset1": np.array([1, 2, 3]), "dataset2": np.array([[1, 2], [3, 4]])}
        >>> write_h5("output.h5", data)
    """
    from dask import array as da

    with h5py.File(path, "w") as h5file:
        for key, value in data.items():
            # Compute dask arrays before h5py takes its global lock; h5py calls in dask's threads would deadlock.
            array = value.compute() if isinstance(value, da.Array) else value
            h5file.create_dataset(key, data=array, compression=compression)
        if attributes:
            h5file.attrs.update(attributes)
    return path

write_npz(path, data)

Write a dictionary to an NPZ file.

Parameters:

• path (str) –

  Path to the NPZ file.

• data (dict) –

  Dictionary containing arrays to write in {'array_name': array} format.

Returns:

• str ( str ) –

  Path to the written NPZ file.

Example

data = {"array1": np.array([1, 2, 3]), "array2": np.array([[1, 2], [3, 4]])} write_npz("output.npz", data)

Source code in src/mesoscopy/io.py

def write_npz(path: str, data: dict) -> str:
    """Write a dictionary to an NPZ file.

    Args:
        path (str): Path to the NPZ file.
        data (dict): Dictionary containing arrays to write in {'array_name': array} format.

    Returns:
        str: Path to the written NPZ file.

    Example:
        >>> data = {"array1": np.array([1, 2, 3]), "array2": np.array([[1, 2], [3, 4]])}
        >>> write_npz("output.npz", data)
    """
    np.savez(path, **data)
    return path

store_interim(array, interim_path, compute=True, chunks=500)

Store an array in an interim Zarr file.

Parameters:

• array (Dask or Numpy Array) –

  Dask or Numpy array to persist on disk.

• interim_path (str) –

  Path to the interim Zarr file. The .zarr extension is added automatically.

• compute (bool, default: True ) –

  Whether to compute the array before storing, applies only to Dask arrays. Defaults to True.

• chunks (int, default: 500 ) –

  Chunk size for the Zarr file. Defaults to 500.

Returns:

• Array –

  Zarr Array: Persistent Zarr array object

Source code in src/mesoscopy/io.py

def store_interim(
    array: da.Array | npt.ArrayLike,
    interim_path: str,
    compute: bool = True,
    chunks: int = 500,
) -> zarr.core.Array:
    """Store an array in an interim Zarr file.

    Args:
        array (Dask or Numpy Array): Dask or Numpy array to persist on disk.
        interim_path (str): Path to the interim Zarr file. The .zarr extension is added automatically.
        compute (bool, optional): Whether to compute the array before storing, applies only to Dask arrays. Defaults to True.
        chunks (int, optional): Chunk size for the Zarr file. Defaults to 500.

    Returns:
        Zarr Array: Persistent Zarr array object
    """
    import zarr
    from dask import array as da

    if not interim_path.endswith(".zarr"):
        interim_path += ".zarr"

    if isinstance(array, da.Array):
        z_interim = zarr.open_array(
            interim_path,
            shape=array.shape,
            dtype=array.dtype,
            chunks=(chunks, array.shape[1], array.shape[2]),
        )
        return array.store(z_interim, return_stored=True, compute=compute)  # type: ignore

    zarr.save(interim_path, array)
    return zarr.load(interim_path)

load_deltaf(path, nwb=False)

Load preprocessed dF/F from an HDF5 or NWB file.

Parameters:

• path (str) –

  Path to the preprocessed file.

• nwb (bool, default: False ) –

  Whether the file is an NWB file. Defaults to False.

Returns:

• tuple[str, ndarray, ndarray] –

  tuple[str, np.ndarray, np.ndarray]: Session identifier, dF/F series, and timestamps.

Source code in src/mesoscopy/io.py

def load_deltaf(path: str, nwb: bool = False) -> tuple[str, np.ndarray, np.ndarray]:
    """Load preprocessed dF/F from an HDF5 or NWB file.

    Args:
        path (str): Path to the preprocessed file.
        nwb (bool, optional): Whether the file is an NWB file. Defaults to False.

    Returns:
        tuple[str, np.ndarray, np.ndarray]: Session identifier, dF/F series, and timestamps.
    """
    if nwb:
        nwbfile = read_nwb(path)
        session_id = nwbfile.identifier
        deltaf_series = np.array(nwbfile.processing["ophys"]["DeltaFSeries"].data)
        timestamps = np.array(nwbfile.processing["ophys"]["DeltaFSeries"].timestamps)
    else:
        session_id = path.split("/")[-1].replace(".h5", "")
        with h5py.File(path, "r") as f:
            deltaf_series = f["/F"][:]
            timestamps = f["/timestamps"][:]

    return session_id, deltaf_series, timestamps

read_timestamps_aligned(path)

Read behaviour-aligned timestamps from an HDF5 file.

Parameters:

• path (str) –

  Path to the HDF5 file.

Returns:

• tuple[ndarray, dict[str, Any]] | None –

  tuple[np.ndarray, dict[str, typing.Any]] | None: The /timestamps_aligned dataset and its attributes

• tuple[ndarray, dict[str, Any]] | None –

  (session_start_time, behaviour_session, offset_s), or None if the file has no such dataset.

Source code in src/mesoscopy/io.py

def read_timestamps_aligned(path: str) -> tuple[np.ndarray, dict[str, typing.Any]] | None:
    """Read behaviour-aligned timestamps from an HDF5 file.

    Args:
        path (str): Path to the HDF5 file.

    Returns:
        tuple[np.ndarray, dict[str, typing.Any]] | None: The `/timestamps_aligned` dataset and its attributes
        (`session_start_time`, `behaviour_session`, `offset_s`), or None if the file has no such dataset.
    """
    if path.endswith(".nwb"):
        return None
    with h5py.File(path, "r") as f:
        if "/timestamps_aligned" not in f:
            return None
        dataset = f["/timestamps_aligned"]
        timestamps = np.asarray(dataset[:], dtype=np.float64)
        attrs = {
            "session_start_time": str(dataset.attrs["session_start_time"]),
            "behaviour_session": str(dataset.attrs["behaviour_session"]),
            "offset_s": float(dataset.attrs["offset_s"]),
        }
    return timestamps, attrs

write_timestamps_aligned(path, timestamps, attrs)

Write behaviour-aligned timestamps to an existing HDF5 file, replacing any existing dataset.

Parameters:

• path (str) –

  Path to the HDF5 file.

• timestamps (ArrayLike) –

  Seconds from behaviour session start, one value per frame.

• attrs (dict[str, Any]) –

  Attributes to set on the dataset (session_start_time, behaviour_session, offset_s).

Returns:

• str ( str ) –

  Path to the HDF5 file.

Source code in src/mesoscopy/io.py

def write_timestamps_aligned(path: str, timestamps: npt.ArrayLike, attrs: dict[str, typing.Any]) -> str:
    """Write behaviour-aligned timestamps to an existing HDF5 file, replacing any existing dataset.

    Args:
        path (str): Path to the HDF5 file.
        timestamps (npt.ArrayLike): Seconds from behaviour session start, one value per frame.
        attrs (dict[str, typing.Any]): Attributes to set on the dataset (`session_start_time`,
            `behaviour_session`, `offset_s`).

    Returns:
        str: Path to the HDF5 file.
    """
    with h5py.File(path, "r+") as f:
        if "/timestamps_aligned" in f:
            del f["/timestamps_aligned"]
        dataset = f.create_dataset("/timestamps_aligned", data=np.asarray(timestamps, dtype=np.float64))
        dataset.attrs.update(attrs)
    return path

read_start_time(path)

Read a recording's start time, as stored in the file.

For HDF5 files this is the first /timestamps entry; for NWB files it is /session_start_time.

Parameters:

• path (str) –

  Path to the HDF5 or NWB file.

Returns:

• str ( str ) –

  The start time.

Raises:

• ValueError –

  If the file has no start time.

Source code in src/mesoscopy/io.py

def read_start_time(path: str) -> str:
    """Read a recording's start time, as stored in the file.

    For HDF5 files this is the first `/timestamps` entry; for NWB files it is `/session_start_time`.

    Args:
        path (str): Path to the HDF5 or NWB file.

    Returns:
        str: The start time.

    Raises:
        ValueError: If the file has no start time.
    """
    key = "/session_start_time" if path.endswith(".nwb") else "/timestamps"
    with h5py.File(path, "r") as f:
        if key not in f or f[key].size == 0:
            msg = f"{path} has no {key} to read a start time from."
            raise ValueError(msg)
        dataset = f[key]
        value = dataset[()] if dataset.ndim == 0 else dataset[0]
    return value.decode("utf-8") if isinstance(value, bytes) else str(value)

read_points(path)

Read a landmark points file.

Coordinates are read as (x, y) tuples, i.e. (column, row).

Parameters:

• path (str) –

  Path to the points file.

Returns:

• dict[str, tuple[float, float]] –

  dict[str, tuple[float, float]]: Dictionary with the landmark names as keys and their (x, y) coordinates

Raises:

• ValueError –

  If the file format is unsupported.

Source code in src/mesoscopy/io.py

def read_points(path: str) -> dict[str, tuple[float, float]]:
    """Read a landmark points file.

    Coordinates are read as (x, y) tuples, i.e. (column, row).

    Args:
        path (str): Path to the points file.

    Returns:
        dict[str, tuple[float, float]]: Dictionary with the landmark names as keys and their (x, y) coordinates

    Raises:
        ValueError: If the file format is unsupported.
    """
    if path.endswith((".xml", ".points")):
        return _read_fiji_points(path)
    if path.endswith(".csv"):
        return _read_csv_points(path)
    msg = "Unsupported file format."
    raise ValueError(msg)

read_regressors(path)

Read a regressor file in NPZ or HDF5 format.

Parameters:

• path (str) –

  Path to the regressor file.

Returns:

• ndarray –

  tuple[np.ndarray, list[str], np.ndarray, dict[str, typing.Any] | None]: Regressor matrix, list of regressor

• list[str] –

  labels, trial indexes, and behaviour alignment (session_start_time, behaviour_session, timestamps) or

• ndarray –

  None if the file has no session_start_time.

Raises:

• ValueError –

  If the file format is unsupported.

Source code in src/mesoscopy/io.py

def read_regressors(path: str) -> tuple[np.ndarray, list[str], np.ndarray, dict[str, typing.Any] | None]:
    """Read a regressor file in NPZ or HDF5 format.

    Args:
        path (str): Path to the regressor file.

    Returns:
        tuple[np.ndarray, list[str], np.ndarray, dict[str, typing.Any] | None]: Regressor matrix, list of regressor
        labels, trial indexes, and behaviour alignment (`session_start_time`, `behaviour_session`, `timestamps`) or
        None if the file has no `session_start_time`.

    Raises:
        ValueError: If the file format is unsupported.
    """
    if path.endswith(".npz"):
        return _read_npz_regressors(path)
    if path.endswith(".h5"):
        return _read_hdf5_regressors(path)
    msg = "Unsupported file format."
    raise ValueError(msg)

read_nuisance_regressors(path)

Read an external nuisance regressor file in NPZ or HDF5 format.

Parameters:

• path (str) –

  Path to the nuisance regressor file.

Returns:

• ndarray –

  tuple[np.ndarray, list[str], np.ndarray]: Nuisance regressor matrix of shape (n_samples, n_regressors),

• list[str] –

  list of regressor labels, and the timestamps (n_samples,) the regressors were recorded at.

Raises:

• ValueError –

  If the file format is unsupported.

Source code in src/mesoscopy/io.py

def read_nuisance_regressors(path: str) -> tuple[np.ndarray, list[str], np.ndarray]:
    """Read an external nuisance regressor file in NPZ or HDF5 format.

    Args:
        path (str): Path to the nuisance regressor file.

    Returns:
        tuple[np.ndarray, list[str], np.ndarray]: Nuisance regressor matrix of shape (n_samples, n_regressors),
        list of regressor labels, and the timestamps (n_samples,) the regressors were recorded at.

    Raises:
        ValueError: If the file format is unsupported.
    """
    if path.endswith(".npz"):
        return _read_npz_nuisance_regressors(path)
    if path.endswith(".h5"):
        return _read_hdf5_nuisance_regressors(path)
    msg = "Unsupported file format."
    raise ValueError(msg)

write_points(path, points)

Write a dictionary of landmark points to a CSV file.

Coordinates are written as (x, y) tuples, i.e. (column, row).

Parameters:

• path (str) –

  Path to output CSV file.

• points (dict[str, tuple[float, float]]) –

  Dictionary with the landmark names as keys and their (x, y) coordinates

Source code in src/mesoscopy/io.py

def write_points(path: str, points: dict[str, tuple[float, float]]) -> None:
    """Write a dictionary of landmark points to a CSV file.

    Coordinates are written as (x, y) tuples, i.e. (column, row).

    Args:
        path (str): Path to output CSV file.
        points (dict[str, tuple[float, float]]): Dictionary with the landmark names as keys and their (x, y) coordinates
    """
    if not path.endswith(".csv"):
        path += ".csv"
    with open(path, "w") as fp:
        csv_writer = csv.DictWriter(fp, fieldnames=["landmark", "x", "y"])
        csv_writer.writeheader()
        for landmark, (x, y) in points.items():
            csv_writer.writerow({"landmark": landmark, "x": x, "y": y})

read_mask(path)

Read one or more custom region masks from an NPY, NPZ or TIFF file.

Masks are 2D arrays matching the frame shape of the recording they are applied to. A boolean or integer mask is read as a single region, named after the NPZ key or, for NPY and TIFF files, the file stem. An integer mask holding distinct non-zero values is read as one region per label, named <name>_<label>.

Parameters:

• path (str) –

  Path to the mask file. NPZ files may hold several masks, one per key.

Returns:

• dict[str, NDArray[bool_]] –

  dict[str, npt.NDArray[np.bool_]]: Dictionary with the region names as keys and their boolean masks as values.

Raises:

• ValueError –

  If the file format is unsupported, a mask is not 2D, a non-integer mask holds values other than 0 and 1, or two masks in the file resolve to the same name.

Example

read_mask("barrel_cortex.npy") {'barrel_cortex': array([[False, ...]])}

Source code in src/mesoscopy/io.py

def read_mask(path: str) -> dict[str, npt.NDArray[np.bool_]]:
    """Read one or more custom region masks from an NPY, NPZ or TIFF file.

    Masks are 2D arrays matching the frame shape of the recording they are applied to. A boolean or integer mask is read
    as a single region, named after the NPZ key or, for NPY and TIFF files, the file stem. An integer mask holding
    distinct non-zero values is read as one region per label, named `<name>_<label>`.

    Args:
        path (str): Path to the mask file. NPZ files may hold several masks, one per key.

    Returns:
        dict[str, npt.NDArray[np.bool_]]: Dictionary with the region names as keys and their boolean masks as values.

    Raises:
        ValueError: If the file format is unsupported, a mask is not 2D, a non-integer mask holds values other than
            0 and 1, or two masks in the file resolve to the same name.

    Example:
        >>> read_mask("barrel_cortex.npy")
        {'barrel_cortex': array([[False, ...]])}
    """
    if path.endswith(".npy"):
        arrays = {Path(path).stem: np.load(path)}
    elif path.endswith(".npz"):
        with np.load(path) as f:
            arrays = {name: f[name] for name in f.files}
    elif path.endswith((".tif", ".tiff")):
        arrays = {Path(path).stem: np.asarray(iio.imread(path))}
    else:
        msg = "Unsupported file format."
        raise ValueError(msg)

    masks: dict[str, npt.NDArray[np.bool_]] = {}
    for name, array in arrays.items():
        for mask_name, mask in _split_mask_labels(array, name).items():
            if mask_name in masks:
                msg = f"Mask name {mask_name} occurs more than once in {path}."
                raise ValueError(msg)
            masks[mask_name] = mask

    return masks