add ability to change sync peak finding algo

Author: darsnack

packages/cheese3d/cheese3d/synchronize/core.py

@@ -3,7 +3,7 @@
 import numpy as np
 import matplotlib.pyplot as plt
 from dataclasses import dataclass
-from typing import List, Tuple, Dict, Any
+from typing import List, Tuple, Dict, Any, Literal
 from pathlib import Path
 
 from cheese3d.synchronize.aligners import (BaseAligner,
@@ -19,6 +19,7 @@
 class SyncConfig:
     pipeline: List[str]
     led_threshold: float = 0.9
+    led_peak_algorithm: Literal["dynamic", "max"] = "dynamic"
     max_regression_rmse: float = 1e-2
     ref_view: str = "bottomcenter"
     ref_crop: str = "default"
@@ -234,12 +235,14 @@ def synchronize_videos(pipeline_cfg: SyncConfig,
     ref_reader = VideoSyncReader(source=ref_video,
                                  sample_rate=fps,
                                  threshold=pipeline_cfg.led_threshold,
+                                 peak_algorithm=pipeline_cfg.led_peak_algorithm,
                                  crop=ref_crop)
     align_params = {"ref": (ref_video, 0, fps)}
     for view, (video, crop) in videos.items():
         target_reader = VideoSyncReader(source=video,
                                         sample_rate=fps,
                                         threshold=pipeline_cfg.led_threshold,
+                                        peak_algorithm=pipeline_cfg.led_peak_algorithm,
                                         crop=crop)
         pipeline = SyncPipeline.from_cfg(pipeline_cfg, ref_reader, target_reader)
         target_path = pipeline.target.root_path()
@@ -267,6 +270,7 @@ def synchronize_ephys(pipeline_cfg: SyncConfig,
     video_reader = VideoSyncReader(source=ref_video,
                                    sample_rate=fps,
                                    threshold=pipeline_cfg.led_threshold,
+                                   peak_algorithm=pipeline_cfg.led_peak_algorithm,
                                    crop=ref_crop)
     ephys_reader = get_ephys_reader(ephys_path, ephys_params)
     pipeline = SyncPipeline.from_cfg(pipeline_cfg, video_reader, ephys_reader)

packages/cheese3d/cheese3d/synchronize/readers.py

@@ -6,7 +6,7 @@
 import matplotlib.patches as patches
 from dataclasses import dataclass, field
 from pathlib import Path
-from typing import Optional, Dict, Any
+from typing import Optional, Dict, Any, Literal
 from tqdm import tqdm
 from open_ephys.analysis import Session as OESession
 
@@ -50,6 +50,7 @@ class VideoSyncReader(SyncSignalReader):
     - `crop`: Tuple of (left, right, top, bottom) coordinates for cropping.
     """
     crop: BoundingBox = field(default_factory=lambda: [None, None, None, None])
+    peak_algorithm: Literal["dynamic", "max"] = "dynamic"
 
     def load_signal(self):
         print(self.source)
@@ -65,7 +66,7 @@ def load_signal(self):
         min_brightness = bin_edges[np.argmax(hist) + 1]
         brightness = brightness - min_brightness
         nz_brightness = brightness[brightness > 2]
-        if len(nz_brightness) > 0:
+        if (self.peak_algorithm == "dynamic") and (len(nz_brightness) > 0):
             mid = np.percentile(nz_brightness, 75)
             # q3 = np.percentile(nz_brightness, 75)
             # iqr = q3 - q1
@@ -87,6 +88,7 @@ def load_signal(self):
 
         fig, ax = plt.subplots(figsize=(8, 6))
         ax.plot(brightness)
+        ax.axhline(peak_brightness, color='k', linestyle='--')
         ax.axhline(led_threshold, color='r', linestyle='--')
         ax.set_title("LED BBox Brightness")
         fig.savefig(f"{video.path.rstrip('.avi')}-qc-brightness.png", bbox_inches="tight")