Add layer_name and mag as args for compress command. (#1141)

* Add layer_name and mag as args for compress command.

* Update changelog.

Author: markbader

webknossos/Changelog.md

@@ -17,6 +17,7 @@ For upgrade instructions, please check the respective _Breaking Changes_ section
 ### Added
 
 ### Changed
+- Added options `--layer-name` and `--mag` for compress command of the CLI. [#1141](https://github.com/scalableminds/webknossos-libs/pull/1141)
 
 ### Fixed
 

webknossos/tests/test_cli.py

@@ -176,6 +176,42 @@ def test_compress() -> None:
         assert result.exit_code == 0
 
 
+def test_compress_with_args() -> None:
+    """Tests the functionality of compress subcommand."""
+
+    with tmp_cwd():
+        wkw_path = TESTDATA_DIR / "simple_wkw_dataset"
+        copytree(wkw_path, Path("testdata") / "simple_wkw_dataset")
+
+        result = runner.invoke(
+            app,
+            [
+                "compress",
+                "--layer-name",
+                "color",
+                "--mag",
+                "1",
+                "testdata/simple_wkw_dataset",
+            ],
+        )
+
+        assert result.exit_code == 0
+
+        result_with_wrong_mag = runner.invoke(
+            app,
+            [
+                "compress",
+                "--layer-name",
+                "color",
+                "--mag",
+                "2",
+                "testdata/simple_wkw_dataset",
+            ],
+        )
+
+        assert result_with_wrong_mag.exit_code == 1
+
+
 @pytest.mark.filterwarnings("ignore::UserWarning")
 def test_convert() -> None:
     """Tests the functionality of convert subcommand."""

webknossos/webknossos/cli/compress.py

@@ -2,14 +2,16 @@
 
 from argparse import Namespace
 from multiprocessing import cpu_count
-from typing import Any, Optional
+from typing import Any, List, Optional
 
 import typer
 from typing_extensions import Annotated
 
+from webknossos.geometry.mag import Mag
+
 from ..dataset import Dataset
 from ..utils import get_executor_for_args
-from ._utils import DistributionStrategy, parse_path
+from ._utils import DistributionStrategy, parse_mag, parse_path
 
 
 def main(
@@ -22,6 +24,22 @@ def main(
             parser=parse_path,
         ),
     ],
+    layer_name: Annotated[
+        Optional[str],
+        typer.Option(
+            help="Name of the layer to be compressed. If not provided, all layers will be compressed.",
+        ),
+    ] = None,
+    mag: Annotated[
+        Optional[List[Mag]],
+        typer.Option(
+            help="Mags that should be compressed. "
+            "Should be number or minus separated string (e.g. 2 or 2-2-2). "
+            "For multiple mags type: --mag 1 --mag 2",
+            parser=parse_mag,
+            metavar="MAG",
+        ),
+    ] = None,
     jobs: Annotated[
         int,
         typer.Option(
@@ -53,5 +71,20 @@ def main(
         job_resources=job_resources,
     )
 
+    ds = Dataset.open(target)
+    if layer_name is None:
+        layers = list(ds.layers.values())
+    else:
+        layers = [ds.get_layer(layer_name)]
+
     with get_executor_for_args(args=executor_args) as executor:
-        Dataset.open(target).compress(executor=executor)
+        for layer in layers:
+            if mag is None:
+                mags = list(layer.mags.values())
+            else:
+                mags = [layer.get_mag(mag) for mag in mag]
+            for current_mag in mags:
+                if not current_mag._is_compressed():
+                    current_mag.compress(executor=executor)
+                else:
+                    typer.echo(f"Skipping {current_mag} as it is already compressed.")