Refactor common functionality out of multiple parts of the python pac…

…kage.

python/ouroboros/common/file_system.py

@@ -6,7 +6,14 @@
     copy_to_host,
     copy_to_volume,
 )
-from ouroboros.helpers.files import combine_unknown_folder
+from ouroboros.helpers.files import (
+    combine_unknown_folder,
+    format_backproject_output_file,
+    format_backproject_output_multiple,
+    format_slice_output_config_file,
+    format_slice_output_file,
+    format_slice_output_multiple,
+)
 from ouroboros.helpers.options import BackprojectOptions, SliceOptions
 
 
@@ -83,11 +90,12 @@ def load_options_for_backproject_docker(
     # Define the output file paths
     host_output_folder = options.output_file_folder
     host_output_file = combine_unknown_folder(
-        host_output_folder, options.output_file_name + "-backprojected.tif"
+        host_output_folder, format_backproject_output_file(options.output_file_name)
     )
     host_output_slices = (
         combine_unknown_folder(
-            host_output_folder, options.output_file_name + "-backprojected"
+            host_output_folder,
+            format_backproject_output_multiple(options.output_file_name),
         )
         if options.make_single_file is False
         else None
@@ -210,17 +218,19 @@ def load_options_for_slice_docker(
 
     host_output_folder = slice_options.output_file_folder
     host_output_file = combine_unknown_folder(
-        host_output_folder, slice_options.output_file_name + ".tif"
+        host_output_folder, format_slice_output_file(slice_options.output_file_name)
     )
     host_output_slices = (
         combine_unknown_folder(
-            host_output_folder, slice_options.output_file_name + "-slices"
+            host_output_folder,
+            format_slice_output_multiple(slice_options.output_file_name),
         )
         if slice_options.make_single_file is False
         else None
     )
     host_output_config_file = combine_unknown_folder(
-        host_output_folder, slice_options.output_file_name + "-configuration.json"
+        host_output_folder,
+        format_slice_output_config_file(slice_options.output_file_name),
     )
 
     # Modify the output file folder to be in the docker volume

python/ouroboros/helpers/files.py

@@ -102,6 +102,7 @@ def get_sorted_tif_files(directory):
 def join_path(*args):
     return str(Path(*args))
 
+
 def combine_unknown_folder(directory_path: str, filename: str) -> str:
     """
     Combine a directory path and a filename into a single path.
@@ -128,4 +129,32 @@ def combine_unknown_folder(directory_path: str, filename: str) -> str:
         else:
             directory_path += "\\"
 
-    return directory_path + filename
+    return directory_path + filename
+
+
+def format_slice_output_file(output_name: str):
+    return output_name + ".tif"
+
+
+def format_slice_output_multiple(output_name: str):
+    return output_name + "-slices"
+
+
+def format_slice_output_config_file(output_name: str):
+    return output_name + "-configuration.json"
+
+
+def format_backproject_output_file(output_name: str):
+    return output_name + "-backprojected.tif"
+
+
+def format_backproject_output_multiple(output_name: str):
+    return output_name + "-backprojected"
+
+
+def format_backproject_tempvolumes(output_name: str):
+    return output_name + "-tempvolumes"
+
+
+def format_backproject_resave_volume(output_name: str):
+    return output_name + "-temp-straightened.tif"

python/ouroboros/helpers/slice.py

@@ -9,6 +9,9 @@
 
 INDEXING = "xy"
 
+NO_COLOR_CHANNELS_DIMENSIONS = 3
+COLOR_CHANNELS_DIMENSIONS = NO_COLOR_CHANNELS_DIMENSIONS + 1
+
 
 def calculate_slice_rects(
     times: np.ndarray, spline: Spline, width, height, spline_points=None
@@ -143,14 +146,14 @@ def slice_volume_from_grids(
     normalized_grid = normalized_grid.reshape(-1, 3).T
 
     # Check if volume has color channels
-    has_color_channels = volume.ndim == 4
+    has_color_channels, num_channels = detect_color_channels(volume)
 
     if has_color_channels:
         # Initialize an empty list to store slices from each channel
         channel_slices = []
 
         # Iterate over each color channel
-        for channel in range(volume.shape[-1]):
+        for channel in range(num_channels):
             # Extract the current channel
             current_channel = volume[..., channel]
 
@@ -202,9 +205,7 @@ def write_slices_to_volume(
     weights = slice_coords - coords_floor
 
     # Check if volume has color channels
-    has_color_channels = volume.ndim == 4
-
-    num_channels = volume.shape[-1] if has_color_channels else 1
+    has_color_channels, num_channels = detect_color_channels(volume)
 
     for channel in range(num_channels):
         # Initialize the accumulation arrays
@@ -269,3 +270,25 @@ def make_volume_binary(volume: np.ndarray, dtype=np.uint8) -> np.ndarray:
     """
 
     return (volume > 0).astype(dtype)
+
+
+def detect_color_channels(data: np.ndarray, none_value=1):
+    """
+    Detect the number of color channels in a volume.
+
+    Parameters:
+    ----------
+        data (numpy.ndarray): The volume data.
+        none_value (int): The value to return if the volume has no color channels.
+
+    Returns:
+    -------
+        tuple: A tuple containing the following:
+            - has_color_channels (bool): Whether the volume has color channels.
+            - num_color_channels (int): The number of color channels in the volume.
+    """
+
+    has_color_channels = data.ndim == COLOR_CHANNELS_DIMENSIONS
+    num_color_channels = data.shape[-1] if has_color_channels else none_value
+
+    return has_color_channels, num_color_channels

python/ouroboros/pipeline/backproject_pipeline.py

@@ -1,5 +1,6 @@
-from ouroboros.helpers.memory_usage import calculate_gigabytes_from_dimensions
+from ouroboros.helpers.memory_usage import GIGABYTE, calculate_gigabytes_from_dimensions
 from ouroboros.helpers.slice import (
+    detect_color_channels,
     generate_coordinate_grid_for_rect,
     make_volume_binary,
     write_slices_to_volume,
@@ -9,6 +10,9 @@
 from .pipeline import PipelineStep
 from ouroboros.helpers.options import BackprojectOptions
 from ouroboros.helpers.files import (
+    format_backproject_output_multiple,
+    format_backproject_resave_volume,
+    format_backproject_tempvolumes,
     get_sorted_tif_files,
     join_path,
     load_and_save_tiff_from_slices,
@@ -75,7 +79,7 @@ def _process(self, input_data: any) -> tuple[any, None] | tuple[None, any]:
             # Create a new path for the straightened volume
             new_straightened_volume_path = join_path(
                 config.output_file_folder,
-                f"{config.output_file_name}-temp-straightened.tif",
+                format_backproject_resave_volume(config.output_file_name),
             )
 
             # Save the straightened volume to a new tif file
@@ -106,7 +110,8 @@ def _process(self, input_data: any) -> tuple[any, None] | tuple[None, any]:
 
         # Create a folder to hold the temporary volume files
         temp_folder_path = join_path(
-            config.output_file_folder, f"{config.output_file_name}-tempvolumes"
+            config.output_file_folder,
+            format_backproject_tempvolumes(config.output_file_name),
         )
         os.makedirs(temp_folder_path, exist_ok=True)
 
@@ -161,7 +166,7 @@ def _process(self, input_data: any) -> tuple[any, None] | tuple[None, any]:
             DEFAULT_CHUNK_SIZE
             if config.max_ram_gb == 0
             else int(
-                (config.max_ram_gb * 1024**3)
+                (config.max_ram_gb * GIGABYTE)
                 / calculate_chunk_size(config, volume_cache)
             )
         )
@@ -182,7 +187,8 @@ def _process(self, input_data: any) -> tuple[any, None] | tuple[None, any]:
 
         # Save the backprojected volume to a series of tif files
         folder_path = join_path(
-            config.output_file_folder, f"{config.output_file_name}-backprojected"
+            config.output_file_folder,
+            format_backproject_output_multiple(config.output_file_name),
         )
         if os.path.exists(folder_path):
             shutil.rmtree(folder_path)
@@ -197,12 +203,10 @@ def _process(self, input_data: any) -> tuple[any, None] | tuple[None, any]:
 
         # Determine the number of channels in the straightened volume
         temp_straightened_volume = make_tiff_memmap(straightened_volume_path, mode="r")
-        num_channels = (
-            None
-            if temp_straightened_volume.ndim == 3
-            else temp_straightened_volume.shape[-1]
+        has_multiple_channels, num_channels = detect_color_channels(
+            temp_straightened_volume, none_value=None
         )
-        has_multiple_channels = num_channels is not None
+        del temp_straightened_volume
 
         # Write the chunks to tif files
         for i, (chunk_bounding_box, bounding_boxes) in enumerate(chunks_and_boxes):
@@ -310,7 +314,8 @@ def _process(self, input_data: any) -> tuple[any, None] | tuple[None, any]:
         # Delete the temporary volume files
         shutil.rmtree(
             join_path(
-                config.output_file_folder, config.output_file_name + "-tempvolumes"
+                config.output_file_folder,
+                format_backproject_tempvolumes(config.output_file_name),
             )
         )
 
@@ -365,9 +370,7 @@ def process_bounding_box(
     # Load the straightened volume
     start = time.perf_counter()
     straightened_volume = make_tiff_memmap(straightened_volume_path, mode="r")
-    num_channels = (
-        None if straightened_volume.ndim == 3 else straightened_volume.shape[-1]
-    )
+    _, num_channels = detect_color_channels(straightened_volume, none_value=None)
     slice_width, slice_height = (
         straightened_volume.shape[1],
         straightened_volume.shape[2],
@@ -406,7 +409,7 @@ def process_bounding_box(
     start = time.perf_counter()
     file_path = join_path(
         config.output_file_folder,
-        f"{config.output_file_name}-tempvolumes",
+        format_backproject_tempvolumes(config.output_file_name),
         f"{index}.tif",
     )
     tifffile.imwrite(file_path, volume, software="ouroboros")
@@ -442,7 +445,7 @@ def calculate_chunk_size(config, volume_cache, axis=0):
         volume_cache.get_volume_dtype(),
     )
 
-    return int((config.max_ram_gb * 1024**3) / bounding_box_memory_usage)
+    return int((config.max_ram_gb * GIGABYTE) / bounding_box_memory_usage)
 
 
 def create_volume_chunks(

python/ouroboros/pipeline/slice_parallel_pipeline.py

@@ -3,7 +3,11 @@
     slice_volume_from_grids,
 )
 from ouroboros.helpers.volume_cache import VolumeCache
-from ouroboros.helpers.files import join_path
+from ouroboros.helpers.files import (
+    format_slice_output_file,
+    format_slice_output_multiple,
+    join_path,
+)
 from .pipeline import PipelineStep
 from ouroboros.helpers.options import SliceOptions
 import numpy as np
@@ -52,11 +56,15 @@ def _process(self, input_data: tuple[any]) -> None | str:
 
         # Create a folder with the same name as the output file
         folder_name = join_path(
-            config.output_file_folder, f"{config.output_file_name}-slices"
+            config.output_file_folder,
+            format_slice_output_multiple(config.output_file_name),
         )
-        os.makedirs(folder_name, exist_ok=True)
+
+        if config.make_single_file:
+            os.makedirs(folder_name, exist_ok=True)
+
         output_file_path = join_path(
-            config.output_file_folder, config.output_file_name + ".tif"
+            config.output_file_folder, format_slice_output_file(config.output_file_name)
         )
 
         # Create an empty tiff to store the slices

python/test/helpers/test_files.py

@@ -1,6 +1,13 @@
 import os
 from tifffile import imwrite
 from ouroboros.helpers.files import (
+    format_backproject_output_file,
+    format_backproject_output_multiple,
+    format_backproject_resave_volume,
+    format_backproject_tempvolumes,
+    format_slice_output_config_file,
+    format_slice_output_file,
+    format_slice_output_multiple,
     load_and_save_tiff_from_slices,
     get_sorted_tif_files,
     join_path,
@@ -100,3 +107,38 @@ def test_combine_unknown_folder_win_slash():
     filename = "file.txt"
     combined = combine_unknown_folder(directory, filename)
     assert combined == "C:\\path\\to\\directory\\file.txt"
+
+
+def test_format_slice_output_file():
+    result = format_slice_output_file("test")
+    assert isinstance(result, str)
+
+
+def test_format_slice_output_multiple():
+    result = format_slice_output_multiple("test")
+    assert isinstance(result, str)
+
+
+def test_format_slice_output_config_file():
+    result = format_slice_output_config_file("test")
+    assert isinstance(result, str)
+
+
+def test_format_backproject_output_file():
+    result = format_backproject_output_file("test")
+    assert isinstance(result, str)
+
+
+def test_format_backproject_output_multiple():
+    result = format_backproject_output_multiple("test")
+    assert isinstance(result, str)
+
+
+def test_format_backproject_tempvolumes():
+    result = format_backproject_tempvolumes("test")
+    assert isinstance(result, str)
+
+
+def test_format_backproject_resave_volume():
+    result = format_backproject_resave_volume("test")
+    assert isinstance(result, str)

python/test/helpers/test_memory_usage.py

@@ -1,4 +1,5 @@
 from ouroboros.helpers.memory_usage import (
+    GIGABYTE,
     calculate_gigabytes_in_array,
     calculate_gigabytes_from_dimensions,
 )
@@ -8,7 +9,7 @@
 
 def test_calculate_gigabytes_in_array():
     # Create a numpy array with 1 GB size
-    array = np.zeros(int((1024**3) / np.dtype(np.float64).itemsize), dtype=np.float64)
+    array = np.zeros(int((GIGABYTE) / np.dtype(np.float64).itemsize), dtype=np.float64)
 
     # Calculate the expected result
     expected_result = 1.0