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).
    """
    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.
    """
    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)
    """
    with h5py.File(path, "w") as h5file:
        for key, value in data.items():
            h5file.create_dataset(key, data=value, 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
    """
    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_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:

• tuple[ndarray, list[str], ndarray] –

  tuple[np.ndarray, list[str], np.ndarray]: Regressor matrix, list of regressor labels, and trial indexes.

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]:
    """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]: Regressor matrix, list of regressor labels, and trial indexes.

    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})