fixed float/double to float32, float64 conversion (#126)

* fixed bug and implemented test

* integrated feedback

* formatted code

Author: MartinBuessemeyer

tests/scripts/simple_anisotropic_tiff_cubing.sh

@@ -10,6 +10,7 @@ docker run \
   --batch_size 8 \
   --layer_name color \
   --max 8 \
+  --scale "11.24,11.24,25"\
   --name awesome_data \
   /testdata/tiff /testoutput/tiff3
 [ -d testoutput/tiff3/color ]

tests/test_metadata.py

@@ -0,0 +1,70 @@
+import numpy as np
+import os
+
+from wkcuber.cubing import ensure_wkw
+from wkcuber.utils import WkwDatasetInfo, open_wkw
+from wkcuber.metadata import (
+    write_webknossos_metadata,
+    refresh_metadata,
+    read_datasource_properties,
+    read_metadata_for_layer,
+)
+
+
+def test_element_class_convertion():
+    test_wkw_path = os.path.join("testoutput", "test_metadata")
+    prediction_layer_name = "prediction"
+    prediction_wkw_info = WkwDatasetInfo(
+        test_wkw_path, prediction_layer_name, np.float32, 1
+    )
+    ensure_wkw(prediction_wkw_info, num_channels=3)
+
+    write_custom_layer(test_wkw_path, "prediction", np.float32, num_channels=3)
+    write_webknossos_metadata(
+        test_wkw_path,
+        "test_metadata",
+        [11.24, 11.24, 28],
+        compute_max_id=True,
+        exact_bounding_box={"topLeft": [0, 0, 0], "width": 4, "height": 4, "depth": 4},
+    )
+    write_custom_layer(test_wkw_path, "segmentation", np.float64, num_channels=1)
+    write_custom_layer(test_wkw_path, "color", np.uint8, num_channels=3)
+
+    refresh_metadata(test_wkw_path)
+
+    check_element_class_of_layer(test_wkw_path, "prediction", "float", np.float32)
+    check_element_class_of_layer(test_wkw_path, "segmentation", "double", np.float64)
+    check_element_class_of_layer(test_wkw_path, "color", "uint24", np.uint8)
+
+
+def check_element_class_of_layer(
+    test_wkw_path, layer_name, expected_element_class, expected_dtype
+):
+    datasource_properties = read_datasource_properties(test_wkw_path)
+    layer_to_check = None
+    for layer in datasource_properties["dataLayers"]:
+        if layer["name"] == layer_name:
+            layer_to_check = layer
+
+    assert (
+        layer_to_check
+    ), f"Did not find layer {layer_name} in datasource_properties.json."
+    assert layer_to_check["elementClass"] == expected_element_class
+    _, converted_dtype, _, _ = read_metadata_for_layer(test_wkw_path, layer_name)
+    assert converted_dtype == expected_dtype
+
+
+def write_custom_layer(target_path, layer_name, dtype, num_channels):
+    data = (
+        np.arange(4 * 4 * 4 * num_channels)
+        .reshape((num_channels, 4, 4, 4))
+        .astype(dtype)
+    )
+    prediction_wkw_info = WkwDatasetInfo(target_path, layer_name, dtype, 1)
+    ensure_wkw(prediction_wkw_info, num_channels=num_channels)
+    with open_wkw(prediction_wkw_info, num_channels=num_channels) as dataset:
+        dataset.write(off=(0, 0, 0), data=data)
+
+
+if __name__ == "__main__":
+    test_element_class_convertion()

wkcuber/metadata.py

@@ -131,13 +131,26 @@ def refresh_metadata(
     write_datasource_properties(wkw_path, datasource_properties)
 
 
+def convert_element_class_to_dtype(elementClass):
+    default_dtype = np.uint8 if "uint" in elementClass else np.dtype(elementClass)
+    conversion_map = {
+        "float": np.float32,
+        "double": np.float64,
+        "uint8": np.uint8,
+        "uint16": np.uint16,
+        "uint32": np.uint32,
+        "uint64": np.uint64,
+    }
+    return conversion_map.get(elementClass, default_dtype)
+
+
 def read_metadata_for_layer(wkw_path, layer_name):
     datasource_properties = read_datasource_properties(wkw_path)
 
     layers = datasource_properties["dataLayers"]
     layer_info = next(layer for layer in layers if layer["name"] == layer_name)
 
-    dtype = np.dtype(layer_info["elementClass"])
+    dtype = convert_element_class_to_dtype(layer_info["elementClass"])
     bounding_box = layer_info["boundingBox"]
     origin = bounding_box["topLeft"]
     bounding_box = [
@@ -149,6 +162,19 @@ def read_metadata_for_layer(wkw_path, layer_name):
     return layer_info, dtype, bounding_box, origin
 
 
+def convert_dype_to_element_class(dtype):
+    element_class_to_dtype_map = {
+        "float": np.float32,
+        "double": np.float64,
+        "uint8": np.uint8,
+        "uint16": np.uint16,
+        "uint32": np.uint32,
+        "uint64": np.uint64,
+    }
+    conversion_map = {v: k for k, v in element_class_to_dtype_map.items()}
+    return conversion_map.get(dtype.type, str(dtype))
+
+
 def detect_dtype(dataset_path, layer, mag: Mag = Mag(1)):
     layer_path = path.join(dataset_path, layer, str(mag))
     if path.exists(layer_path):
@@ -159,7 +185,7 @@ def detect_dtype(dataset_path, layer, mag: Mag = Mag(1)):
             if voxel_size == np.uint8 and num_channels > 1:
                 return "uint" + str(8 * num_channels)
             else:
-                return str(np.dtype(voxel_type))
+                return convert_dype_to_element_class(voxel_size)
 
 
 def detect_cubeLength(dataset_path, layer, mag: Mag = Mag(1)):
@@ -302,7 +328,7 @@ def detect_layers(dataset_path, max_id, compute_max_id, exact_bounding_box=None)
         )
     else:
         if args.name is not None:
-            logging.warn(
+            logging.warning(
                 "The --name argument is ignored, since --refresh was provided."
             )
         refresh_metadata(args.path, args.max_id, args.compute_max_id)