seekable

simba/SimBA.py

@@ -108,6 +108,7 @@
 from simba.ui.pop_ups.roi_analysis_time_bins_pop_up import \
     ROIAnalysisTimeBinsPopUp
 from simba.ui.pop_ups.roi_features_plot_pop_up import VisualizeROIFeaturesPopUp
+from simba.ui.pop_ups.check_videos_seekable_pop_up import CheckVideoSeekablePopUp
 from simba.ui.pop_ups.roi_size_standardizer_popup import \
     ROISizeStandardizerPopUp
 from simba.ui.pop_ups.roi_tracking_plot_pop_up import VisualizeROITrackingPopUp
@@ -893,7 +894,8 @@ def __init__(self):
         video_process_menu.add_command(label="Box blur videos", compound="left", image=self.menu_icons["blur"]["img"], command=BoxBlurPopUp, font=Formats.FONT_REGULAR.value)
         video_process_menu.add_command(label="Cross-fade videos", compound="left", image=self.menu_icons["crossfade"]["img"], command=CrossfadeVideosPopUp, font=Formats.FONT_REGULAR.value)
         video_process_menu.add_command(label="Create average frames from videos", compound="left", image=self.menu_icons["average"]["img"], command=CreateAverageFramePopUp, font=Formats.FONT_REGULAR.value)
-        video_process_menu.add_command(label="Video background remover...", compound="left", image=self.menu_icons["remove_bg"]["img"], command=BackgroundRemoverPopUp, font=Formats.FONT_REGULAR.value)
+        video_process_menu.add_command(label="Video background remover", compound="left", image=self.menu_icons["remove_bg"]["img"], command=BackgroundRemoverPopUp, font=Formats.FONT_REGULAR.value)
+        video_process_menu.add_command(label="Validate video seekability", compound="left", image=self.menu_icons["search"]["img"], command=CheckVideoSeekablePopUp, font=Formats.FONT_REGULAR.value)
         video_process_menu.add_command(label="Visualize pose-estimation in folder...", compound="left", image=self.menu_icons["visualize"]["img"], command=VisualizePoseInFolderPopUp, font=Formats.FONT_REGULAR.value)
         help_menu = Menu(menu)
         menu.add_cascade(label="Help", menu=help_menu)

simba/mixins/image_mixin.py

@@ -989,6 +989,10 @@ def read_img_batch_from_video(video_path: Union[str, os.PathLike],
         :example:
         >>> ImageMixin().read_img_batch_from_video(video_path='/Users/simon/Desktop/envs/troubleshooting/two_black_animals_14bp/videos/Together_1.avi', start_frm=0, end_frm=50)
         """
+        if platform.system() == "Darwin":
+            if not multiprocessing.get_start_method(allow_none=True):
+                multiprocessing.set_start_method("fork", force=True)
+
         check_file_exist_and_readable(file_path=video_path)
         video_meta_data = get_video_meta_data(video_path=video_path)
         check_int(name=ImageMixin().__class__.__name__,value=start_frm, min_value=0,max_value=video_meta_data["frame_count"])

simba/ui/pop_ups/check_videos_seekable_pop_up.py

@@ -0,0 +1,72 @@
+import os
+from datetime import datetime
+from simba.mixins.pop_up_mixin import PopUpMixin
+from simba.ui.tkinter_functions import SimbaCheckbox, DropDownMenu, FileSelect, FolderSelect, SimbaButton
+from simba.utils.enums import Formats, Options
+from tkinter import *
+from simba.utils.checks import check_if_dir_exists, is_valid_video_file
+from simba.utils.read_write import find_files_of_filetypes_in_directory, get_desktop_path
+from simba.video_processors.video_processing import is_video_seekable
+
+class CheckVideoSeekablePopUp(PopUpMixin):
+    """
+    GUI pop-up window for checking if a video, or a directory of videos, are seekable.
+
+    :example:
+    _ = CheckVideoSeekablePopUp()
+    """
+
+    def __init__(self):
+        PopUpMixin.__init__(self, title="CHECK IF VIDEOS ARE SEEKABLE")
+        settings_frm = LabelFrame(self.main_frm, text="SETTINGS", font=Formats.FONT_HEADER.value)
+        batch_size_options = list(range(100, 5100, 100))
+        batch_size_options.insert(0, 'NONE')
+        self.use_gpu_cb, self.use_gpu_var = SimbaCheckbox(parent=settings_frm, txt="Use GPU (reduced runtime)", txt_img='gpu_2')
+        self.batch_size_dropdown = DropDownMenu(settings_frm, "FRAME BATCH SIZE:", batch_size_options, "15")
+        self.batch_size_dropdown.setChoices('NONE')
+        single_video_frm = LabelFrame(self.main_frm, text="SINGLE VIDEO", font=Formats.FONT_HEADER.value)
+        self.single_video_path = FileSelect(single_video_frm, "VIDEO PATH:", title="Select a video file", lblwidth=25, file_types=[("VIDEO", Options.ALL_VIDEO_FORMAT_STR_OPTIONS.value)])
+        single_run_btn = SimbaButton(parent=single_video_frm, txt="RUN", img='rocket', font=Formats.FONT_REGULAR.value, cmd=self.run, cmd_kwargs={'directory': lambda: False})
+
+        multiple_video_frm = LabelFrame(self.main_frm, text="VIDEO DIRECTORY", font=Formats.FONT_HEADER.value)
+        self.directory_path = FolderSelect(multiple_video_frm, "VIDEO DIRECTORY PATH:", title="Select folder with videos: ", lblwidth="25")
+        dir_run_btn = SimbaButton(parent=multiple_video_frm, txt="RUN", img='rocket', font=Formats.FONT_REGULAR.value, cmd=self.run, cmd_kwargs={'directory': lambda: True})
+
+
+        settings_frm.grid(row=0, column=0, sticky=NW)
+        self.use_gpu_cb.grid(row=0, column=0, sticky=NW)
+        self.batch_size_dropdown.grid(row=1, column=0, sticky=NW)
+
+
+        single_video_frm.grid(row=1, column=0, sticky=NW)
+        self.single_video_path.grid(row=0, column=0, sticky=NW)
+        single_run_btn.grid(row=1, column=0, sticky=NW)
+
+        multiple_video_frm.grid(row=2, column=0, sticky=NW)
+        self.directory_path.grid(row=0, column=0, sticky=NW)
+        dir_run_btn.grid(row=1, column=0, sticky=NW)
+        #self.main_frm.mainloop()
+
+    def run(self, directory: bool):
+        if directory:
+            data_path = self.directory_path.folder_path
+            check_if_dir_exists(in_dir=dir, source=self.__class__.__name__)
+            file_paths = find_files_of_filetypes_in_directory(directory=data_path, extensions=Options.ALL_VIDEO_FORMAT_OPTIONS.value, raise_error=True)
+            for file_path in file_paths:
+                _ = is_valid_video_file(file_path=file_path, raise_error=True)
+        else:
+            data_path = self.single_video_path.file_path
+            is_valid_video_file(file_path=data_path, raise_error=True)
+        gpu = self.use_gpu_var.get()
+        batch_size = self.batch_size_dropdown.getChoices()
+        if batch_size == 'NONE':
+            batch_size = None
+        else:
+            batch_size = int(batch_size)
+        desktop_path = get_desktop_path()
+        save_path = os.path.join(desktop_path, f'seekability_test_{datetime.now().strftime("%Y%m%d%H%M%S")}.csv')
+        _ = is_video_seekable(data_path=data_path,
+                              gpu=gpu,
+                              batch_size=batch_size,
+                              verbose=False,
+                              save_path=save_path)

simba/utils/checks.py

@@ -1513,4 +1513,28 @@ def check_all_dfs_in_list_has_same_cols(dfs: List[pd.DataFrame], raise_error: bo
             raise MissingColumnsError(msg=f"The data in {source} directory do not contain the same headers. Some files are missing the headers: {missing_headers}", source=check_all_dfs_in_list_has_same_cols.__name__)
         else:
             return False
-    return True
+    return True
+
+
+def is_valid_video_file(file_path: Union[str, os.PathLike], raise_error: bool = True):
+    """
+    Check if a file path is a valid video file.
+    """
+    check_file_exist_and_readable(file_path=file_path)
+    try:
+        cap = cv2.VideoCapture(file_path)
+        if not cap.isOpened():
+            if not raise_error:
+                return False
+            else:
+                raise InvalidFilepathError(msg=f'The path {file_path} is not a valid video file', source=is_valid_video_file.__name__)
+        return True
+    except Exception:
+        if not raise_error:
+            return False
+        else:
+            raise InvalidFilepathError(msg=f'The path {file_path} is not a valid video file', source=is_valid_video_file.__name__)
+    finally:
+        if 'cap' in locals():
+            if cap.isOpened():
+                cap.release()

simba/utils/read_write.py

@@ -2615,5 +2615,14 @@ def create_empty_xlsx_file(xlsx_path: Union[str, os.PathLike]):
     check_if_dir_exists(in_dir=os.path.dirname(xlsx_path))
     pd.DataFrame().to_excel(xlsx_path, index=False)
 
-
+def get_desktop_path(raise_error: bool = False):
+    """ Get the path to the user desktop directory """
+    desktop_path = os.path.join(os.path.expanduser("~"), "Desktop")
+    if not os.path.isdir(desktop_path):
+        if raise_error:
+            raise InvalidFilepathError(msg=f'{desktop_path} is not a valid directory')
+        else:
+            return None
+    else:
+        return desktop_path
 

simba/video_processors/video_processing.py

@@ -16,6 +16,7 @@
 
 import cv2
 import numpy as np
+import pandas as pd
 from PIL import Image, ImageTk
 from shapely.geometry import Polygon
 from skimage.color import label2rgb
@@ -46,7 +47,7 @@
                                 InvalidFileTypeError, InvalidInputError,
                                 InvalidVideoFileError, NoDataError,
                                 NoFilesFoundError, NotDirectoryError,
-                                ResolutionError)
+                                ResolutionError, SimBAGPUError)
 from simba.utils.lookups import (get_ffmpeg_crossfade_methods, get_fonts,
                                  get_named_colors, percent_to_crf_lookup,
                                  percent_to_qv_lk,
@@ -56,7 +57,7 @@
     check_if_hhmmss_timestamp_is_valid_part_of_video,
     concatenate_videos_in_folder, find_all_videos_in_directory, find_core_cnt,
     find_files_of_filetypes_in_directory, get_fn_ext, get_video_meta_data,
-    read_config_entry, read_config_file, read_frm_of_video)
+    read_config_entry, read_config_file, read_frm_of_video, read_img_batch_from_video_gpu)
 from simba.utils.warnings import (FileExistWarning, FrameRangeWarning,
                                   InValidUserInputWarning,
                                   SameInputAndOutputWarning)
@@ -4574,6 +4575,82 @@ def get_video_slic(video_path: Union[str, os.PathLike],
     concatenate_videos_in_folder(in_folder=temp_folder, save_path=save_path)
     stdout_success(msg=f'SLIC video saved at {save_path}', elapsed_time=timer.elapsed_time_str)
 
+
+
+def is_video_seekable(data_path: Union[str, os.PathLike],
+                      gpu: bool = False,
+                      batch_size: Optional[int] = None,
+                      verbose: bool = False,
+                      raise_error: bool = True,
+                      save_path: Optional[Union[str, os.PathLike]] = None) -> Union[None, bool, Tuple[Dict[str, List[int]]]]:
+    """
+    Determines if the given video file(s) are seekable and can be processed frame-by-frame without issues.
+
+    This function checks if all frames in the specified video(s) can be read sequentially. It can process videos
+    using either CPU or GPU, with optional batch processing to handle memory limitations. If unreadable frames are
+    detected, the function can either raise an error or return a result indicating the issue.
+
+    :param Union[str, os.PathLike] data_path: Path to the video file or a path to a directory containing video files.
+    :param bool gpu: If True, then use GPU. Else, CPU.
+    :param Optional[int] batch_size: Optional int representing the number of frames in each video to process sequentially. If None, all frames in a video is processed at once. Use a smaller value to avoid MemoryErrors. Default None.
+    :param bool verbose: If True, prints progress. Default None.
+    :param bool raise_error: If True, raises error if not all passed videos are seeakable.
+
+    :example:
+    >>> _ = is_video_seekable(data_path='/Users/simon/Desktop/unseekable/20200730_AB_7dpf_850nm_0003_fps_5.mp4', batch_size=400)
+    """
+
+    if batch_size is not None:
+        check_int(name=f'{is_video_seekable.__name__}', value=batch_size, min_value=1)
+    if save_path is not None:
+        check_if_dir_exists(in_dir=os.path.dirname(save_path))
+    check_valid_boolean(value=[verbose], source=f'{is_video_seekable.__name__} verbose')
+    if not check_ffmpeg_available():
+        raise FFMPEGNotFoundError(msg='SimBA could not find FFMPEG on the computer.', source=is_video_seekable.__name__)
+    if gpu and not check_nvidea_gpu_available():
+        raise SimBAGPUError(msg='SimBA could not find a NVIDEA GPU on the computer and GPU is set to True.', source=is_video_seekable.__name__)
+    if os.path.isfile(data_path):
+        data_paths = [data_path]
+    elif os.path.isdir(data_path):
+        data_paths = find_files_of_filetypes_in_directory(directory=data_path, extensions=Options.ALL_VIDEO_FORMAT_OPTIONS.value, raise_error=True)
+    else:
+        raise InvalidInputError(msg=f'{data_path} is not a valid in directory or file path.', source=is_video_seekable.__name__)
+    _ = [get_video_meta_data(video_path=x) for x in data_paths]
+
+    results = {}
+    for file_cnt, file_path in enumerate(data_paths):
+        _, video_name, _ = get_fn_ext(filepath=file_path)
+        print(f'Checking seekability video {video_name}...')
+        video_meta_data = get_video_meta_data(video_path=file_path)
+        video_frm_ranges = np.arange(0, video_meta_data['frame_count']+1)
+        if batch_size is not None:
+            video_frm_ranges = np.array_split(video_frm_ranges, max(1, int(video_frm_ranges.shape[0]/batch_size)))
+        else:
+            video_frm_ranges = [video_frm_ranges]
+        video_error_frms = []
+        for video_frm_range in video_frm_ranges:
+            if not gpu:
+                imgs = ImageMixin.read_img_batch_from_video(video_path=file_path, start_frm=video_frm_range[0], end_frm=video_frm_range[-1], verbose=verbose)
+            else:
+                imgs = read_img_batch_from_video_gpu(video_path=file_path, start_frm=video_frm_range[0], end_frm=video_frm_range[-1], verbose=verbose)
+            invalid_frms = [k for k, v in imgs.items() if v is None]
+            video_error_frms.extend(invalid_frms)
+        results[video_name] = video_error_frms
+
+    if all(len(v) == 0 for v in results.values()):
+        if verbose:
+            stdout_success(msg=f'The {len(data_paths)} videos are valid.', source=is_video_seekable.__name__)
+        return True
+    else:
+        if save_path is not None:
+            out_df = pd.DataFrame.from_dict(data=results).T
+            out_df.to_csv(save_path)
+            FrameRangeWarning(msg=f'Some videos have unseekable frames. See {save_path} for results', source=is_video_seekable.__name__)
+        if raise_error:
+            raise FrameRangeError(msg=f'{results} The frames in the videos listed are unreadable. Consider re-encoding these videos.', source=is_video_seekable.__name__)
+        else:
+            return (False, results)
+
 # video_paths = ['/Users/simon/Desktop/envs/simba/troubleshooting/beepboop174/project_folder/merge/Trial    10_clipped_gantt.mp4',
 #                '/Users/simon/Desktop/envs/simba/troubleshooting/beepboop174/project_folder/merge/Trial    10_clipped.mp4',
 #                '/Users/simon/Desktop/envs/simba/troubleshooting/beepboop174/project_folder/merge/Trial    10_clipped_line.mp4',