[FEAT] Plain nifti <-> zarr converter

.github/workflows/ci.yml

@@ -11,7 +11,7 @@ jobs:
         # glymur needs dynamic library on macos and windows, and causing error
         # os: [ubuntu-22.04, macos-latest, windows-latest]
         os: [ubuntu-22.04]
-        
+
     runs-on: ${{ matrix.os }}
     steps:
       - uses: actions/checkout@v4
@@ -34,7 +34,7 @@ jobs:
           path: ./.venv
           key: venv-${{ hashFiles('poetry.lock') }}
       - name: Install the project dependencies
-        run: poetry install --with dev,lsm,df,wk
+        run: poetry install --with dev --extras all
       - name: Run the automated tests
         run: poetry run pytest -v
         working-directory: ./tests

linc_convert/__init__.py

@@ -1,4 +1,4 @@
 """Data conversion tools for the LINC project."""
 
-__all__ = ["modalities", "utils"]
-from . import modalities, utils
+__all__ = ["modalities", "plain", "utils"]
+from . import modalities, plain, utils

linc_convert/cli.py

@@ -1,6 +1,10 @@
 """Root command line entry point."""
 
-from cyclopts import App
+from cyclopts import App, Group
 
 help = "Collection of conversion scripts for LINC datasets"
 main = App("linc-convert", help=help)
+
+
+modalities_group = Group.create_ordered("Modality-specific converters")
+plain_group = Group.create_ordered("Plain converters")

linc_convert/modalities/df/cli.py

@@ -2,8 +2,8 @@
 
 from cyclopts import App
 
-from linc_convert.cli import main
+from linc_convert.cli import main, modalities_group
 
 help = "Converters for Dark Field microscopy"
-df = App(name="df", help=help)
+df = App(name="df", help=help, group=modalities_group)
 main.command(df)

linc_convert/modalities/lsm/cli.py

@@ -2,8 +2,8 @@
 
 from cyclopts import App
 
-from linc_convert.cli import main
+from linc_convert.cli import main, modalities_group
 
 help = "Converters for Light Sheet Microscopy"
-lsm = App(name="lsm", help=help)
+lsm = App(name="lsm", help=help, group=modalities_group)
 main.command(lsm)

linc_convert/modalities/lsm/mosaic.py

@@ -10,6 +10,7 @@
 import os
 import re
 from glob import glob
+from typing import Annotated
 
 # externals
 import cyclopts
@@ -41,7 +42,7 @@ def convert(
     center: bool = True,
     thickness: float | None = None,
     voxel_size: list[float] = (1, 1, 1),
-    **kwargs
+    **kwargs: Annotated[dict, cyclopts.Parameter(show=False)],
 ) -> None:
     """
     Convert a collection of tiff files generated by the LSM pipeline into ZARR.

linc_convert/modalities/psoct/cli.py

@@ -2,8 +2,8 @@
 
 from cyclopts import App
 
-from linc_convert.cli import main
+from linc_convert.cli import main, modalities_group
 
 help = "Converters for PS-OCT .mat files"
-psoct = App(name="psoct", help=help)
+psoct = App(name="psoct", help=help, group=modalities_group)
 main.command(psoct)

linc_convert/modalities/wk/__init__.py

@@ -1,4 +1,10 @@
 """Webknossos annotation converters."""
 
-__all__ = ["cli", "webknossos_annotation"]
-from . import cli, webknossos_annotation
+
+try:
+    import wkw as _  # noqa: F401
+
+    __all__ = ["cli", "webknossos_annotation"]
+    from . import cli, webknossos_annotation
+except ImportError:
+    pass

linc_convert/modalities/wk/cli.py

@@ -2,8 +2,8 @@
 
 from cyclopts import App
 
-from linc_convert.cli import main
+from linc_convert.cli import main, modalities_group
 
 help = "Converters for Webknossos annotation"
-wk = App(name="wk", help=help)
+wk = App(name="wk", help=help, group=modalities_group)
 main.command(wk)

linc_convert/plain/__init__.py

@@ -0,0 +1,15 @@
+"""Plain format converters."""
+
+__all__ = []
+
+try:
+    from . import cli
+    __all__ += ["cli"]
+except ImportError:
+    pass
+
+try:
+    from . import nii2zarr, zarr2nii
+    __all__ += ["nii2zarr", "zarr2nii"]
+except ImportError:
+    pass

linc_convert/plain/cli.py

@@ -0,0 +1,119 @@
+"""Entry-points for Dark Field microscopy converter."""
+# stdlib
+import os.path as op
+import warnings
+from enum import EnumType, StrEnum
+from enum import _EnumDict as EnumDict
+from typing import Annotated
+
+# externals
+from cyclopts import Parameter
+
+# internals
+from linc_convert.cli import main, plain_group
+from linc_convert.plain.register import (
+    format_to_extension,
+    known_converters,
+    known_extensions,
+)
+from linc_convert.utils.opener import stringify_path
+
+known_inp_formats = []
+known_out_formats = []
+for src, dst in known_converters:
+    known_inp_formats.append(src)
+    known_out_formats.append(dst)
+known_inp_formats = list(set(known_inp_formats))
+known_out_formats = list(set(known_out_formats))
+
+_inp_dict = EnumDict()
+_inp_dict.update({x for x in known_inp_formats})
+inp_hints = EnumType("inp_hints", (StrEnum,), _inp_dict)
+
+_out_dict = EnumDict()
+_out_dict.update({x for x in known_inp_formats})
+out_hints = EnumType("out_hints", (StrEnum,), _out_dict)
+
+
+@main.command(name="any", group=plain_group)
+def convert(
+    inp: str,
+    out: str | None = None,
+    *,
+    inp_hint: inp_hints | None = None,
+    out_hint: out_hints | None = None,
+    **kwargs: Annotated[dict, Parameter(show=False)],
+) -> None:
+    """
+    Convert between formats while preserving (meta)data as much as possible.
+
+    Parameters
+    ----------
+    inp
+        Path to input file.
+    out
+        Path to output file.
+        If output path not provided, a hint MUST be provided.
+    inp_hint
+        Input format. Default: guess from extension.
+    out_hint
+        Output format. Default: guess from extension.
+    """
+    inp = stringify_path(inp)
+    out = stringify_path(out)
+    inp_hint = inp_hint or []
+    out_hint = out_hint or []
+    if not isinstance(inp_hint, list):
+        inp_hint = [inp_hint]
+    if not isinstance(out_hint, list):
+        out_hint = [out_hint]
+
+    # Find type hints from extensions
+    for ext, format in known_extensions.items():
+        for compression in ('', '.gz', '.bz2'):
+            if inp.endswith(ext + compression):
+                inp_hint += [format]
+            if out and out.endswith(ext + compression):
+                out_hint += [format]
+
+    if not inp_hint:
+        raise ValueError("Could not guess inoput format from extension")
+    if not out_hint:
+        raise ValueError("Could not guess output format from extension")
+
+    # Default output name
+    if not out:
+        out = make_output_path(inp, out_hint[0])
+
+    # Try converter(s)
+    for inp_format in inp_hint:
+        for out_format in out_hint:
+            if (inp_format, out_format) not in known_converters:
+                continue
+            converter = known_converters[(inp_format, out_format)]
+            try:
+                return converter(inp, out, **kwargs)
+            except Exception:
+                warnings.warn(
+                    f"Failed to convert from {inp_format} to {out_format}."
+                )
+
+    raise RuntimeError("All converters failed.")
+
+
+def make_output_path(inp: str, format: str) -> str:
+    """Create an output path if not provided."""
+    base, ext = op.splitext(inp)
+    if ext in ('.gz', '.bz2'):
+        base, ext = op.splitext(base)
+    if ext in ('.zarr', '.tiff') and base.endswith('.ome', '.nii'):
+        base, ext = op.splitext(base)
+
+    if "://" in base:
+        if base.startswith("file://"):
+            base = base[7:]
+        else:
+            # Input file is remote, make output file local.
+            base = op.basename(base)
+
+    return base + format_to_extension[format][0]

linc_convert/plain/nii2zarr.py

@@ -0,0 +1,126 @@
+"""Convert from NIfTI to Zarr."""
+import dataclasses
+from typing import Annotated, Literal
+
+from cyclopts import App, Parameter
+from niizarr import nii2zarr
+
+from linc_convert.cli import main, plain_group
+from linc_convert.plain.cli import make_output_path
+from linc_convert.plain.register import register_converter
+from linc_convert.utils.opener import filesystem, open
+
+app = App(name="nii2zarr", help_format="markdown", group=plain_group)
+main.command(app)
+
+
+@dataclasses.dataclass
+class _Nii2ZarrConfig:
+    """
+    Configuration specific to NIfTI-to-Zarr conversion.
+
+    Parameters
+    ----------
+    chunk
+        Chunk size for spatial dimensions.
+        The tuple allows different chunk sizes to be used along each dimension.
+    chunk_channel
+        Chunk size of the channel dimension. If 0, combine all channels
+        in a single chunk.
+    chunk_time
+        Chunk size for the time dimension. If 0, combine all timepoints
+        in a single chunk.
+    shard
+        Shard size for spatial dimensions.
+        The tuple allows different shard sizes to be used along each dimension.
+    shard_channel
+        Shard size of the channel dimension. If 0, combine all channels
+        in a single shard.
+    shard_time
+        Shard size for the time dimension. If 0, combine all timepoints
+        in a single shard.
+    nb_levels
+        Number of pyramid levels to generate.
+        If -1, make all possible levels until the level can be fit into
+        one chunk.
+    method
+        Method used to compute the pyramid.
+    label
+        Is this is a label volume?  If `None`, guess from intent code.
+    no_time
+        If True, there is no time dimension so the 4th dimension
+        (if it exists) should be interpreted as the channel dimensions.
+    no_pyramid_axis
+        Axis that should not be downsampled. If None, downsample
+        across all three dimensions.
+    fill_value
+        Value to use for missing tiles
+    compressor
+        Compression to use
+    compressor_options
+        Options for the compressor.
+    zarr_version
+        Zarr format version.
+    ome_version
+        OME-Zarr version.
+    """
+
+    chunk: tuple[int] = 64
+    chunk_channel: int = 1
+    chunk_time: int = 1
+    shard: tuple[int] | None = None
+    shard_channel: int | None = None
+    shard_time: int | None = None
+    nb_levels: int = -1
+    method: Literal['gaussian', 'laplacian'] = 'gaussian'
+    label: bool | None = None
+    no_time: bool = False
+    no_pyramid_axis: str | int = None
+    fill_value: int | float | complex | None = None
+    compressor: Literal['blosc', 'zlib'] = 'blosc'
+    compressor_options: dict = dataclasses.field(default_factory=(lambda: {}))
+    zarr_version: Literal[2, 3] = 2
+    ome_version: Literal["0.4", "0.5"] = "0.4"
+
+
+Nii2ZarrConfig = Annotated[_Nii2ZarrConfig, Parameter(name="*")]
+
+
+@app.default
+@register_converter('zarr', 'nifti')
+@register_converter('omezarr', 'nifti')
+@register_converter('niftizarr', 'nifti')
+def convert(
+    inp: str,
+    out: str | None,
+    *,
+    config: Nii2ZarrConfig | None = None,
+    **kwargs: Annotated[dict, Parameter(show=False)],
+) -> None:
+    """
+    Convert from NIfTI to Zarr.
+
+    Parameters
+    ----------
+    inp
+        Path to input file.
+    out
+        Path to output file. Default: "{base}.nii.zarr".
+
+    """
+    config = config or Nii2ZarrConfig()
+    dataclasses.replace(config, **kwargs)
+
+    # Default output name
+    if not out:
+        out = make_output_path(inp, "niftizarr")
+
+    if inp.startswith("dandi://"):
+        # Create an authentified fsspec store to pass to
+        import linc_convert.utils.dandifs  # noqa: F401  (register filesystem)
+        url = filesystem(inp).s3_url(inp)
+        fs = filesystem(url)
+        with open(fs.open(url)) as stream:
+            return nii2zarr(stream, out, **dataclasses.asdict(config))
+
+    return nii2zarr(inp, out, **dataclasses.asdict(config))