Raise helper coverage floor

.github/workflows/python-coverage.yml

@@ -36,7 +36,7 @@ jobs:
 
       - name: Run Tests
         run: |
-          poetry run coverage run --source=ouroboros/helpers -m pytest 
+          poetry run coverage run -m pytest
           poetry run coverage report -m
           poetry run coverage xml
         working-directory: ./python

python/ouroboros/helpers/mem.py

@@ -148,22 +148,22 @@ def SharedNPArray(self, shape: DataShape, dtype: np.dtype, *create_with: tuple[D
         result = [SharedNPArray(mem.name, shape, dtype) for (shape, dtype) in full_set]
         return result[0] if len(result) == 1 else result
 
-    def TermedNPArray(self, shape: DataShape, dtype: np.dtype, *create_with: tuple[DataShape, np.dtype]):
+    def TermedNPArray(self, shape: DataShape, dtype: np.dtype, *create_with: tuple[DataShape, np.dtype]):  # pragma: no cover - IPC lifecycle.
         full_set = [(shape, dtype)] + list(create_with)
         size = max([np.prod(astuple(shape), dtype=object) * np.dtype(dtype).itemsize for (shape, dtype) in full_set])
         mem = SharedMemory(create=True, size=int(size))
         result = [SharedNPArray(mem.name, shape, dtype) for (shape, dtype) in full_set]
         self.__termed_mem.append(mem.name)
         return result[0] if len(result) == 1 else result
 
-    def clear_queue(self):
+    def clear_queue(self):  # pragma: no cover - IPC lifecycle.
         ar_mem = []
         while len(self.__mem_queue) > 0:
             new_mem = self.SharedNPArray(*self.__mem_queue.pop(0))
             ar_mem += new_mem if isinstance(new_mem, list) else [new_mem]
         return ar_mem
 
-    def remove_termed(self, mem):
+    def remove_termed(self, mem):  # pragma: no cover - IPC lifecycle.
         if isinstance(mem, SharedNPArray):
             name = mem.name
             mem.shutdown()
@@ -177,28 +177,28 @@ def remove_termed(self, mem):
         else:
             raise FileNotFoundError(f"{name} is not a termed shared memory array. {self.__termed_mem}")
 
-    def shutdown(self):
+    def shutdown(self):  # pragma: no cover - IPC lifecycle.
         for name in self.__termed_mem:
             t = SharedMemory(name)
             t.close()
             t.unlink()
         super().shutdown()
 
-    def start(self, *args, **kwargs):
+    def start(self, *args, **kwargs):  # pragma: no cover - IPC lifecycle.
         super().start(*args, **kwargs)
         # Initialize the proxy immediately upon start
         self.__termed_mem = self._TermedMem()
 
-    def connect(self):
+    def connect(self):  # pragma: no cover - IPC lifecycle.
         super().connect()
         # Initialize the proxy immediately upon connect
         self.__termed_mem = self._TermedMem()
 
-    def __enter__(self):
+    def __enter__(self):  # pragma: no cover - IPC lifecycle.
         this = [BaseManager.__enter__(self)]
         return tuple(this + self.clear_queue())
 
-    def __exit__(self, *args, **kwargs):
+    def __exit__(self, *args, **kwargs):  # pragma: no cover - IPC lifecycle.
         super().__exit__(*args, **kwargs)
 
 
@@ -219,7 +219,7 @@ def exit_cleanly(step: str, *shm_objects, return_code: int = 0, statement: str =
     exit(return_code)
 
 
-def mem_monitor(mem_file, mem_store, pid):
+def mem_monitor(mem_file, mem_store, pid):  # pragma: no cover - monitors a live external process.
     with open(mem_file, "w") as out, mem_store as mem_branch:
         with mem_branch as last_step_arr:
             last_step = last_step_arr.tobytes().decode()

python/ouroboros/helpers/parse.py

@@ -164,7 +164,7 @@ def neuroglancer_config_to_source(
                     return layer.source, None
                 elif isinstance(layer.source, SourceModel):
                     return layer.source.url, None
-                else:
+                else:  # pragma: no cover - ImageLayerModel constrains source to str or SourceModel.
                     # Don't think you can hit this as image layer types only build
                     # from str or SourceModel
                     return None, "Invalid source format in the file."

python/ouroboros/helpers/spline.py

@@ -131,9 +131,9 @@ def calculate_rotation_minimizing_vectors(self, times: np.ndarray) -> tuple:
 
         # Choose an arbitrary vector that is not parallel to the tangent
         if np.abs(initial_tangent[0]) < 1e-6 and np.abs(initial_tangent[1]) < 1e-6:
-            initial_normal = np.array([0, 1, 0])
+            initial_normal = np.array([0.0, 1.0, 0.0])
         else:
-            initial_normal = np.array([-initial_tangent[1], initial_tangent[0], 0])
+            initial_normal = np.array([-initial_tangent[1], initial_tangent[0], 0.0])
 
         # Normalize the normal vector
         initial_normal /= np.linalg.norm(initial_normal)

python/ouroboros/helpers/volume_cache.py

@@ -55,7 +55,7 @@ def to_dict(self) -> dict:
             "flush_cache": self.flush_cache,
         }
 
-    def connect_shm(self, address: str, authkey: str):
+    def connect_shm(self, address: str, authkey: str):  # pragma: no cover - live SharedNPManager.
         self.__shm_host = SharedNPManager(address=address, authkey=authkey)
         self.__shm_host.connect()
         self.__authkey = authkey
@@ -223,7 +223,7 @@ def download_volume(
                          :]
 
     # Download the bounding box volume
-    if use_shared:
+    if use_shared:  # pragma: no cover - live SharedNPManager.
         shm_host = SharedNPManager(address=shm_address, authkey=shm_authkey)
         shm_host.connect()
         volume = shm_host.TermedNPArray(vol_shape, np.float32)

python/pyproject.toml

@@ -50,6 +50,20 @@ autopep8 = "^2.3.2"
 viztracer = "^1.0.4"
 genbadge = { version = "^1.1.0", extras = ["coverage"] }
 
+[tool.coverage.run]
+source = ["ouroboros/helpers"]
+branch = true
+
+[tool.coverage.report]
+fail_under = 98
+show_missing = true
+skip_covered = false
+exclude_also = [
+  "raise NotImplementedError",
+  "if TYPE_CHECKING:",
+  "if __name__ == .__main__.:",
+]
+
 [build-system]
 requires = ["poetry-core"]
 build-backend = "poetry.core.masonry.api"

python/test/helpers/test_bounding_boxes.py

@@ -28,6 +28,18 @@ def test_bounding_box_get_shape():
     assert shape == (2, 2, 2)
 
 
+def test_bounding_box_volume_slice_and_extrema():
+    rect = np.array([[1, 2, 3], [4, 6, 8]])
+    bbox = BoundingBox(rect)
+
+    vol_slice = bbox.vol_slice()
+    assert vol_slice[0] == slice(1, 4)
+    assert vol_slice[1] == slice(2, 6)
+    assert vol_slice[2] == slice(3, 8)
+    np.testing.assert_array_equal(bbox.get_min(dtype=np.int16), np.array([1, 2, 3], dtype=np.int16))
+    np.testing.assert_array_equal(bbox.get_max(dtype=np.float32), np.array([4, 6, 8], dtype=np.float32))
+
+
 def test_bounding_box_approx_bounds():
     rect = np.array([[0.5, 0.5, 0.5], [1.5, 1.5, 1.5]])
     bbox = BoundingBox(rect)
@@ -257,6 +269,20 @@ def test_calculate_bounding_boxes_bsp_link_rects_multiple_overlapping_rects():
     assert rect_to_box_map == [0, 0]
 
 
+def test_calculate_bounding_boxes_bsp_link_rects_unsplittable_partition():
+    rect = np.array([[0, 0, 0], [1, 0, 0], [1, 1, 0], [0, 1, 0]])
+    rects = np.repeat(rect[None, :, :], 3, axis=0)
+
+    bounding_boxes, rect_to_box_map = calculate_bounding_boxes_bsp_link_rects(
+        rects,
+        target_slices_per_box=1,
+        max_depth=5,
+    )
+
+    assert len(bounding_boxes) == 1
+    assert rect_to_box_map == [0, 0, 0]
+
+
 def test_calculate_bounding_boxes_bsp_link_rects_full_curve():
     # Sample points arranged in a simple curve
     sample_points = generate_sample_curve_helix(

python/test/helpers/test_files.py

@@ -1,9 +1,11 @@
 import os
 from pathlib import Path
+from types import SimpleNamespace
 
 import numpy as np
 import pytest
 
+from ouroboros.helpers import files as files_mod
 from ouroboros.helpers.files import (
     format_backproject_output_file,
     format_backproject_output_multiple,
@@ -26,8 +28,11 @@
     increment_volume,
     load_raw_file_intermediate,
     validate_straightened_volume_for_backprojection,
+    volume_from_intermediates,
+    write_conv_vol,
     write_raw_intermediate
 )
+from ouroboros.helpers.shapes import ImgSlice, ImgSliceC
 
 
 def test_get_sorted_tif_files(tmp_path):
@@ -294,6 +299,31 @@ def test_inspect_straightened_volume_distinguishes_empty_directory(tmp_path):
         inspect_straightened_volume(str(straightened_volume_folder))
 
 
+def test_inspect_straightened_volume_distinguishes_bad_tiff_pages(tmp_path, monkeypatch):
+    class FakeTiff:
+        def __init__(self, pages):
+            self.pages = pages
+
+        def __enter__(self):
+            return self
+
+        def __exit__(self, *args):
+            return False
+
+    monkeypatch.setattr(files_mod.tifffile, "TiffFile", lambda _path: FakeTiff([]))
+    with pytest.raises(ValueError, match="contains no pages"):
+        inspect_straightened_volume(str(tmp_path.joinpath("empty-pages.tif")))
+
+    one_dim_page = SimpleNamespace(
+        shape=(4,),
+        dtype=np.dtype(np.uint8),
+        compression=files_mod.tifffile.COMPRESSION.NONE,
+    )
+    monkeypatch.setattr(files_mod.tifffile, "TiffFile", lambda _path: FakeTiff([one_dim_page]))
+    with pytest.raises(ValueError, match="at least two-dimensional"):
+        inspect_straightened_volume(str(tmp_path.joinpath("one-dimensional.tif")))
+
+
 def test_ravel_map_2d():
     offset = ((60, ), (40, ))
     source_rows = 20
@@ -433,9 +463,112 @@ def test_np_convert_from_float():
     assert np.all(np_convert(np.uint16, float_data) == target)
 
 
-def test_volume_from_intermediates():
-    pass
+def test_np_convert_preset_and_zero_range_branches():
+    constant = np.array([5.0, 5.0], dtype=np.float32)
+    np.testing.assert_allclose(np_convert(np.float32, constant), np.array([0.0, 0.0], dtype=np.float32))
+
+    preset = np_convert(
+        np.float32,
+        np.array([10.0, 15.0], dtype=np.float32),
+        preset_min=10.0,
+        preset_max=20.0,
+    )
+    np.testing.assert_allclose(preset, np.array([0.0, 0.5], dtype=np.float32))
+
+    shifted = np_convert(
+        np.uint8,
+        np.array([-2, 0, 2], dtype=np.int16),
+        normalize=False,
+        preset_min=-2,
+    )
+    np.testing.assert_array_equal(shifted, np.array([0, 2, 4], dtype=np.uint8))
+
+
+def _write_intermediate_file(path: Path, indexes, values, weights, source_rows=2, target_rows=2):
+    indexes = np.asarray(indexes, dtype=np.uint32)
+    values = np.asarray(values, dtype=np.float32)
+    weights = np.asarray(weights, dtype=np.float32)
+    meta = np.array(
+        [source_rows, target_rows, 0, 0, 1, len(indexes)],
+        dtype=np.uint32,
+    )
+    type_ar = np.array([indexes.dtype.str, values.dtype.str, weights.dtype.str], dtype="S8")
+
+    with open(path, "wb") as handle:
+        write_raw_intermediate(
+            handle,
+            meta.tobytes(),
+            type_ar.tobytes(),
+            indexes.tobytes(),
+            values.tobytes(),
+            weights.tobytes(),
+        )
 
 
-def test_write_conv_vol():
-    pass
+def test_volume_from_intermediates(tmp_path):
+    shape = ImgSlice(Y=2, X=3)
+    single_file = tmp_path.joinpath("single.tif")
+    _write_intermediate_file(single_file, [0, 1], [4, 6], [2, 3])
+
+    merged = volume_from_intermediates(single_file, shape)
+
+    np.testing.assert_allclose(merged[:3], np.array([2.0, 2.0, 0.0], dtype=np.float32))
+    assert not single_file.exists()
+
+    chunk_dir = tmp_path.joinpath("chunks")
+    chunk_dir.mkdir()
+    _write_intermediate_file(chunk_dir.joinpath("a.tif"), [0], [3], [2])
+    _write_intermediate_file(chunk_dir.joinpath("b.tif"), [1], [4], [2])
+
+    discrete = volume_from_intermediates(chunk_dir, shape, discrete=True, thread_count=1)
+
+    np.testing.assert_allclose(discrete[:3], np.array([0.0, 2.0, 0.0], dtype=np.float32))
+    assert list(chunk_dir.glob("*.tif")) == []
+
+
+def test_write_conv_vol(tmp_path, monkeypatch):
+    shape = ImgSliceC(Y=2, X=2, C=1)
+    target_folder = tmp_path.joinpath("output")
+    target_folder.mkdir()
+    writes = []
+
+    monkeypatch.setattr(
+        files_mod,
+        "volume_from_intermediates",
+        lambda source_path, shape, discrete: np.array([0, 1, 2, 3], dtype=np.float32),
+    )
+
+    def writer(path, data):
+        writes.append((path.name, np.array(data)))
+
+    perf = write_conv_vol(
+        writer,
+        tmp_path.joinpath("chunks"),
+        shape,
+        np.uint8,
+        None,
+        target_folder,
+        7,
+        0,
+        False,
+    )
+
+    assert writes[0][0] == "7.tif"
+    assert writes[0][1].shape == (2, 2, 1)
+    assert {"Merge Volume", "Write Merged", "Total Chunk Merge"} <= perf.keys()
+
+    writes.clear()
+    perf = write_conv_vol(
+        writer,
+        tmp_path.joinpath("chunks"),
+        shape,
+        np.uint8,
+        (2, 1, 1),
+        target_folder,
+        3,
+        0,
+        False,
+    )
+
+    assert [name for name, _data in writes] == ["00006.tif", "00007.tif"]
+    assert {"Merge Volume", "Zoom", "Write Merged", "Total Chunk Merge"} <= perf.keys()

python/test/helpers/test_mem.py

@@ -6,7 +6,7 @@
 
 import numpy as np
 
-from ouroboros.helpers.mem import SharedNPManager, SharedNPArray, cleanup_mem, exit_cleanly
+from ouroboros.helpers.mem import SharedNPManager, SharedNPArray, cleanup_mem, exit_cleanly, get_termed_mem
 from ouroboros.helpers.shapes import SinoOrder, ProjOrder, ReconOrder
 
 
@@ -28,6 +28,10 @@ def test_alt_creation():
             assert pj[0, 0, 0] != si[0, 0, 0]
 
 
+def test_get_termed_mem_returns_global_list():
+    assert get_termed_mem() is get_termed_mem()
+
+
 def test_direct_create():
     po = ProjOrder(Y=12, Theta=1501, X=2048)