We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
float/double instead of float32/float64
webknossos expects the datatypes to be called `float` and `double` instead of `float32` or `float64` in the `datasource-properties.json`. This needs to be converted both ways in `wkcuber.metadata`

--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `wkcuber/metadata.py`
Content:
```
1 import json
2 import re
3 import wkw
4 import logging
5 import numpy as np
6 
7 from argparse import ArgumentParser
8 from glob import iglob
9 from os import path, listdir
10 from typing import Optional
11 from .mag import Mag
12 from typing import List
13 from .utils import add_verbose_flag, setup_logging, add_scale_flag
14 
15 
16 def get_datasource_path(dataset_path):
17     return path.join(dataset_path, "datasource-properties.json")
18 
19 
20 def create_parser():
21     parser = ArgumentParser()
22 
23     parser.add_argument("path", help="Directory containing the dataset.")
24 
25     parser.add_argument("--name", "-n", help="Name of the dataset", default=None)
26 
27     parser.add_argument("--refresh", "-r", default=False, action="store_true")
28 
29     group = parser.add_mutually_exclusive_group()
30     group.add_argument(
31         "--compute_max_id",
32         "-c",
33         help="set to compute max id",
34         default=False,
35         action="store_true",
36     )
37     group.add_argument("--max_id", help="set max id of segmentation.", default=0)
38 
39     add_scale_flag(parser, required=False)
40     add_verbose_flag(parser)
41 
42     return parser
43 
44 
45 def write_datasource_properties(dataset_path, datasource_properties):
46     datasource_properties_path = get_datasource_path(dataset_path)
47     with open(datasource_properties_path, "wt") as datasource_properties_file:
48         json.dump(datasource_properties, datasource_properties_file, indent=2)
49 
50 
51 def read_datasource_properties(dataset_path):
52     with open(get_datasource_path(dataset_path), "r") as datasource_properties_file:
53         return json.load(datasource_properties_file)
54 
55 
56 """
57 Creates a datasource-properties.json file with the specified properties
58 for the given dataset path. Common layers are detected automatically.
59 """
60 
61 
62 def write_webknossos_metadata(
63     dataset_path,
64     name,
65     scale,
66     max_id=0,
67     compute_max_id=False,
68     exact_bounding_box: Optional[dict] = None,
69 ):
70     if name == None:
71         name = path.basename(dataset_path)
72 
73     # Generate a metadata file for webKnossos
74     # Currently includes no source of information for team
75     layers = list(
76         detect_layers(dataset_path, max_id, compute_max_id, exact_bounding_box)
77     )
78     write_datasource_properties(
79         dataset_path,
80         {
81             "id": {"name": name, "team": "<unknown>"},
82             "dataLayers": layers,
83             "scale": scale,
84         },
85     )
86 
87 
88 """
89 Updates the datasource-properties.json file for a given dataset.
90 Use this method if you added (or removed) layers and/or changed magnifications for
91 existing layers.
92 
93 Raises an exception if the datasource-properties.json file does not exist, yet.
94 In this case, use write_webknossos_metadata instead.
95 """
96 
97 
98 def refresh_metadata(
99     wkw_path, max_id=0, compute_max_id=False, exact_bounding_box: Optional[dict] = None
100 ):
101     dataset_path = get_datasource_path(wkw_path)
102     if not path.exists(dataset_path):
103         raise Exception(
104             "datasource-properties.json file could not be found. Please use write_webknossos_metadata to create it."
105         )
106 
107     datasource_properties = read_datasource_properties(wkw_path)
108     existing_layers_dict = {
109         layer["name"]: layer for layer in datasource_properties["dataLayers"]
110     }
111 
112     new_layers = list(
113         detect_layers(wkw_path, max_id, compute_max_id, exact_bounding_box)
114     )
115 
116     # Merge the freshly read layers with the existing layer information, so that information,
117     # such as bounding boxes, are not lost for existing layers.
118     # For existing layers, only the resolutions will be updated.
119     merged_layers = []
120     for new_layer in new_layers:
121         layer_name = new_layer["name"]
122         if layer_name in existing_layers_dict:
123             existing_layer = existing_layers_dict[layer_name]
124             # Update the resolutions
125             existing_layer["wkwResolutions"] = new_layer["wkwResolutions"]
126             merged_layers.append(existing_layer)
127         else:
128             merged_layers.append(new_layer)
129 
130     datasource_properties["dataLayers"] = merged_layers
131     write_datasource_properties(wkw_path, datasource_properties)
132 
133 
134 def read_metadata_for_layer(wkw_path, layer_name):
135     datasource_properties = read_datasource_properties(wkw_path)
136 
137     layers = datasource_properties["dataLayers"]
138     layer_info = next(layer for layer in layers if layer["name"] == layer_name)
139 
140     dtype = np.dtype(layer_info["elementClass"])
141     bounding_box = layer_info["boundingBox"]
142     origin = bounding_box["topLeft"]
143     bounding_box = [
144         bounding_box["width"],
145         bounding_box["height"],
146         bounding_box["depth"],
147     ]
148 
149     return layer_info, dtype, bounding_box, origin
150 
151 
152 def detect_dtype(dataset_path, layer, mag: Mag = Mag(1)):
153     layer_path = path.join(dataset_path, layer, str(mag))
154     if path.exists(layer_path):
155         with wkw.Dataset.open(layer_path) as dataset:
156             voxel_type = dataset.header.voxel_type
157             num_channels = dataset.header.num_channels
158             voxel_size = np.dtype(voxel_type)
159             if voxel_size == np.uint8 and num_channels > 1:
160                 return "uint" + str(8 * num_channels)
161             else:
162                 return str(np.dtype(voxel_type))
163 
164 
165 def detect_cubeLength(dataset_path, layer, mag: Mag = Mag(1)):
166     layer_path = path.join(dataset_path, layer, str(mag))
167     if path.exists(layer_path):
168         with wkw.Dataset.open(layer_path) as dataset:
169             return dataset.header.block_len * dataset.header.file_len
170 
171 
172 def detect_bbox(dataset_path, layer, mag: Mag = Mag(1)):
173     # Detect the coarse bounding box of a dataset by iterating
174     # over the WKW cubes
175     layer_path = path.join(dataset_path, layer, str(mag))
176 
177     def list_files(layer_path):
178         return iglob(path.join(layer_path, "*", "*", "*.wkw"), recursive=True)
179 
180     def parse_cube_file_name(filename):
181         CUBE_REGEX = re.compile(r"z(\d+)/y(\d+)/x(\d+)(\.wkw)$")
182         m = CUBE_REGEX.search(filename)
183         return (int(m.group(3)), int(m.group(2)), int(m.group(1)))
184 
185     def list_cubes(layer_path):
186         return (parse_cube_file_name(f) for f in list_files(layer_path))
187 
188     xs, ys, zs = list(zip(*list_cubes(layer_path)))
189 
190     min_x, min_y, min_z = min(xs), min(ys), min(zs)
191     max_x, max_y, max_z = max(xs), max(ys), max(zs)
192 
193     cubeLength = detect_cubeLength(dataset_path, layer, mag)
194 
195     return {
196         "topLeft": [min_x * cubeLength, min_y * cubeLength, min_z * cubeLength],
197         "width": (1 + max_x - min_x) * cubeLength,
198         "height": (1 + max_y - min_y) * cubeLength,
199         "depth": (1 + max_z - min_z) * cubeLength,
200     }
201 
202 
203 def detect_resolutions(dataset_path, layer) -> List[Mag]:
204     for mag in listdir(path.join(dataset_path, layer)):
205         try:
206             yield Mag(mag)
207         except ValueError:
208             logging.info("ignoring {} as resolution".format(mag))
209 
210 
211 def detect_standard_layer(
212     dataset_path, layer_name, exact_bounding_box=None, category="color"
213 ):
214     # Perform metadata detection for well-known layers
215 
216     mags = list(detect_resolutions(dataset_path, layer_name))
217     mags = sorted(mags)
218     assert len(mags) > 0, "No resolutions found"
219 
220     if exact_bounding_box is None:
221         bbox = detect_bbox(dataset_path, layer_name, mags[0])
222     else:
223         bbox = exact_bounding_box
224 
225     resolutions = [
226         {
227             "resolution": mag.to_array(),
228             "cubeLength": detect_cubeLength(dataset_path, layer_name, mag),
229         }
230         for mag in mags
231     ]
232 
233     dtype = detect_dtype(dataset_path, layer_name, mags[0])
234 
235     return {
236         "dataFormat": "wkw",
237         "name": layer_name,
238         "category": category,
239         "elementClass": dtype,
240         "boundingBox": bbox,
241         "wkwResolutions": list(resolutions),
242     }
243 
244 
245 def detect_segmentation_layer(
246     dataset_path, layer_name, max_id, compute_max_id=False, exact_bounding_box=None
247 ):
248     layer_info = detect_standard_layer(
249         dataset_path, layer_name, exact_bounding_box, category="segmentation"
250     )
251     layer_info["mappings"] = []
252     layer_info["largestSegmentId"] = max_id
253 
254     if compute_max_id:
255         logging.info("Computing max id of layer={}".format(layer_name))
256         # Computing the current largest segment id
257         # This may take very long due to IO load
258         layer_path = path.join(dataset_path, layer_name, "1")
259         with wkw.Dataset.open(layer_path) as dataset:
260             bbox = layer_info["boundingBox"]
261             layer_info["largestSegmentId"] = int(
262                 np.max(
263                     dataset.read(
264                         bbox["topLeft"], [bbox["width"], bbox["height"], bbox["depth"]]
265                     )
266                 )
267             )
268             logging.info(
269                 "Max id of layer={} is {}".format(
270                     layer_name, layer_info["largestSegmentId"]
271                 )
272             )
273 
274     return layer_info
275 
276 
277 def detect_layers(dataset_path, max_id, compute_max_id, exact_bounding_box=None):
278     # Detect metadata for well-known layers (i.e., color, prediction and segmentation)
279     for layer_name in ["color", "prediction"]:
280         if path.exists(path.join(dataset_path, layer_name)):
281             yield detect_standard_layer(dataset_path, layer_name, exact_bounding_box)
282     if path.exists(path.join(dataset_path, "segmentation")):
283         yield detect_segmentation_layer(
284             dataset_path, "segmentation", max_id, compute_max_id, exact_bounding_box
285         )
286 
287 
288 if __name__ == "__main__":
289     args = create_parser().parse_args()
290     setup_logging(args)
291 
292     if not args.refresh:
293         assert (
294             args.scale is not None
295         ), "The scale has to be specified when creating metadata for a dataset."
296         assert (
297             args.name is not None
298         ), "Please provide a name via --name to create meta data."
299         scale = tuple(float(x) for x in args.scale.split(","))
300         write_webknossos_metadata(
301             args.path, args.name, scale, args.max_id, args.compute_max_id
302         )
303     else:
304         if args.name is not None:
305             logging.warn(
306                 "The --name argument is ignored, since --refresh was provided."
307             )
308         refresh_metadata(args.path, args.max_id, args.compute_max_id)
309
```
--- END FILES ---

Please first localize the bug based on the issue statement, and then generate a patch according to the `git diff` format fenced by three backticks.

Here is an example:

```diff
diff --git a/examples/server_async.py b/examples/server_async.py
--- a/examples/server_async.py
+++ b/examples/server_async.py
@@ -313,4 +313,4 @@
 
 
 if __name__ == "__main__":
-    asyncio.run(run_async_server("."), debug=True)
+    asyncio.run(run_async_server(), debug=True)
diff --git a/examples/server_sync.py b/examples/server_sync.py
--- a/examples/server_sync.py
+++ b/examples/server_sync.py
@@ -313,5 +313,5 @@
 
 
 if __name__ == "__main__":
-    server = run_sync_server(".")
+    server = run_sync_server()
     server.shutdown()

```

diff --git a/wkcuber/metadata.py b/wkcuber/metadata.py
--- a/wkcuber/metadata.py
+++ b/wkcuber/metadata.py
@@ -131,13 +131,26 @@
     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 @@
     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 @@
             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 @@
         )
     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)

{"golden_diff": "diff --git a/wkcuber/metadata.py b/wkcuber/metadata.py\n--- a/wkcuber/metadata.py\n+++ b/wkcuber/metadata.py\n@@ -131,13 +131,26 @@\n     write_datasource_properties(wkw_path, datasource_properties)\n \n \n+def convert_element_class_to_dtype(elementClass):\n+    default_dtype = np.uint8 if \"uint\" in elementClass else np.dtype(elementClass)\n+    conversion_map = {\n+        \"float\": np.float32,\n+        \"double\": np.float64,\n+        \"uint8\": np.uint8,\n+        \"uint16\": np.uint16,\n+        \"uint32\": np.uint32,\n+        \"uint64\": np.uint64,\n+    }\n+    return conversion_map.get(elementClass, default_dtype)\n+\n+\n def read_metadata_for_layer(wkw_path, layer_name):\n     datasource_properties = read_datasource_properties(wkw_path)\n \n     layers = datasource_properties[\"dataLayers\"]\n     layer_info = next(layer for layer in layers if layer[\"name\"] == layer_name)\n \n-    dtype = np.dtype(layer_info[\"elementClass\"])\n+    dtype = convert_element_class_to_dtype(layer_info[\"elementClass\"])\n     bounding_box = layer_info[\"boundingBox\"]\n     origin = bounding_box[\"topLeft\"]\n     bounding_box = [\n@@ -149,6 +162,19 @@\n     return layer_info, dtype, bounding_box, origin\n \n \n+def convert_dype_to_element_class(dtype):\n+    element_class_to_dtype_map = {\n+        \"float\": np.float32,\n+        \"double\": np.float64,\n+        \"uint8\": np.uint8,\n+        \"uint16\": np.uint16,\n+        \"uint32\": np.uint32,\n+        \"uint64\": np.uint64,\n+    }\n+    conversion_map = {v: k for k, v in element_class_to_dtype_map.items()}\n+    return conversion_map.get(dtype.type, str(dtype))\n+\n+\n def detect_dtype(dataset_path, layer, mag: Mag = Mag(1)):\n     layer_path = path.join(dataset_path, layer, str(mag))\n     if path.exists(layer_path):\n@@ -159,7 +185,7 @@\n             if voxel_size == np.uint8 and num_channels > 1:\n                 return \"uint\" + str(8 * num_channels)\n             else:\n-                return str(np.dtype(voxel_type))\n+                return convert_dype_to_element_class(voxel_size)\n \n \n def detect_cubeLength(dataset_path, layer, mag: Mag = Mag(1)):\n@@ -302,7 +328,7 @@\n         )\n     else:\n         if args.name is not None:\n-            logging.warn(\n+            logging.warning(\n                 \"The --name argument is ignored, since --refresh was provided.\"\n             )\n         refresh_metadata(args.path, args.max_id, args.compute_max_id)\n", "issue": "float/double instead of float32/float64\nwebknossos expects the datatypes to be called `float` and `double` instead of `float32` or `float64` in the `datasource-properties.json`. This needs to be converted both ways in `wkcuber.metadata`\n", "before_files": [{"content": "import json\nimport re\nimport wkw\nimport logging\nimport numpy as np\n\nfrom argparse import ArgumentParser\nfrom glob import iglob\nfrom os import path, listdir\nfrom typing import Optional\nfrom .mag import Mag\nfrom typing import List\nfrom .utils import add_verbose_flag, setup_logging, add_scale_flag\n\n\ndef get_datasource_path(dataset_path):\n    return path.join(dataset_path, \"datasource-properties.json\")\n\n\ndef create_parser():\n    parser = ArgumentParser()\n\n    parser.add_argument(\"path\", help=\"Directory containing the dataset.\")\n\n    parser.add_argument(\"--name\", \"-n\", help=\"Name of the dataset\", default=None)\n\n    parser.add_argument(\"--refresh\", \"-r\", default=False, action=\"store_true\")\n\n    group = parser.add_mutually_exclusive_group()\n    group.add_argument(\n        \"--compute_max_id\",\n        \"-c\",\n        help=\"set to compute max id\",\n        default=False,\n        action=\"store_true\",\n    )\n    group.add_argument(\"--max_id\", help=\"set max id of segmentation.\", default=0)\n\n    add_scale_flag(parser, required=False)\n    add_verbose_flag(parser)\n\n    return parser\n\n\ndef write_datasource_properties(dataset_path, datasource_properties):\n    datasource_properties_path = get_datasource_path(dataset_path)\n    with open(datasource_properties_path, \"wt\") as datasource_properties_file:\n        json.dump(datasource_properties, datasource_properties_file, indent=2)\n\n\ndef read_datasource_properties(dataset_path):\n    with open(get_datasource_path(dataset_path), \"r\") as datasource_properties_file:\n        return json.load(datasource_properties_file)\n\n\n\"\"\"\nCreates a datasource-properties.json file with the specified properties\nfor the given dataset path. Common layers are detected automatically.\n\"\"\"\n\n\ndef write_webknossos_metadata(\n    dataset_path,\n    name,\n    scale,\n    max_id=0,\n    compute_max_id=False,\n    exact_bounding_box: Optional[dict] = None,\n):\n    if name == None:\n        name = path.basename(dataset_path)\n\n    # Generate a metadata file for webKnossos\n    # Currently includes no source of information for team\n    layers = list(\n        detect_layers(dataset_path, max_id, compute_max_id, exact_bounding_box)\n    )\n    write_datasource_properties(\n        dataset_path,\n        {\n            \"id\": {\"name\": name, \"team\": \"<unknown>\"},\n            \"dataLayers\": layers,\n            \"scale\": scale,\n        },\n    )\n\n\n\"\"\"\nUpdates the datasource-properties.json file for a given dataset.\nUse this method if you added (or removed) layers and/or changed magnifications for\nexisting layers.\n\nRaises an exception if the datasource-properties.json file does not exist, yet.\nIn this case, use write_webknossos_metadata instead.\n\"\"\"\n\n\ndef refresh_metadata(\n    wkw_path, max_id=0, compute_max_id=False, exact_bounding_box: Optional[dict] = None\n):\n    dataset_path = get_datasource_path(wkw_path)\n    if not path.exists(dataset_path):\n        raise Exception(\n            \"datasource-properties.json file could not be found. Please use write_webknossos_metadata to create it.\"\n        )\n\n    datasource_properties = read_datasource_properties(wkw_path)\n    existing_layers_dict = {\n        layer[\"name\"]: layer for layer in datasource_properties[\"dataLayers\"]\n    }\n\n    new_layers = list(\n        detect_layers(wkw_path, max_id, compute_max_id, exact_bounding_box)\n    )\n\n    # Merge the freshly read layers with the existing layer information, so that information,\n    # such as bounding boxes, are not lost for existing layers.\n    # For existing layers, only the resolutions will be updated.\n    merged_layers = []\n    for new_layer in new_layers:\n        layer_name = new_layer[\"name\"]\n        if layer_name in existing_layers_dict:\n            existing_layer = existing_layers_dict[layer_name]\n            # Update the resolutions\n            existing_layer[\"wkwResolutions\"] = new_layer[\"wkwResolutions\"]\n            merged_layers.append(existing_layer)\n        else:\n            merged_layers.append(new_layer)\n\n    datasource_properties[\"dataLayers\"] = merged_layers\n    write_datasource_properties(wkw_path, datasource_properties)\n\n\ndef read_metadata_for_layer(wkw_path, layer_name):\n    datasource_properties = read_datasource_properties(wkw_path)\n\n    layers = datasource_properties[\"dataLayers\"]\n    layer_info = next(layer for layer in layers if layer[\"name\"] == layer_name)\n\n    dtype = np.dtype(layer_info[\"elementClass\"])\n    bounding_box = layer_info[\"boundingBox\"]\n    origin = bounding_box[\"topLeft\"]\n    bounding_box = [\n        bounding_box[\"width\"],\n        bounding_box[\"height\"],\n        bounding_box[\"depth\"],\n    ]\n\n    return layer_info, dtype, bounding_box, origin\n\n\ndef detect_dtype(dataset_path, layer, mag: Mag = Mag(1)):\n    layer_path = path.join(dataset_path, layer, str(mag))\n    if path.exists(layer_path):\n        with wkw.Dataset.open(layer_path) as dataset:\n            voxel_type = dataset.header.voxel_type\n            num_channels = dataset.header.num_channels\n            voxel_size = np.dtype(voxel_type)\n            if voxel_size == np.uint8 and num_channels > 1:\n                return \"uint\" + str(8 * num_channels)\n            else:\n                return str(np.dtype(voxel_type))\n\n\ndef detect_cubeLength(dataset_path, layer, mag: Mag = Mag(1)):\n    layer_path = path.join(dataset_path, layer, str(mag))\n    if path.exists(layer_path):\n        with wkw.Dataset.open(layer_path) as dataset:\n            return dataset.header.block_len * dataset.header.file_len\n\n\ndef detect_bbox(dataset_path, layer, mag: Mag = Mag(1)):\n    # Detect the coarse bounding box of a dataset by iterating\n    # over the WKW cubes\n    layer_path = path.join(dataset_path, layer, str(mag))\n\n    def list_files(layer_path):\n        return iglob(path.join(layer_path, \"*\", \"*\", \"*.wkw\"), recursive=True)\n\n    def parse_cube_file_name(filename):\n        CUBE_REGEX = re.compile(r\"z(\\d+)/y(\\d+)/x(\\d+)(\\.wkw)$\")\n        m = CUBE_REGEX.search(filename)\n        return (int(m.group(3)), int(m.group(2)), int(m.group(1)))\n\n    def list_cubes(layer_path):\n        return (parse_cube_file_name(f) for f in list_files(layer_path))\n\n    xs, ys, zs = list(zip(*list_cubes(layer_path)))\n\n    min_x, min_y, min_z = min(xs), min(ys), min(zs)\n    max_x, max_y, max_z = max(xs), max(ys), max(zs)\n\n    cubeLength = detect_cubeLength(dataset_path, layer, mag)\n\n    return {\n        \"topLeft\": [min_x * cubeLength, min_y * cubeLength, min_z * cubeLength],\n        \"width\": (1 + max_x - min_x) * cubeLength,\n        \"height\": (1 + max_y - min_y) * cubeLength,\n        \"depth\": (1 + max_z - min_z) * cubeLength,\n    }\n\n\ndef detect_resolutions(dataset_path, layer) -> List[Mag]:\n    for mag in listdir(path.join(dataset_path, layer)):\n        try:\n            yield Mag(mag)\n        except ValueError:\n            logging.info(\"ignoring {} as resolution\".format(mag))\n\n\ndef detect_standard_layer(\n    dataset_path, layer_name, exact_bounding_box=None, category=\"color\"\n):\n    # Perform metadata detection for well-known layers\n\n    mags = list(detect_resolutions(dataset_path, layer_name))\n    mags = sorted(mags)\n    assert len(mags) > 0, \"No resolutions found\"\n\n    if exact_bounding_box is None:\n        bbox = detect_bbox(dataset_path, layer_name, mags[0])\n    else:\n        bbox = exact_bounding_box\n\n    resolutions = [\n        {\n            \"resolution\": mag.to_array(),\n            \"cubeLength\": detect_cubeLength(dataset_path, layer_name, mag),\n        }\n        for mag in mags\n    ]\n\n    dtype = detect_dtype(dataset_path, layer_name, mags[0])\n\n    return {\n        \"dataFormat\": \"wkw\",\n        \"name\": layer_name,\n        \"category\": category,\n        \"elementClass\": dtype,\n        \"boundingBox\": bbox,\n        \"wkwResolutions\": list(resolutions),\n    }\n\n\ndef detect_segmentation_layer(\n    dataset_path, layer_name, max_id, compute_max_id=False, exact_bounding_box=None\n):\n    layer_info = detect_standard_layer(\n        dataset_path, layer_name, exact_bounding_box, category=\"segmentation\"\n    )\n    layer_info[\"mappings\"] = []\n    layer_info[\"largestSegmentId\"] = max_id\n\n    if compute_max_id:\n        logging.info(\"Computing max id of layer={}\".format(layer_name))\n        # Computing the current largest segment id\n        # This may take very long due to IO load\n        layer_path = path.join(dataset_path, layer_name, \"1\")\n        with wkw.Dataset.open(layer_path) as dataset:\n            bbox = layer_info[\"boundingBox\"]\n            layer_info[\"largestSegmentId\"] = int(\n                np.max(\n                    dataset.read(\n                        bbox[\"topLeft\"], [bbox[\"width\"], bbox[\"height\"], bbox[\"depth\"]]\n                    )\n                )\n            )\n            logging.info(\n                \"Max id of layer={} is {}\".format(\n                    layer_name, layer_info[\"largestSegmentId\"]\n                )\n            )\n\n    return layer_info\n\n\ndef detect_layers(dataset_path, max_id, compute_max_id, exact_bounding_box=None):\n    # Detect metadata for well-known layers (i.e., color, prediction and segmentation)\n    for layer_name in [\"color\", \"prediction\"]:\n        if path.exists(path.join(dataset_path, layer_name)):\n            yield detect_standard_layer(dataset_path, layer_name, exact_bounding_box)\n    if path.exists(path.join(dataset_path, \"segmentation\")):\n        yield detect_segmentation_layer(\n            dataset_path, \"segmentation\", max_id, compute_max_id, exact_bounding_box\n        )\n\n\nif __name__ == \"__main__\":\n    args = create_parser().parse_args()\n    setup_logging(args)\n\n    if not args.refresh:\n        assert (\n            args.scale is not None\n        ), \"The scale has to be specified when creating metadata for a dataset.\"\n        assert (\n            args.name is not None\n        ), \"Please provide a name via --name to create meta data.\"\n        scale = tuple(float(x) for x in args.scale.split(\",\"))\n        write_webknossos_metadata(\n            args.path, args.name, scale, args.max_id, args.compute_max_id\n        )\n    else:\n        if args.name is not None:\n            logging.warn(\n                \"The --name argument is ignored, since --refresh was provided.\"\n            )\n        refresh_metadata(args.path, args.max_id, args.compute_max_id)\n", "path": "wkcuber/metadata.py"}], "after_files": [{"content": "import json\nimport re\nimport wkw\nimport logging\nimport numpy as np\n\nfrom argparse import ArgumentParser\nfrom glob import iglob\nfrom os import path, listdir\nfrom typing import Optional\nfrom .mag import Mag\nfrom typing import List\nfrom .utils import add_verbose_flag, setup_logging, add_scale_flag\n\n\ndef get_datasource_path(dataset_path):\n    return path.join(dataset_path, \"datasource-properties.json\")\n\n\ndef create_parser():\n    parser = ArgumentParser()\n\n    parser.add_argument(\"path\", help=\"Directory containing the dataset.\")\n\n    parser.add_argument(\"--name\", \"-n\", help=\"Name of the dataset\", default=None)\n\n    parser.add_argument(\"--refresh\", \"-r\", default=False, action=\"store_true\")\n\n    group = parser.add_mutually_exclusive_group()\n    group.add_argument(\n        \"--compute_max_id\",\n        \"-c\",\n        help=\"set to compute max id\",\n        default=False,\n        action=\"store_true\",\n    )\n    group.add_argument(\"--max_id\", help=\"set max id of segmentation.\", default=0)\n\n    add_scale_flag(parser, required=False)\n    add_verbose_flag(parser)\n\n    return parser\n\n\ndef write_datasource_properties(dataset_path, datasource_properties):\n    datasource_properties_path = get_datasource_path(dataset_path)\n    with open(datasource_properties_path, \"wt\") as datasource_properties_file:\n        json.dump(datasource_properties, datasource_properties_file, indent=2)\n\n\ndef read_datasource_properties(dataset_path):\n    with open(get_datasource_path(dataset_path), \"r\") as datasource_properties_file:\n        return json.load(datasource_properties_file)\n\n\n\"\"\"\nCreates a datasource-properties.json file with the specified properties\nfor the given dataset path. Common layers are detected automatically.\n\"\"\"\n\n\ndef write_webknossos_metadata(\n    dataset_path,\n    name,\n    scale,\n    max_id=0,\n    compute_max_id=False,\n    exact_bounding_box: Optional[dict] = None,\n):\n    if name == None:\n        name = path.basename(dataset_path)\n\n    # Generate a metadata file for webKnossos\n    # Currently includes no source of information for team\n    layers = list(\n        detect_layers(dataset_path, max_id, compute_max_id, exact_bounding_box)\n    )\n    write_datasource_properties(\n        dataset_path,\n        {\n            \"id\": {\"name\": name, \"team\": \"<unknown>\"},\n            \"dataLayers\": layers,\n            \"scale\": scale,\n        },\n    )\n\n\n\"\"\"\nUpdates the datasource-properties.json file for a given dataset.\nUse this method if you added (or removed) layers and/or changed magnifications for\nexisting layers.\n\nRaises an exception if the datasource-properties.json file does not exist, yet.\nIn this case, use write_webknossos_metadata instead.\n\"\"\"\n\n\ndef refresh_metadata(\n    wkw_path, max_id=0, compute_max_id=False, exact_bounding_box: Optional[dict] = None\n):\n    dataset_path = get_datasource_path(wkw_path)\n    if not path.exists(dataset_path):\n        raise Exception(\n            \"datasource-properties.json file could not be found. Please use write_webknossos_metadata to create it.\"\n        )\n\n    datasource_properties = read_datasource_properties(wkw_path)\n    existing_layers_dict = {\n        layer[\"name\"]: layer for layer in datasource_properties[\"dataLayers\"]\n    }\n\n    new_layers = list(\n        detect_layers(wkw_path, max_id, compute_max_id, exact_bounding_box)\n    )\n\n    # Merge the freshly read layers with the existing layer information, so that information,\n    # such as bounding boxes, are not lost for existing layers.\n    # For existing layers, only the resolutions will be updated.\n    merged_layers = []\n    for new_layer in new_layers:\n        layer_name = new_layer[\"name\"]\n        if layer_name in existing_layers_dict:\n            existing_layer = existing_layers_dict[layer_name]\n            # Update the resolutions\n            existing_layer[\"wkwResolutions\"] = new_layer[\"wkwResolutions\"]\n            merged_layers.append(existing_layer)\n        else:\n            merged_layers.append(new_layer)\n\n    datasource_properties[\"dataLayers\"] = merged_layers\n    write_datasource_properties(wkw_path, datasource_properties)\n\n\ndef convert_element_class_to_dtype(elementClass):\n    default_dtype = np.uint8 if \"uint\" in elementClass else np.dtype(elementClass)\n    conversion_map = {\n        \"float\": np.float32,\n        \"double\": np.float64,\n        \"uint8\": np.uint8,\n        \"uint16\": np.uint16,\n        \"uint32\": np.uint32,\n        \"uint64\": np.uint64,\n    }\n    return conversion_map.get(elementClass, default_dtype)\n\n\ndef read_metadata_for_layer(wkw_path, layer_name):\n    datasource_properties = read_datasource_properties(wkw_path)\n\n    layers = datasource_properties[\"dataLayers\"]\n    layer_info = next(layer for layer in layers if layer[\"name\"] == layer_name)\n\n    dtype = convert_element_class_to_dtype(layer_info[\"elementClass\"])\n    bounding_box = layer_info[\"boundingBox\"]\n    origin = bounding_box[\"topLeft\"]\n    bounding_box = [\n        bounding_box[\"width\"],\n        bounding_box[\"height\"],\n        bounding_box[\"depth\"],\n    ]\n\n    return layer_info, dtype, bounding_box, origin\n\n\ndef convert_dype_to_element_class(dtype):\n    element_class_to_dtype_map = {\n        \"float\": np.float32,\n        \"double\": np.float64,\n        \"uint8\": np.uint8,\n        \"uint16\": np.uint16,\n        \"uint32\": np.uint32,\n        \"uint64\": np.uint64,\n    }\n    conversion_map = {v: k for k, v in element_class_to_dtype_map.items()}\n    return conversion_map.get(dtype.type, str(dtype))\n\n\ndef detect_dtype(dataset_path, layer, mag: Mag = Mag(1)):\n    layer_path = path.join(dataset_path, layer, str(mag))\n    if path.exists(layer_path):\n        with wkw.Dataset.open(layer_path) as dataset:\n            voxel_type = dataset.header.voxel_type\n            num_channels = dataset.header.num_channels\n            voxel_size = np.dtype(voxel_type)\n            if voxel_size == np.uint8 and num_channels > 1:\n                return \"uint\" + str(8 * num_channels)\n            else:\n                return convert_dype_to_element_class(voxel_size)\n\n\ndef detect_cubeLength(dataset_path, layer, mag: Mag = Mag(1)):\n    layer_path = path.join(dataset_path, layer, str(mag))\n    if path.exists(layer_path):\n        with wkw.Dataset.open(layer_path) as dataset:\n            return dataset.header.block_len * dataset.header.file_len\n\n\ndef detect_bbox(dataset_path, layer, mag: Mag = Mag(1)):\n    # Detect the coarse bounding box of a dataset by iterating\n    # over the WKW cubes\n    layer_path = path.join(dataset_path, layer, str(mag))\n\n    def list_files(layer_path):\n        return iglob(path.join(layer_path, \"*\", \"*\", \"*.wkw\"), recursive=True)\n\n    def parse_cube_file_name(filename):\n        CUBE_REGEX = re.compile(r\"z(\\d+)/y(\\d+)/x(\\d+)(\\.wkw)$\")\n        m = CUBE_REGEX.search(filename)\n        return (int(m.group(3)), int(m.group(2)), int(m.group(1)))\n\n    def list_cubes(layer_path):\n        return (parse_cube_file_name(f) for f in list_files(layer_path))\n\n    xs, ys, zs = list(zip(*list_cubes(layer_path)))\n\n    min_x, min_y, min_z = min(xs), min(ys), min(zs)\n    max_x, max_y, max_z = max(xs), max(ys), max(zs)\n\n    cubeLength = detect_cubeLength(dataset_path, layer, mag)\n\n    return {\n        \"topLeft\": [min_x * cubeLength, min_y * cubeLength, min_z * cubeLength],\n        \"width\": (1 + max_x - min_x) * cubeLength,\n        \"height\": (1 + max_y - min_y) * cubeLength,\n        \"depth\": (1 + max_z - min_z) * cubeLength,\n    }\n\n\ndef detect_resolutions(dataset_path, layer) -> List[Mag]:\n    for mag in listdir(path.join(dataset_path, layer)):\n        try:\n            yield Mag(mag)\n        except ValueError:\n            logging.info(\"ignoring {} as resolution\".format(mag))\n\n\ndef detect_standard_layer(\n    dataset_path, layer_name, exact_bounding_box=None, category=\"color\"\n):\n    # Perform metadata detection for well-known layers\n\n    mags = list(detect_resolutions(dataset_path, layer_name))\n    mags = sorted(mags)\n    assert len(mags) > 0, \"No resolutions found\"\n\n    if exact_bounding_box is None:\n        bbox = detect_bbox(dataset_path, layer_name, mags[0])\n    else:\n        bbox = exact_bounding_box\n\n    resolutions = [\n        {\n            \"resolution\": mag.to_array(),\n            \"cubeLength\": detect_cubeLength(dataset_path, layer_name, mag),\n        }\n        for mag in mags\n    ]\n\n    dtype = detect_dtype(dataset_path, layer_name, mags[0])\n\n    return {\n        \"dataFormat\": \"wkw\",\n        \"name\": layer_name,\n        \"category\": category,\n        \"elementClass\": dtype,\n        \"boundingBox\": bbox,\n        \"wkwResolutions\": list(resolutions),\n    }\n\n\ndef detect_segmentation_layer(\n    dataset_path, layer_name, max_id, compute_max_id=False, exact_bounding_box=None\n):\n    layer_info = detect_standard_layer(\n        dataset_path, layer_name, exact_bounding_box, category=\"segmentation\"\n    )\n    layer_info[\"mappings\"] = []\n    layer_info[\"largestSegmentId\"] = max_id\n\n    if compute_max_id:\n        logging.info(\"Computing max id of layer={}\".format(layer_name))\n        # Computing the current largest segment id\n        # This may take very long due to IO load\n        layer_path = path.join(dataset_path, layer_name, \"1\")\n        with wkw.Dataset.open(layer_path) as dataset:\n            bbox = layer_info[\"boundingBox\"]\n            layer_info[\"largestSegmentId\"] = int(\n                np.max(\n                    dataset.read(\n                        bbox[\"topLeft\"], [bbox[\"width\"], bbox[\"height\"], bbox[\"depth\"]]\n                    )\n                )\n            )\n            logging.info(\n                \"Max id of layer={} is {}\".format(\n                    layer_name, layer_info[\"largestSegmentId\"]\n                )\n            )\n\n    return layer_info\n\n\ndef detect_layers(dataset_path, max_id, compute_max_id, exact_bounding_box=None):\n    # Detect metadata for well-known layers (i.e., color, prediction and segmentation)\n    for layer_name in [\"color\", \"prediction\"]:\n        if path.exists(path.join(dataset_path, layer_name)):\n            yield detect_standard_layer(dataset_path, layer_name, exact_bounding_box)\n    if path.exists(path.join(dataset_path, \"segmentation\")):\n        yield detect_segmentation_layer(\n            dataset_path, \"segmentation\", max_id, compute_max_id, exact_bounding_box\n        )\n\n\nif __name__ == \"__main__\":\n    args = create_parser().parse_args()\n    setup_logging(args)\n\n    if not args.refresh:\n        assert (\n            args.scale is not None\n        ), \"The scale has to be specified when creating metadata for a dataset.\"\n        assert (\n            args.name is not None\n        ), \"Please provide a name via --name to create meta data.\"\n        scale = tuple(float(x) for x in args.scale.split(\",\"))\n        write_webknossos_metadata(\n            args.path, args.name, scale, args.max_id, args.compute_max_id\n        )\n    else:\n        if args.name is not None:\n            logging.warning(\n                \"The --name argument is ignored, since --refresh was provided.\"\n            )\n        refresh_metadata(args.path, args.max_id, args.compute_max_id)\n", "path": "wkcuber/metadata.py"}]}