Added --number-of-checks option, WiP, #82.

Capture/nosignal/reprostim/nosignal.py

@@ -1,6 +1,7 @@
 import logging.config
 import sys
 import time
+from datetime import timedelta
 from pydantic import BaseModel, Field
 import click
 import cv2
@@ -60,11 +61,20 @@ def main_exit(code: int) -> int:
     return code
 
 
-def find_no_signal(video_path: str, step: int = 1) -> VideoInfo:
+def find_no_signal(video_path: str, step: int = 1,
+                   number_of_checks: int = 0) -> VideoInfo:
     if step < 1:
         logger.error("Step must be greater than 0")
         raise ValueError("Step must be greater than 0")
 
+    if number_of_checks < 0:
+        logger.error("Number of checks must be 0 or greater than 0")
+        raise ValueError("Number of checks must be 0 or greater than 0")
+
+    if number_of_checks > 0 and step > 1:
+        logger.warning(f"Number of checks is set, specified step value({step}) will be ignored.")
+        step = 1
+
     vi: VideoInfo = VideoInfo(fps=0, frames_count=0, nosignal_count=0,
                               nosignal_rate=0.0, scanned_count=0)
     cap = cv2.VideoCapture(video_path)
@@ -77,8 +87,13 @@ def find_no_signal(video_path: str, step: int = 1) -> VideoInfo:
     frame_count = int(cap.get(cv2.CAP_PROP_FRAME_COUNT))
     logger.debug(f"frame_count={str(frame_count)}")
     fps = cap.get(cv2.CAP_PROP_FPS)
+    frame_width: int = int(cap.get(cv2.CAP_PROP_FRAME_WIDTH))
+    frame_height: int = int(cap.get(cv2.CAP_PROP_FRAME_HEIGHT))
     vi.fps = round(fps, 2)
-    logger.debug(f"fps={str(fps)}")
+    logger.info(f"Video resolution={frame_width}x{frame_height}, fps={str(fps)}, "
+                f"frames count={str(frame_count)}")
+
+    logger.info(f"pos ms={cap.get(cv2.CAP_PROP_POS_MSEC)}, pos frames={cap.get(cv2.CAP_PROP_POS_FRAMES)}")
     nosignal_frames = []
 
     #    for i in range(frame_count):
@@ -91,29 +106,51 @@ def find_no_signal(video_path: str, step: int = 1) -> VideoInfo:
     #        if not has_rainbow(frame):
     #            no_signal_frames.append(i)
 
-    frame_counter: int = 0
+    pos_first_frame: int = 0
+    pos_last_frame: int = pos_first_frame + frame_count
+
+    if pos_first_frame > pos_last_frame:
+        logger.error(f"Invalid frame range: {pos_first_frame} - {pos_last_frame}")
+        #raise RuntimeError(f"Invalid frame range: {pos_first_frame} - {pos_last_frame}")
+
+    pos_cur_frame: int = pos_first_frame
+    pos_next_frame: int = pos_cur_frame
     nosignal_counter: int = 0
     scan_counter: int = 0
+    ts_progress = time.time()
+    progress_interval: float = 1
     while True:
-        logger.debug(f"frame_counter={frame_counter}, "
-                     f"time={round(frame_counter / fps, 1)}")
-        cap.set(cv2.CAP_PROP_POS_FRAMES, frame_counter)
+        if time.time() > ts_progress:
+            dt: str = str(timedelta(milliseconds=int(cap.get(cv2.CAP_PROP_POS_MSEC))))
+            logger.info(f"Scanning progress: {pos_cur_frame} / {pos_last_frame}, {dt}")
+            ts_progress = time.time() + progress_interval
+
+        logger.debug(f"pos_frame={pos_cur_frame}, "
+                     f"time={round(pos_cur_frame / fps, 1)}")
+        cap.set(cv2.CAP_PROP_POS_FRAMES, pos_cur_frame)
         ret, frame = cap.read()
         if ret is False:
-            logger.debug(f"Failed reading frame {frame_counter}")
+            logger.debug(f"Failed reading frame {pos_cur_frame}")
             break
-        frame_counter += step
         scan_counter += 1
 
+        # set next frame position
+        if number_of_checks > 0:
+            pos_next_frame = pos_first_frame + frame_count * scan_counter / number_of_checks
+        else:
+            pos_next_frame = pos_cur_frame + step
+
         if has_rainbow(frame):
             logger.debug("rainbow-yes")
             nosignal_counter = nosignal_counter + 1
-            nosignal_frames.append(frame_counter)
+            nosignal_frames.append(pos_cur_frame)
         else:
             logger.debug("rainbow-no")
 
-    logger.debug(f"frame_counter={frame_counter}")
-    vi.frames_count = frame_counter
+        pos_cur_frame = pos_next_frame
+
+    logger.debug(f"pos_frame={pos_cur_frame}")
+    vi.frames_count = pos_cur_frame
     vi.nosignal_count = nosignal_counter
     vi.scanned_count = scan_counter
     if scan_counter > 0:
@@ -136,20 +173,27 @@ def find_no_signal(video_path: str, step: int = 1) -> VideoInfo:
                                  'WARNING', 'ERROR',
                                  'CRITICAL']),
               help='Set the logging level')
-@click.option('--step', default=1, type=int, help='Specify the step '
-                                                  'for processing '
-                                                  'frames')
+@click.option('--step', default=1, type=int,
+              help='Specify the step '
+                   'for processing '
+                   'frames, default is 1.')
+@click.option('--number-of-checks', default=0, type=int,
+              help='Specify the number '
+                   'of checks across entire video'
+                   'frames. When set to 0, '
+                   'all frames are checked.')
 @click.option('--threshold', default=0.01, type=float,
               help='Specify the threshold for nosignal frames, default is '
-                   '0.01 which means 1% of the total frames.')
+                   '0.01 which means 1% of the totally checked frames.')
 @click.pass_context
-def main(ctx, path: str, log_level, step: int, threshold: float):
+def main(ctx, path: str, log_level, step: int,
+         number_of_checks: int, threshold: float):
     logger.setLevel(log_level)
     logger.debug("nosignal.py tool")
 
     logger.debug(f"path={path}")
 
-    res = find_no_signal(path, step)
+    res = find_no_signal(path, step, number_of_checks)
     logger.info(f"Scan result    : {str(res)}")
     if res.nosignal_count > 0 and res.nosignal_rate > threshold:
         click.echo(