Support pathlib Path in api calls (#293)

* use Path instead of str for paths

* cleanup path support

* fix a few type error that were not found by typecheking

* reformat code

* fix more type errors

* restructure paths in test

* set type of argument parser to 'Path'

* fix paths in tests

* also allow 'Path' for KnossosDataset

* fix type in test case

tests/test_downsampling.py

@@ -1,4 +1,5 @@
 import logging
+from pathlib import Path
 from typing import Tuple, cast
 
 import numpy as np
@@ -22,6 +23,8 @@
 WKW_CUBE_SIZE = 1024
 CUBE_EDGE_LEN = 256
 
+TESTOUTPUT_DIR = Path("testoutput")
+
 
 def read_wkw(
     wkw_info: WkwDatasetInfo, offset: Tuple[int, int, int], size: Tuple[int, int, int]
@@ -79,8 +82,8 @@ def test_non_linear_filter_reshape() -> None:
 
 
 def downsample_test_helper(use_compress: bool) -> None:
-    source_path = "testdata/WT1_wkw"
-    target_path = "testoutput/WT1_wkw"
+    source_path = Path("testdata", "WT1_wkw")
+    target_path = TESTOUTPUT_DIR / "WT1_wkw"
 
     try:
         shutil.rmtree(target_path)
@@ -153,11 +156,11 @@ def test_downsample_multi_channel() -> None:
     file_len = 32
 
     try:
-        shutil.rmtree("testoutput/multi-channel-test")
+        shutil.rmtree(TESTOUTPUT_DIR / "multi-channel-test")
     except:
         pass
 
-    ds = WKDataset.create("testoutput/multi-channel-test", (1, 1, 1))
+    ds = WKDataset.create(TESTOUTPUT_DIR / "multi-channel-test", (1, 1, 1))
     l = ds.add_layer(
         "color", Layer.COLOR_TYPE, dtype_per_channel="uint8", num_channels=num_channels
     )
@@ -249,7 +252,7 @@ def test_downsampling_padding() -> None:
         ),
     ]
     for args in padding_tests:
-        ds_path = "./testoutput/larger_wk_dataset/"
+        ds_path = TESTOUTPUT_DIR / "larger_wk_dataset"
         try:
             shutil.rmtree(ds_path)
         except:

tests/test_image_readers.py

@@ -1,11 +1,13 @@
+from pathlib import Path
+
 from wkcuber.image_readers import ImageReaderManager
 import numpy as np
 
 
 def test_rgb_tiff_case() -> None:
     image_reader_manager = ImageReaderManager()
     result = image_reader_manager.read_array(
-        "./testdata/rgb_tiff/test_rgb.tif", np.uint8, 0
+        Path("./testdata/rgb_tiff/test_rgb.tif"), np.uint8, 0
     )
     assert result.shape == (32, 32, 3, 1)
     assert np.all(result[0][0] == [[0], [255], [0]])

tests/test_mag.py

@@ -1,10 +1,12 @@
+from pathlib import Path
+
 import numpy as np
 from wkcuber.mag import Mag
 from wkcuber.metadata import detect_resolutions
 
 
 def test_detect_resolutions() -> None:
-    resolutions = sorted(list(detect_resolutions("testdata/WT1_wkw", "color")))
+    resolutions = sorted(list(detect_resolutions(Path("testdata", "WT1_wkw"), "color")))
     assert [mag.to_layer_name() for mag in resolutions] == ["1", "2-2-1"]
 
 

tests/test_metadata.py

@@ -1,5 +1,6 @@
+from pathlib import Path
+
 import numpy as np
-import os
 import wkw
 
 from wkcuber.cubing import ensure_wkw
@@ -14,7 +15,7 @@
 
 
 def test_element_class_conversion() -> None:
-    test_wkw_path = os.path.join("testoutput", "test_metadata")
+    test_wkw_path = Path("testoutput", "test_metadata")
     prediction_layer_name = "prediction"
     prediction_wkw_info = WkwDatasetInfo(
         test_wkw_path, prediction_layer_name, 1, wkw.Header(np.float32, num_channels=3)
@@ -40,7 +41,7 @@ def test_element_class_conversion() -> None:
 
 
 def check_element_class_of_layer(
-    test_wkw_path: str,
+    test_wkw_path: Path,
     layer_name: str,
     expected_element_class: str,
     expected_dtype: type,
@@ -60,7 +61,7 @@ def check_element_class_of_layer(
 
 
 def write_custom_layer(
-    target_path: str, layer_name: str, dtype: type, num_channels: int
+    target_path: Path, layer_name: str, dtype: type, num_channels: int
 ) -> None:
     data = (
         np.arange(4 * 4 * 4 * num_channels)
@@ -84,7 +85,7 @@ def test_mapping_detection() -> None:
         "test_mapping_4.json",
         "test_mapping_5.json",
     ]
-    datapath_with_mappings = "testdata/test_metadata"
+    datapath_with_mappings = Path("testdata", "test_metadata")
     layer_name_with_mapping = "segmentation"
     detected_mappings = detect_mappings(datapath_with_mappings, layer_name_with_mapping)
 
@@ -97,7 +98,7 @@ def test_mapping_detection() -> None:
         detected_mappings
     ), "Did not find all mappings."
 
-    datapath_without_mappings = "testdata/WT1_wkw"
+    datapath_without_mappings = Path("testdata", "WT1_wkw")
     layer_name_without_mapping = "color"
     detected_mappings = detect_mappings(
         datapath_without_mappings, layer_name_without_mapping

tests/test_upsampling.py

@@ -1,5 +1,6 @@
 import shutil
 import tempfile
+from pathlib import Path
 from typing import Tuple, cast
 
 from wkcuber.api.Dataset import WKDataset
@@ -15,7 +16,7 @@
 
 def test_upsampling() -> None:
     with tempfile.TemporaryDirectory() as temp_dir:
-        ds = WKDataset.create(temp_dir, scale=(1, 1, 1))
+        ds = WKDataset.create(Path(temp_dir), scale=(1, 1, 1))
         layer = ds.add_layer("color", "COLOR")
         mag = layer.add_mag([4, 4, 2])
         mag.write(
@@ -48,7 +49,7 @@ def test_upsample_cube() -> None:
 
 def upsample_test_helper(use_compress: bool) -> None:
     with tempfile.TemporaryDirectory() as temp_dir:
-        ds = WKDataset.create(temp_dir, scale=(10.5, 10.5, 5))
+        ds = WKDataset.create(Path(temp_dir), scale=(10.5, 10.5, 5))
         layer = ds.add_layer("color", "COLOR")
         mag2 = layer.add_mag([2, 2, 2])
 
@@ -112,11 +113,11 @@ def test_upsample_multi_channel() -> None:
     file_len = 32
 
     try:
-        shutil.rmtree("testoutput/multi-channel-test")
+        shutil.rmtree(Path("testoutput", "multi-channel-test"))
     except:
         pass
 
-    ds = WKDataset.create("testoutput/multi-channel-test", (1, 1, 1))
+    ds = WKDataset.create(Path("testoutput", "multi-channel-test"), (1, 1, 1))
     l = ds.add_layer(
         "color", Layer.COLOR_TYPE, dtype_per_channel="uint8", num_channels=num_channels
     )

tests/test_utils.py

@@ -1,3 +1,6 @@
+from pathlib import Path
+from typing import Union
+
 import numpy as np
 from wkcuber.utils import get_chunks, get_regular_chunks, BufferedSliceWriter
 import wkw
@@ -8,7 +11,7 @@
 BLOCK_LEN = 32
 
 
-def delete_dir(relative_path: str) -> None:
+def delete_dir(relative_path: Union[str, Path]) -> None:
     if os.path.exists(relative_path) and os.path.isdir(relative_path):
         rmtree(relative_path)
 
@@ -43,17 +46,17 @@ def test_buffered_slice_writer() -> None:
     dtype = test_img.dtype
     bbox = {"topleft": (0, 0, 0), "size": (24, 24, 35)}
     origin = [0, 0, 0]
-    dataset_dir = "testoutput/buffered_slice_writer"
+    dataset_dir = Path("testoutput", "buffered_slice_writer")
     layer_name = "color"
     mag = Mag(1)
-    dataset_path = os.path.join(dataset_dir, layer_name, mag.to_layer_name())
+    dataset_path = dataset_dir / layer_name / mag.to_layer_name()
 
     delete_dir(dataset_dir)
 
     with BufferedSliceWriter(dataset_dir, layer_name, dtype, origin, mag=mag) as writer:
         for i in range(13):
             writer.write_slice(i, test_img)
-        with wkw.Dataset.open(dataset_path, wkw.Header(dtype)) as data:
+        with wkw.Dataset.open(str(dataset_path), wkw.Header(dtype)) as data:
             try:
                 read_data = data.read(origin, (24, 24, 13))
                 if read_data[read_data.nonzero()].size != 0:
@@ -67,7 +70,7 @@ def test_buffered_slice_writer() -> None:
 
         for i in range(13, 32):
             writer.write_slice(i, test_img)
-        with wkw.Dataset.open(dataset_path, wkw.Header(dtype)) as data:
+        with wkw.Dataset.open(str(dataset_path), wkw.Header(dtype)) as data:
             read_data = data.read(origin, (24, 24, 32))
             assert np.squeeze(read_data).shape == (24, 24, 32), (
                 "The read data should have the shape: (24, 24, 32) "
@@ -80,7 +83,7 @@ def test_buffered_slice_writer() -> None:
         for i in range(32, 35):
             writer.write_slice(i, test_img)
 
-    with wkw.Dataset.open(dataset_path, wkw.Header(dtype)) as data:
+    with wkw.Dataset.open(str(dataset_path), wkw.Header(dtype)) as data:
         read_data = data.read(origin, (24, 24, 35))
         read_data = np.squeeze(read_data)
         assert read_data.shape == (24, 24, 35), (

wkcuber/__main__.py

@@ -13,10 +13,9 @@
 from pathlib import Path
 
 
-def detect_present_mags(target_path: str) -> Dict[Path, List[Mag]]:
+def detect_present_mags(target_path: Path) -> Dict[Path, List[Mag]]:
     layer_path_to_mags: Dict[Path, List[Mag]] = dict()
-    dataset_path = Path(target_path)
-    layer_paths = list([p for p in dataset_path.iterdir() if p.is_dir()])
+    layer_paths = list([p for p in target_path.iterdir() if p.is_dir()])
     for layer_p in layer_paths:
         layer_path_to_mags.setdefault(layer_p, list())
         mag_paths = list([p for p in layer_p.iterdir() if p.is_dir()])