Add method to Dataset API to copy an existing layer from a different dataset (#345)

* implement add_copy_layer

* update changelog

Author: rschwanhold

Changelog.md

@@ -12,6 +12,7 @@ For upgrade instructions, please check the respective *Breaking Changes* section
 ### Breaking Changes in Config & CLI
 
 ### Added
+- Added `add_copy_layer()` to `wkcuber.api.dataset.Dataset` to copy the layer of a different dataset. [#345](https://github.com/scalableminds/webknossos-cuber/pull/345)
 
 ### Changed
 

tests/test_dataset.py

@@ -14,6 +14,7 @@
 from wkw import wkw
 from wkw.wkw import WKWException
 
+from wkcuber.api.bounding_box import BoundingBox
 from wkcuber.api.dataset import Dataset
 from os import makedirs
 
@@ -1439,3 +1440,37 @@ def test_dataset_name(tmp_path: Path) -> None:
         tmp_path / "some_new_name", scale=(1, 1, 1), name="very important dataset"
     )
     assert ds2.name == "very important dataset"
+
+
+def test_add_copy_layer(tmp_path: Path) -> None:
+    ds = Dataset.create(tmp_path / "ds", scale=(2, 2, 1))
+
+    # Create dataset to copy data from
+    other_ds = Dataset.create(tmp_path / "other_ds", scale=(2, 2, 1))
+    original_color_layer = other_ds.add_layer("color", LayerCategories.COLOR_TYPE)
+    original_color_layer.add_mag(1).write(
+        offset=(10, 20, 30), data=(np.random.rand(32, 64, 128) * 255).astype(np.uint8)
+    )
+
+    # Copies the "color" layer from a different dataset
+    ds.add_copy_layer(tmp_path / "other_ds" / "color")
+    assert len(ds.layers) == 1
+    color_layer = ds.get_layer("color")
+    assert color_layer.get_bounding_box() == BoundingBox(
+        topleft=(10, 20, 30), size=(32, 64, 128)
+    )
+    assert color_layer.mags.keys() == original_color_layer.mags.keys()
+    assert len(color_layer.mags.keys()) >= 1
+    for mag in color_layer.mags.keys():
+        np.array_equal(
+            color_layer.get_mag(mag).read(), original_color_layer.get_mag(mag).read()
+        )
+        # Test if the copied layer contains actual data
+        assert np.max(color_layer.get_mag(mag).read()) > 0
+
+    with pytest.raises(IndexError):
+        # The dataset already has a layer called "color".
+        ds.add_copy_layer(tmp_path / "other_ds" / "color")
+
+    # Test if the changes of the properties are persisted on disk by opening it again
+    assert "color" in Dataset(tmp_path / "ds").layers.keys()

wkcuber/api/dataset.py

@@ -1,4 +1,5 @@
 import operator
+import shutil
 from argparse import Namespace
 from shutil import rmtree
 from os import makedirs
@@ -442,6 +443,40 @@ def add_symlink_layer(self, foreign_layer_path: Union[str, Path]) -> Layer:
             self.get_layer(layer_name)._setup_mag(resolution.mag.to_layer_name())
         return self.layers[layer_name]
 
+    def add_copy_layer(self, foreign_layer_path: Union[str, Path]) -> Layer:
+        """
+        Copies the data at `foreign_layer_path` which belongs to another dataset to the current dataset.
+        Additionally, the relevant information from the `datasource-properties.json` of the other dataset are copied too.
+        """
+
+        foreign_layer_path = Path(os.path.abspath(foreign_layer_path))
+        layer_name = foreign_layer_path.name
+        if layer_name in self.layers.keys():
+            raise IndexError(
+                f"Cannot copy {foreign_layer_path}. This dataset already has a layer called {layer_name}."
+            )
+
+        shutil.copytree(foreign_layer_path, join(self.path, layer_name))
+
+        # copy the properties of the layer into the properties of this dataset
+        layer_properties = Dataset(foreign_layer_path.parent).properties.data_layers[
+            layer_name
+        ]
+        self.properties.data_layers[layer_name] = layer_properties
+        self.properties._export_as_json()
+
+        self._layers[layer_name] = self._create_layer(
+            layer_name,
+            _dtype_per_layer_to_dtype_per_channel(
+                layer_properties.element_class, layer_properties.num_channels
+            ),
+            layer_properties.num_channels,
+            layer_properties.category,
+        )
+        for resolution in layer_properties.wkw_magnifications:
+            self.get_layer(layer_name)._setup_mag(resolution.mag.to_layer_name())
+        return self.layers[layer_name]
+
     def copy_dataset(
         self,
         new_dataset_path: Union[str, Path],