Skip to content

Data I/O Module

Data I/O functions for qi2lab 3D MERFISH.

This module provides utilities for reading and writing data in various formats used by qi2lab 3D MERFISH datasets.

History:
  • 2024/12: Refactored repo structure.
  • 2024/12: Updated docstrings.
  • 2024/07: Removed native NDTiff reading package; integrated tifffile/zarr. Reduced dask dependencies.

Functions:

Name Description
read_config_file

Read config data from csv file.

read_metadatafile

Read metadata from csv file.

resolve_datastore_path

Find an existing datastore from its root or its experiment directory.

return_data_zarr

Return NDTIFF data as a numpy array via tiffile.

time_stamp

Generate timestamp string.

write_metadata

Write dictionary as CSV file.

write_sparse_mtx

Write sparse matrix in MTX format.

write_tsv

Write data to TSV file.

read_config_file(config_path)

Read config data from csv file.

Parameters:

Name Type Description Default
config_path Path | str

Location of configuration file

required

Returns:

Name Type Description
dict_from_csv dict

instrument configuration metadata

Source code in src/merfish3danalysis/utils/dataio.py
def read_config_file(config_path: Path | str) -> dict:
    """Read config data from csv file.

    Parameters
    ----------
    config_path: Path
        Location of configuration file

    Returns
    -------
    dict_from_csv: dict
        instrument configuration metadata
    """
    dict_from_csv = (
        pd.read_csv(config_path, header=None, index_col=0).squeeze("columns").to_dict()
    )

    return dict_from_csv

read_metadatafile(fname)

Read metadata from csv file.

Parameters:

Name Type Description Default
fname str | Path

filename

required

Returns:

Name Type Description
metadata Dict

metadata dictionary

Source code in src/merfish3danalysis/utils/dataio.py
def read_metadatafile(fname: str | Path) -> dict:
    """Read metadata from csv file.

    Parameters
    ----------
    fname: str or Path
        filename

    Returns
    -------
    metadata: Dict
        metadata dictionary
    """
    scan_data_raw_lines = []

    with open(fname) as f:
        for line in f:
            scan_data_raw_lines.append(line.replace("\n", ""))

    titles = scan_data_raw_lines[0].split(",")

    # convert values to appropriate datatypes
    vals = scan_data_raw_lines[1].split(",")
    for ii in range(len(vals)):
        if re.fullmatch(r"\d+", vals[ii]):
            vals[ii] = int(vals[ii])
        elif re.fullmatch(r"\d*.\d+", vals[ii]):
            vals[ii] = float(vals[ii])
        elif vals[ii].lower() == "False".lower():
            vals[ii] = False
        elif vals[ii].lower() == "True".lower():
            vals[ii] = True
        else:
            # otherwise, leave as string
            pass

    # convert to dictionary
    metadata = {}
    for t, v in zip(titles, vals, strict=False):
        metadata[t] = v

    return metadata

resolve_datastore_path(path)

Find an existing datastore from its root or its experiment directory.

Parameters:

Name Type Description Default
path Path or str

Datastore directory or experiment directory containing qi2labdatastore.

required

Returns:

Type Description
Path

Absolute datastore directory identified by datastore_state.json.

Raises:

Type Description
FileNotFoundError

Neither the supplied directory nor its qi2labdatastore child has state metadata. This function never creates directories.

Source code in src/merfish3danalysis/utils/dataio.py
def resolve_datastore_path(path: Path | str) -> Path:
    """Find an existing datastore from its root or its experiment directory.

    Parameters
    ----------
    path : Path or str
        Datastore directory or experiment directory containing qi2labdatastore.

    Returns
    -------
    Path
        Absolute datastore directory identified by datastore_state.json.

    Raises
    ------
    FileNotFoundError
        Neither the supplied directory nor its qi2labdatastore child has state
        metadata. This function never creates directories.
    """
    path = Path(path).expanduser().resolve()
    for candidate in (path, path / "qi2labdatastore"):
        if (candidate / "datastore_state.json").is_file():
            return candidate
    raise FileNotFoundError(
        f"No qi2lab datastore found at {path} or its qi2labdatastore child."
    )

return_data_zarr(dataset_path, ch_idx, ch_idx_offset=0)

Return NDTIFF data as a numpy array via tiffile.

Parameters:

Name Type Description Default
dataset_path Path | str

pycromanager dataset object

required
ch_idx int

channel index in ZarrTiffStore file

required
ch_idx_offset int | None

channel index offset for unused phase channels

0

Returns:

Name Type Description
data ArrayLike

data stack

Source code in src/merfish3danalysis/utils/dataio.py
def return_data_zarr(
    dataset_path: Path | str, ch_idx: int, ch_idx_offset: int | None = 0
) -> ArrayLike:
    """Return NDTIFF data as a numpy array via tiffile.

    Parameters
    ----------
    dataset_path: Dataset
        pycromanager dataset object
    ch_idx: int
        channel index in ZarrTiffStore file
    ch_idx_offset: int
        channel index offset for unused phase channels

    Returns
    -------
    data: ArrayLike
        data stack
    """
    ndtiff_zarr_store = imread(dataset_path, mode="r+", aszarr=True)
    ndtiff_zarr = zarr.open(ndtiff_zarr_store, mode="r+")
    first_dim = str(ndtiff_zarr.attrs["_ARRAY_DIMENSIONS"][0])

    if first_dim == "C":
        data = np.asarray(ndtiff_zarr[ch_idx - ch_idx_offset, :], dtype=np.uint16)
    else:
        data = np.asarray(ndtiff_zarr[:, ch_idx - ch_idx_offset, :], dtype=np.uint16)
    del ndtiff_zarr_store, ndtiff_zarr

    return np.squeeze(data)

time_stamp()

Generate timestamp string.

Returns:

Name Type Description
timestamp str

timestamp formatted as string

Source code in src/merfish3danalysis/utils/dataio.py
def time_stamp() -> str:
    """Generate timestamp string.

    Returns
    -------
    timestamp: str
        timestamp formatted as string
    """
    return datetime.now().strftime("%Y-%m-%d %H:%M:%S")

write_metadata(data_dict, save_path)

Write dictionary as CSV file.

Parameters:

Name Type Description Default
data_dict dict

metadata dictionary

required
save_path str | Path

path for file

required
Source code in src/merfish3danalysis/utils/dataio.py
def write_metadata(data_dict: dict, save_path: str | Path) -> None:
    """Write dictionary as CSV file.

    Parameters
    ----------
    data_dict: dict
        metadata dictionary
    save_path: str or Path
        path for file
    """
    pd.DataFrame([data_dict]).to_csv(save_path)

write_sparse_mtx(output_dir_path, matrix, cells, features)

Write sparse matrix in MTX format.

Parameters:

Name Type Description Default
output_dir_path Path | str

Path to output directory

required
matrix ArrayLike

Sparse matrix

required
cells Sequence[str]

Cell names

required
features Sequence[str]

Feature names

required
Source code in src/merfish3danalysis/utils/dataio.py
def write_sparse_mtx(
    output_dir_path: Path | str,
    matrix: ArrayLike,
    cells: Sequence[str],
    features: Sequence[str],
) -> None:
    """Write sparse matrix in MTX format.

    Parameters
    ----------
    output_dir_path: Path or str
        Path to output directory
    matrix: ArrayLike
        Sparse matrix
    cells: Sequence[str]
        Cell names
    features: Sequence[str]
        Feature names
    """
    sparse_mat = sparse.coo_matrix(matrix.values)
    sio.mmwrite(str(output_dir_path / "matrix.mtx"), sparse_mat)
    write_tsv(output_dir_path / "barcodes.tsv", ["cell_" + str(cell) for cell in cells])
    write_tsv(
        output_dir_path / "features.tsv",
        [
            [
                str(f),
                str(f),
                "Blank Codeword" if str(f).startswith("Blank") else "Gene Expression",
            ]
            for f in features
        ],
    )
    subprocess.run(f"gzip -f {output_dir_path!s}/*", shell=True)

write_tsv(filename, data)

Write data to TSV file.

Parameters:

Name Type Description Default
filename str | Path

Filename

required
data Sequence[str | Sequence[str]]

Data to write

required
Source code in src/merfish3danalysis/utils/dataio.py
def write_tsv(filename: str | Path, data: Sequence[str | Sequence[str]]) -> None:
    """Write data to TSV file.

    Parameters
    ----------
    filename: str or Path
        Filename
    data: Sequence[str or Sequence[str]]
        Data to write
    """
    with open(filename, "w", newline="") as tsvfile:
        writer = csv.writer(tsvfile, delimiter="\t", lineterminator="\n")
        for item in data:
            writer.writerow([item] if isinstance(item, str) else item)