timestamp plots, towards synchronization (maybe not needed)

Cameras/exploration_scripts/file_explore.py → Cameras/dianostics/file_explore.py

@@ -5,8 +5,12 @@
 from pathlib import Path
 from datetime import datetime
 
+from skellycam_plots import create_timestamp_diagnostic_plots, timestamps_array_to_dictionary
+
 Z_SCORE_95_CI = 1.96
 
+# TODO: Make plots of timestamps for each camera
+
 def print_video_info(videos_path: Path):
     for video_path in videos_path.iterdir():
         if video_path.suffix not in {".avi", ".AVI", ".mp4", ".MP4"}:
@@ -31,26 +35,47 @@ def print_timestamp_info(timestamp_path: Path):
     print(f"shape of timestamps: {timestamps.shape}")
     starting_time = np.min(timestamps)
 
+    by_camera_fps = []
+    by_camera_frame_duration = []
+
     for i in range(timestamps.shape[0]):
         num_samples = timestamps.shape[1]
         samples = (timestamps[i, :] - starting_time) / 1e9
+        fps = num_samples / (samples[-1] - samples[0])
+        mean_frame_duration = np.mean(np.diff(timestamps[i, :])) / 1e6
+        by_camera_fps.append(fps)
+        by_camera_frame_duration.append(mean_frame_duration)
+        units = "seconds"
         print(f"cam {i} Descriptive Statistics:")
-        print(f"\tEarliest Timestamp: {np.min(samples):.3f}")
-        print(f"\tLatest Timestamp: {np.max(samples):.3f}")
+        print(f"\tEarliest Timestamp: {np.min(samples):.3f} {units}")
+        print(f"\tLatest Timestamp: {np.max(samples):.3f} {units}")
+        print(f"\tFPS: {fps}")
+        print(f"\tMean Frame Duration for Camera: {mean_frame_duration} ms")
 
-    for i in range(0, timestamps.shape[1], 15):
+    print("Overall FPS and Mean Frame Duration")
+    print(f"\tMean Overall FPS: {np.nanmean(by_camera_fps)}")
+    print(f"\tMean Overall Mean Frame Duration: {np.nanmean(by_camera_frame_duration)}")
+
+    for i in range(0, timestamps.shape[1]-1, 15):
         num_samples = timestamps.shape[0]
         samples = (timestamps[:, i] - starting_time) / 1e9
+        units = "seconds"
         print(f"frame {i} Descriptive Statistics")
-        print(f"\tNumber of Samples: {num_samples}")
-        print(f"\tMean: {np.nanmean(samples):.3f}")
-        print(f"\tMedian: {np.nanmedian(samples):.3f}")
-        print(f"\tStandard Deviation: {np.nanstd(samples):.3f}")
-        print(f"\tMedian Absolute Deviation: {np.nanmedian(np.abs(samples - np.nanmedian(samples))):.3f}")
-        print(f"\tInterquartile Range: {np.nanpercentile(samples, 75) - np.nanpercentile(samples, 25):.3f}")
-        print(f"\t95% Confidence Interval: {(Z_SCORE_95_CI * np.nanstd(samples) / (num_samples**0.5)):.3f}")
+        print(f"\tNumber of Samples: {num_samples} {units}")
+        print(f"\tMean: {np.nanmean(samples):.3f} {units}")
+        print(f"\tMedian: {np.nanmedian(samples):.3f} {units}")
+        print(f"\tStandard Deviation: {np.nanstd(samples):.3f} {units}")
+        print(f"\tMedian Absolute Deviation: {np.nanmedian(np.abs(samples - np.nanmedian(samples))):.3f} {units}")
+        print(f"\tInterquartile Range: {np.nanpercentile(samples, 75) - np.nanpercentile(samples, 25):.3f} {units}")
+        print(f"\t95% Confidence Interval: {(Z_SCORE_95_CI * np.nanstd(samples) / (num_samples**0.5)):.3f} {units}")
         print(f"\tEarliest Timestamp: {np.min(samples):.3f}")
-        print(f"\tLatest Timestamp: {np.max(samples):.3f}")  
+        print(f"\tLatest Timestamp: {np.max(samples):.3f}") 
+        print(f"\tMean Frame Duration for Camera: {np.nanmean(timestamps[:, i+1] - timestamps[:, i]) / 1e6} ms") 
+
+    create_timestamp_diagnostic_plots(
+        timestamp_dictionary=timestamps_array_to_dictionary(timestamps),
+        path_to_save_plots_png=timestamp_path.parent / "timestamp_diagnostic_plot.png"
+    )
 
 
 def get_ffprobe_fps(video_path: Path) -> float:
@@ -103,7 +128,7 @@ def get_ffprobe_fps(video_path: Path) -> float:
     raw_videos_path = folder_path / "raw_videos"
     synched_videos_path = folder_path / "synchronized_videos"
 
-    print_video_info(folder_path)
-    # print_video_info(raw_videos_path)
+    # print_video_info(folder_path)
+    print_video_info(raw_videos_path)
     # print_video_info(synched_videos_path)
 

Cameras/dianostics/skellycam_plots.py

@@ -0,0 +1,155 @@
+import logging
+import time
+from pathlib import Path
+from typing import Dict, List, Union
+
+import numpy as np
+from pydantic import BaseModel
+from scipy.stats import median_abs_deviation
+
+logger = logging.getLogger(__name__)
+
+
+class TimestampDiagnosticsDataClass(BaseModel):
+    mean_framerates_per_camera: dict
+    standard_deviation_framerates_per_camera: dict
+    median_framerates_per_camera: dict
+    median_absolute_deviation_per_camera: dict
+    mean_mean_framerate: float
+    mean_standard_deviation_framerates: float
+    mean_median_framerates: float
+    mean_median_absolute_deviation_per_camera: float
+
+
+def create_timestamp_diagnostic_plots(
+    timestamp_dictionary: Dict[int, np.ndarray],
+    path_to_save_plots_png: Union[str, Path],
+):
+    """plot some diagnostics to assess quality of camera sync"""
+
+    # opportunistic load of matplotlib to avoid startup time costs
+    from matplotlib import pyplot as plt
+
+    plt.set_loglevel("warning")
+
+    for timestamp_array in timestamp_dictionary.values():
+        timestamp_array /= 1e9
+
+    max_frame_duration = 0.1
+    fig = plt.figure(figsize=(18, 12))
+    session_name = Path(path_to_save_plots_png).parent.parent.parent.stem
+    recording_name = Path(path_to_save_plots_png).parent.parent.stem
+    fig.suptitle(f"Timestamps of synchronized frames\nsession: {session_name}, recording: {recording_name}")
+
+    ax1 = plt.subplot(
+        231,
+        title="(Raw) Camera Frame Timestamp vs Frame#\n(Lines should have same slope)",
+        xlabel="Frame#",
+        ylabel="Timestamp (sec)",
+    )
+    ax2 = plt.subplot(
+        232,
+        ylim=(0, max_frame_duration),
+        title="(Raw) Camera Frame Duration Trace",
+        xlabel="Frame#",
+        ylabel="Duration (sec)",
+    )
+    ax3 = plt.subplot(
+        233,
+        xlim=(0, max_frame_duration),
+        title="(Raw) Camera Frame Duration Histogram (count)",
+        xlabel="Duration(s, 1ms bins)",
+        ylabel="Probability",
+    )
+    ax4 = plt.subplot(
+        234,
+        title="(Synchronized) Camera Frame Timestamp vs Frame#\n(Lines should be on top of each other)",
+        xlabel="Frame#",
+        ylabel="Timestamp (sec)",
+    )
+    ax5 = plt.subplot(
+        235,
+        ylim=(0, max_frame_duration),
+        title="(Synchronized) Camera Frame Duration Trace",
+        xlabel="Frame#",
+        ylabel="Duration (sec)",
+    )
+    ax6 = plt.subplot(
+        236,
+        xlim=(0, max_frame_duration),
+        title="(Synchronized) Camera Frame Duration Histogram (count)",
+        xlabel="Duration(s, 1ms bins)",
+        ylabel="Probability",
+    )
+
+    for camera_id, timestamps in timestamp_dictionary.items():
+        ax1.plot(timestamps, label=f"Camera# {str(camera_id)}")
+        ax1.legend()
+        ax2.plot(np.diff(timestamps), ".")
+        ax3.hist(
+            np.diff(timestamps),
+            bins=np.arange(0, max_frame_duration, 0.0025),
+            alpha=0.5,
+        )
+
+    # for camera_id, timestamps in synchronized_timestamps_dictionary.items():
+    #     ax4.plot(timestamps, label=f"Camera# {str(camera_id)}")
+    #     ax4.legend()
+    #     ax5.plot(np.diff(timestamps), ".")
+    #     ax6.hist(
+    #         np.diff(timestamps),
+    #         bins=np.arange(0, max_frame_duration, 0.0025),
+    #         alpha=0.5,
+    #     )
+
+    plt.tight_layout()
+
+    fig_save_path = Path(path_to_save_plots_png)
+    plt.savefig(str(fig_save_path))
+    logger.info(f"Saving diagnostic figure tp: {fig_save_path}")
+
+def timestamps_array_to_dictionary(timestamps_array: np.ndarray) -> Dict[int, np.ndarray]:
+    return {i: timestamps_array[i, :] for i in range(timestamps_array.shape[0])}
+
+def calculate_camera_diagnostic_results(
+    timestamps_dictionary,
+) -> TimestampDiagnosticsDataClass:
+    mean_framerates_per_camera = {}
+    standard_deviation_framerates_per_camera = {}
+    median_framerates_per_camera = {}
+    median_absolute_deviation_per_camera = {}
+
+    for cam_id, timestamps in timestamps_dictionary.items():
+        timestamps_formatted = (np.asarray(timestamps) - timestamps[0]) / 1e9
+        frame_durations = np.diff(timestamps_formatted)
+        framerate_per_frame = 1 / frame_durations
+        mean_framerates_per_camera[cam_id] = np.nanmean(framerate_per_frame)
+        median_framerates_per_camera[cam_id] = np.nanmedian(framerate_per_frame)
+        standard_deviation_framerates_per_camera[cam_id] = np.nanstd(
+            framerate_per_frame
+        )
+        median_absolute_deviation_per_camera[cam_id] = median_abs_deviation(
+            framerate_per_frame
+        )
+
+    mean_mean_framerate = np.nanmean(list(mean_framerates_per_camera.values()))
+    mean_standard_deviation_framerates = np.nanmean(
+        list(standard_deviation_framerates_per_camera.values())
+    )
+    mean_median_framerates = np.nanmean(list(median_framerates_per_camera.values()))
+    mean_median_absolute_deviation_per_camera = np.nanmean(
+        list(median_absolute_deviation_per_camera.values())
+    )
+
+    return TimestampDiagnosticsDataClass(
+        mean_framerates_per_camera=mean_framerates_per_camera,
+        standard_deviation_framerates_per_camera=standard_deviation_framerates_per_camera,
+        median_framerates_per_camera=median_framerates_per_camera,
+        median_absolute_deviation_per_camera=median_absolute_deviation_per_camera,
+        mean_mean_framerate=float(mean_mean_framerate),
+        mean_standard_deviation_framerates=float(mean_standard_deviation_framerates),
+        mean_median_framerates=float(mean_median_framerates),
+        mean_median_absolute_deviation_per_camera=float(
+            mean_median_absolute_deviation_per_camera
+        ),
+    )

Cameras/exploration_scripts/pylon_viewer_timestamp_synchronize.py

@@ -0,0 +1,131 @@
+from datetime import datetime, timedelta
+from typing import Dict
+import cv2
+
+from pathlib import Path
+
+
+class TimestampSynchronize:
+    def __init__(self, folder_path: Path):
+        self.raw_videos_path = folder_path / "raw_videos"
+        self.synched_videos_path = folder_path / "synched_videos"
+
+        self.synched_videos_path.mkdir(parents=True, exist_ok=True)
+
+    def synchronize(self):
+        self.setup()
+        target_framecount = (
+            self.get_lowest_postoffset_frame_count() - 1
+        )  # -1 accounts for rounding errors in calculating offset
+        print(f"synchronizing videos to target framecount: {target_framecount}")
+        for video_name, cap in self.capture_dict.items():
+            print(f"synchronizing: {video_name}")
+            current_framecount = 0
+            offset = self.frame_offset_dict[video_name]
+            while current_framecount < target_framecount:  # < to account for 0 indexing
+                ret, frame = cap.read()
+                if not ret:
+                    raise ValueError(f"{video_name} has no more frames.")
+                if offset <= 0:
+                    self.writer_dict[video_name].write(frame)
+                    current_framecount += 1
+                else:
+                    offset -= 1
+        self.close()
+        print("Done synchronizing")
+
+    def setup(self):
+        print("Setting up for synchronization...")
+        self.create_capture_dict()
+        self.validate_fps()
+        self.create_writer_dict()
+        self.create_starting_timestamp_dict()
+        self.create_frame_offset_dict()
+
+    def create_capture_dict(self):
+        self.capture_dict = {
+            video_path.name: cv2.VideoCapture(str(video_path))
+            for video_path in self.raw_videos_path.iterdir()
+        }
+
+    def create_writer_dict(self):
+        self.writer_dict = {
+            video_name: cv2.VideoWriter(
+                str(self.synched_videos_path / (video_name.split(".")[0] + ".mp4")),
+                cv2.VideoWriter.fourcc(*"mp4v"),
+                cap.get(cv2.CAP_PROP_FPS),
+                (
+                    int(cap.get(cv2.CAP_PROP_FRAME_WIDTH)),
+                    int(cap.get(cv2.CAP_PROP_FRAME_HEIGHT)),
+                ),
+            )
+            for video_name, cap in self.capture_dict.items()
+        }
+
+    def create_starting_timestamp_dict(self):
+        self.starting_timestamp_dict = {
+            video_name: datetime.fromisoformat(video_name.split(".")[0].split("__")[-1])
+            for video_name in self.capture_dict.keys()
+        }
+
+    def create_frame_offset_dict(self):
+        latest_start = sorted(self.starting_timestamp_dict.values())[-1]
+        frame_duration_seconds = 1 / self.fps
+
+        self.frame_offset_dict: Dict[str, int] = {}
+
+        for video_name, time in self.starting_timestamp_dict.items():
+            offset_microseconds = (latest_start - time) / timedelta(microseconds=1)
+            offset_frames = round(
+                timedelta(microseconds=offset_microseconds)
+                / timedelta(seconds=frame_duration_seconds)
+            )
+            print(f"{video_name} offset in microseconds: {offset_microseconds}")
+            print(f"{video_name} offset in frames: {offset_frames}")
+            print(
+                f"{video_name} total frames: {int(self.capture_dict[video_name].get(cv2.CAP_PROP_FRAME_COUNT))}"
+            )
+            self.frame_offset_dict[video_name] = offset_frames
+
+    def get_lowest_postoffset_frame_count(self) -> int:
+        return int(
+            min(
+                cap.get(cv2.CAP_PROP_FRAME_COUNT) - self.frame_offset_dict[video_name]
+                for video_name, cap in self.capture_dict.items()
+            )
+        )
+
+    def validate_fps(self):
+        fps = set(cap.get(cv2.CAP_PROP_FPS) for cap in self.capture_dict.values())
+
+        if len(fps) > 1:
+            print(f"set of video fps: {fps}")
+            raise ValueError("Not all videos have the same fps")
+
+        self.fps = fps.pop()
+
+    def close(self):
+        print("Closing all capture objects and writers")
+        self.release_captures()
+        self.release_writers()
+
+    def release_captures(self):
+        for cap in self.capture_dict.values():
+            cap.release()
+
+    def release_writers(self):
+        for writer in self.writer_dict.values():
+            writer.release()
+
+
+if __name__ == "__main__":
+    # folder_path = Path(
+    #     "/Users/philipqueen/Documents/Humon Research Lab/Basler Stuff/calibration_attempt/"
+    # )
+
+    folder_path = Path(
+        "/Users/philipqueen/Documents/Humon Research Lab/Basler Stuff/fabio_hand/"
+    )
+
+    timestamp_synchronize = TimestampSynchronize(folder_path)
+    timestamp_synchronize.synchronize()