Adds deinterleave

src/ci_pipe/modules/isx_module.py

@@ -4,11 +4,13 @@
 
 from ci_pipe.decorators import step
 from ci_pipe.errors.isx_backend_not_configured_error import ISXBackendNotConfiguredError
+from ci_pipe.utils.isx_utils import get_efocus
 from ci_pipe.utils.project_template import load_project_templates
 
 
 class ISXModule:
     # TODO: Consider moving these constants to a different config file
+    DEINTERLEAVE_VIDEOS_STEP = "ISX Deinterleave Videos"
     PREPROCESS_VIDEOS_STEP = "ISX Preprocess Videos"
     BANDPASS_FILTER_VIDEOS_STEP = "ISX Bandpass Filter Videos"
     MOTION_CORRECTION_VIDEOS_STEP = "ISX Motion Correction Videos"
@@ -32,6 +34,7 @@ class ISXModule:
     LONGITUDINAL_REGISTRATION_CORRESPONDENCES_TABLE_NAME = "LR-correspondences-table"
     LONGITUDINAL_REGISTRATION_CROP_RECT_NAME = "LR-crop-rect"
     LONGITUDINAL_REGISTRATION_TRANSFORM_NAME = "LR-transform"
+    DEINTERLEAVE_VIDEOS_SUFFIX = "efocus"
     GUI_VISUALIZATION_STEP = "ISX Gui Visualization"
     GUI_VISUALIZATION_SUFFIX = "GUI"
 
@@ -47,6 +50,52 @@ def __init__(self, isx, ci_pipe):
 
     # TODO: Find the best way to remove repetition on these step methods, without losing clarity of what each step does
 
+    @step(DEINTERLEAVE_VIDEOS_STEP)
+    def deinterleave_videos(self, inputs):
+        """
+        De-interleave multiplane movies into one video per focal plane.
+
+        For each input video, resolves e-focus from GPIO or acquisition metadata
+        (checking raw files like .gpio alongside the movie). Produces one output
+        movie per plane; each output keeps the same ids and adds an 'efocus' key.
+        """
+        output = []
+        output_dir = self._ci_pipe.create_output_directory_for_next_step(
+            self.DEINTERLEAVE_VIDEOS_STEP
+        )
+
+        for input in inputs('videos-isxd'):
+            input_path = input['value']
+            ids = input['ids']
+
+            video = self._isx.Movie.read(str(input_path))
+            efocus_list = get_efocus(
+                self._isx, input_path, output_dir, video
+            )
+            del video
+
+            output_paths = [
+                self._isx.make_output_file_path(
+                    input_path,
+                    output_dir,
+                    f"{self.DEINTERLEAVE_VIDEOS_SUFFIX}-{ef}",
+                )
+                for ef in efocus_list
+            ]
+
+            self._isx.de_interleave(
+                input_movie_files=[input_path],
+                output_movie_files=output_paths,
+                in_efocus_values=efocus_list,
+            )
+
+            for ef, path in zip(efocus_list, output_paths):
+                output.append({'ids': ids, 'efocus': ef, 'value': path})
+
+        return {
+            'videos-isxd': output
+        }
+
     @step(PREPROCESS_VIDEOS_STEP)
     def preprocess_videos(
             self,
@@ -77,7 +126,10 @@ def preprocess_videos(
                 trim_early_frames=isx_pp_trim_early_frames
             )
 
-            output.append({'ids': input['ids'], 'value': output_path})
+            item = {'ids': input['ids'], 'value': output_path}
+            if 'efocus' in input:
+                item['efocus'] = input['efocus']
+            output.append(item)
 
         return {
             'videos-isxd': output
@@ -109,7 +161,10 @@ def bandpass_filter_videos(
                 subtract_global_minimum=isx_bp_subtract_global_minimum
             )
 
-            output.append({'ids': input['ids'], 'value': output_path})
+            item = {'ids': input['ids'], 'value': output_path}
+            if 'efocus' in input:
+                item['efocus'] = input['efocus']
+            output.append(item)
 
         return {
             'videos-isxd': output
@@ -170,7 +225,10 @@ def motion_correction_videos(
                 preserve_input_dimensions=isx_mc_preserve_input_dimensions
             )
 
-            output_videos.append({'ids': input['ids'], 'value': output_video_path})
+            item = {'ids': input['ids'], 'value': output_video_path}
+            if 'efocus' in input:
+                item['efocus'] = input['efocus']
+            output_videos.append(item)
             output_translations.append({'ids': input['ids'], 'value': output_translations_path})
             output_crop_rects.append({'ids': input['ids'], 'value': output_crop_rect_path})
             output_mean_images.append({'ids': input['ids'], 'value': output_mean_image_path})
@@ -202,7 +260,10 @@ def normalize_dff_videos(
                 f0_type=isx_dff_f0_type
             )
 
-            output.append({'ids': input['ids'], 'value': output_path})
+            item = {'ids': input['ids'], 'value': output_path}
+            if 'efocus' in input:
+                item['efocus'] = input['efocus']
+            output.append(item)
 
         return {
             'videos-isxd': output

src/ci_pipe/plotter.py

@@ -67,11 +67,16 @@ def get_all_trace_from_branch(self, trace, branch_name):
         self.console.print(f"\n[bold underline]Pipeline Trace of branch: {branch_name}[/bold underline]\n")
         self.console.print(*panels, justify="center")
 
+    def _format_params(self, params):
+        if not params:
+            return "None"
+        return "\n".join(f"  {k}: {v}" for k, v in params.items())
+
     def _build_trace_panels(self, steps):
         panels = [
             Panel(
                 f"[bold]{index}.[/bold] {step.name()}\n"
-                f"Params: {', '.join(step.arguments()) or 'None'}",
+                f"Params:\n{self._format_params(step.arguments())}",
                 padding=(1, 2),
             )
             for index, step in enumerate(steps, start=1)

src/ci_pipe/utils/isx_utils.py

@@ -0,0 +1,91 @@
+"""
+ISX-specific helpers for the pipeline: metadata extraction, GPIO resolution, etc.
+
+Used by ISX steps (e.g. e-focus / multiplane handling). May read or write
+intermediate files (e.g. copying GPIO into the output folder).
+"""
+
+import os
+import shutil
+from pathlib import Path
+import numpy as np
+import warnings
+
+def get_efocus(isx, file: str, outputfolder: str, video) -> list:
+    """
+    Resolve e-focus values for a given ISX file, from GPIO or acquisition metadata.
+
+    Looks for existing converted/copied GPIO in the output folder, then the
+    converted file next to the input, then the raw GPIO (copying it into the
+    output folder if needed). Falls back to acquisition info when no GPIO is
+    available.
+
+    Parameters
+    ----------
+    isx
+        ISX backend (same as passed to ISXModule), providing GpioSet, verify_deinterleave, etc.
+    file : str
+        Path to the ISX file (movie).
+    outputfolder : str
+        Output directory where intermediate GPIO may be written.
+    video
+        Opened ISX movie (or object with get_acquisition_info()) for metadata fallback.
+
+    Returns
+    -------
+    list
+        List of e-focus values (e.g. [0] or [0, 1, 2] for multiplane).
+    """
+    raw_gpio_file = os.path.splitext(file)[0] + ".gpio"
+    updated_gpio_file = os.path.splitext(file)[0] + "_gpio.isxd"
+    local_updated_gpio_file = os.path.join(outputfolder, Path(updated_gpio_file).name)
+
+    if os.path.exists(local_updated_gpio_file):
+        return get_efocus_from_gpio(isx, local_updated_gpio_file)
+    if os.path.exists(updated_gpio_file):
+        return get_efocus_from_gpio(isx, updated_gpio_file)
+    if os.path.exists(raw_gpio_file):
+        local_raw_gpio_file = os.path.join(outputfolder, Path(raw_gpio_file).name)
+        shutil.copy2(raw_gpio_file, local_raw_gpio_file)
+        return get_efocus_from_gpio(isx, local_raw_gpio_file)
+
+    acquisition_info = video.get_acquisition_info().copy()
+    if "Microscope Focus" in acquisition_info:
+        if not isx.verify_deinterleave(file, acquisition_info["Microscope Focus"]):
+            warnings.warn(
+                f"Info {file}: Multiple Microscope Focus but no GPIO file",
+                UserWarning,
+                stacklevel=2,
+            )
+            return [0]
+        return [acquisition_info["Microscope Focus"]]
+
+    print(f"Info: Unable to verify Microscope Focus config in: {file}")
+    return [0]
+
+
+def get_efocus_from_gpio(isx, gpio_file: str) -> list:
+    """
+    Read e-focus channel from a GPIO file and return the video e-focus values.
+
+    Parameters
+    ----------
+    isx
+        ISX backend (same as passed to ISXModule), providing GpioSet.
+    gpio_file : str
+        Path to the GPIO file (.gpio or .isxd).
+
+    Returns
+    -------
+    list
+        List of integer e-focus values (one per plane) with at least
+        min_frames_per_efocus frames.
+    """
+    gpio_set = isx.GpioSet.read(gpio_file)
+    efocus_values = gpio_set.get_channel_data(gpio_set.channel_dict["e-focus"])[1]
+    efocus_values, efocus_counts = np.unique(efocus_values, return_counts=True)
+    min_frames_per_efocus = 100
+    video_efocus = efocus_values[efocus_counts >= min_frames_per_efocus]
+    if video_efocus.shape[0] >= 4:
+        warnings.warn("Too many efocus detected, early frames issue.")
+    return [int(v) for v in video_efocus]

src/external_dependencies/isx/in_memory_isx.py

@@ -2,6 +2,21 @@ class InMemoryISX:
     def __init__(self, file_system=None):
         self._file_system = file_system
 
+    def deinterleave(
+            self,
+            input_movie_files,
+            output_movie_files,
+            in_efocus_values,
+    ):
+        """
+        De-interleave multiplane movies into one file per plane.
+
+        output_movie_files length must be input_movie_files * len(in_efocus_values).
+        Order: [in_1_plane0, in_1_plane1, ..., in_2_plane0, ...].
+        """
+        for output_file in output_movie_files:
+            self._file_system.write(output_file, "")
+
     def preprocess(
             self,
             input_movie_files,
@@ -206,4 +221,7 @@ def get_frame_data(self, index):
                     [2.0, 3.0],
                 ]
 
+            def get_acquisition_info(self):
+                return {}
+
         return Movie