feat: v0.0.2 release

Author: minkyu-choi07

main.py

@@ -3,7 +3,7 @@
 from ns_vfs.config.loader import load_config
 from ns_vfs.model.vision.grounding_dino import GroundingDino
 from ns_vfs.processor.video_processor import (
-    VideoFrameProcessor,
+    VideoFrameWindowProcessor,
 )
 from ns_vfs.video_to_automaton import VideotoAutomaton
 
@@ -20,14 +20,14 @@
             weight_path=config.GROUNDING_DINO.GROUNDING_DINO_CHECKPOINT_PATH,
             config_path=config.GROUNDING_DINO.GROUNDING_DINO_CONFIG_PATH,
         ),
-        video_processor=VideoFrameProcessor(
+        video_processor=VideoFrameWindowProcessor(
             video_path=sample_video_path,
             artifact_dir=config.VERSION_AND_PATH.ARTIFACTS_PATH,
         ),
-        proposition_set=["face", "person", "car"],
+        proposition_set=["person", "drink"],
         artifact_dir=config.VERSION_AND_PATH.ARTIFACTS_PATH,
     )
 
-    frame_window_automata = frame2automaton.start()
+    frame_window_automata = frame2automaton.run()
 
     print("Development is in progress.")

ns_vfs/data/frame.py

@@ -5,33 +5,33 @@
 
 @dataclasses.dataclass
 class Frame:
+    """Frame class."""
+
     frame_index: int
     frame_image: np.ndarray
     propositional_probability: dict = dataclasses.field(default_factory=dict)
 
 
 @dataclasses.dataclass
 class FrameWindow:
+    """Frame window class."""
+
     frame_window_idx: int
     frame_image_set: list = dataclasses.field(default_factory=list)
     states: list = dataclasses.field(default_factory=list)
     transitions: list = dataclasses.field(default_factory=list)
+    verification_result: bool = False
 
     def get_propositional_confidence(self):
         propositional_confidence = [
-            []
-            for i in range(
-                len(self.frame_image_set[0].propositional_probability.keys())
-            )
+            [] for i in range(len(self.frame_image_set[0].propositional_probability.keys()))
         ]
         propositional_confidence
         for frame in self.frame_image_set:
             frame: Frame
             idx = 0
             for prop in frame.propositional_probability.keys():
-                propositional_confidence[idx].append(
-                    frame.propositional_probability[prop]
-                )
+                propositional_confidence[idx].append(frame.propositional_probability[prop])
                 idx += 1
         self.propositional_confidence = propositional_confidence
         return propositional_confidence

ns_vfs/processor/video_processor.py

@@ -9,14 +9,13 @@
 import numpy as np
 
 from ns_vfs.data.frame import Frame, FrameWindow
-from ns_vfs.verification import build_label_func, build_trans_matrix
+from ns_vfs.verification import check_automaton
 
 
 class VideoProcessor(abc.ABC):
     @abc.abstractmethod
     def import_video(self, video_path) -> any:
         """Read video from video_path."""
-        pass
 
 
 class VideoFrameProcessor(VideoProcessor):
@@ -27,22 +26,27 @@ def __init__(self, video_path: str, artifact_dir: str) -> None:
 
         Args:
             video_path (str): Path to video file.
+            artifact_dir (str): Path to artifact directory.
         """
         self._video_path = video_path
         self._artifact_dir = os.path.join(artifact_dir, "video_frame_processor")
         self._processed_frames = None
         self._frame_window = None
         self.import_video(video_path)
 
-    def import_video(self, video_path):
-        """Read video from video_path."""
-        self._cap = cv2.VideoCapture(video_path)
-        self.original_video_height = self._cap.get(cv2.CAP_PROP_FRAME_HEIGHT)
-        self.original_video_width = self._cap.get(cv2.CAP_PROP_FRAME_WIDTH)
-        self.original_vidoe_fps = self._cap.get(cv2.CAP_PROP_FPS)
-        self.original_frame_count = self._cap.get(cv2.CAP_PROP_FRAME_COUNT)
+    def _resize_frame(
+        self, frame_img: np.ndarray, frame_scale: int
+    ) -> np.ndarray:
+        """Resize frame image.
+
+        Args:
+            frame_img (np.ndarray): Frame image.
+            frame_scale (int): Scale of frame.
 
-    def _resize_frame(self, frame_img, frame_scale):
+
+        Returns:
+            np.ndarray: Resized frame image.
+        """
         return cv2.resize(
             frame_img,
             (
@@ -51,7 +55,76 @@ def _resize_frame(self, frame_img, frame_scale):
             ),
         )
 
-    def build_frame_synchronously(
+    def import_video(self, video_path: str) -> None:
+        """Read video from video_path.
+
+        Args:
+            video_path (str): Path to video file.
+        """
+        self._cap = cv2.VideoCapture(video_path)
+        self.original_video_height = self._cap.get(cv2.CAP_PROP_FRAME_HEIGHT)
+        self.original_video_width = self._cap.get(cv2.CAP_PROP_FRAME_WIDTH)
+        self.original_vidoe_fps = self._cap.get(cv2.CAP_PROP_FPS)
+        self.original_frame_count = self._cap.get(cv2.CAP_PROP_FRAME_COUNT)
+
+    def get_video_by_frame(
+        self,
+        frame_scale: int = 5,
+        frame_duration_sec: int = 2,
+        return_format: str = "ndarray",
+    ) -> np.ndarray | list:
+        """Get video frames by frame_scale and second_per_frame.
+
+        Args:
+            frame_scale (int, optional): Scale of frame. Defaults to 5.
+            second_per_frame (int, optional): Second per frame. Defaults to 2.
+            return_format (str, optional): Return format. Defaults to "npndarray".
+
+        Returns:
+        any: Video frames.
+        """
+        frames = list()
+        frame_counter = 0
+        frame_per_sec = int(round(self.original_vidoe_fps) * frame_duration_sec)
+        while self._cap.isOpened():
+            ret, frame_img = self._cap.read()
+            if not ret:
+                break
+            if frame_counter % frame_per_sec == 0:
+                if frame_scale is not None:
+                    frame_img = self._resize_frame(frame_img, frame_scale)
+
+                frames.append(frame_img)
+            if cv2.waitKey(1) & 0xFF == ord("q"):  # on press of q break
+                break
+            frame_counter += 1
+        self._cap.release()
+        cv2.destroyAllWindows()
+        if return_format == "npndarray":
+            self._processed_frames = np.array(frames)
+            return np.array(frames)
+        else:
+            self._processed_frames = frames
+            return frames
+
+
+class VideoFrameWindowProcessor(VideoFrameProcessor):
+    """Video Frame Window Processor.
+
+    It processes video frame, build automaton and verification
+    concurretnly by sliding window.
+    """
+
+    def __init__(self, video_path: str, artifact_dir: str) -> None:
+        """Video Frame Processor.
+
+        Args:
+            video_path (str): Path to video file.
+            artifact_dir (str): Path to artifact directory.
+        """
+        super().__init__(video_path, artifact_dir)
+
+    def build_frame_window_synchronously(
         self,
         proposition_set: list,
         calculate_propositional_confidence: callable,
@@ -60,10 +133,25 @@ def build_frame_synchronously(
         frame_scale: int | None = None,
         sliding_window_size: int = 5,
         save_image: bool = False,
-    ):
+    ) -> None:
+        """Build frame window synchronously.
+
+        It processes video frame, build automaton and verification
+        concurrently by sliding window.
+
+        Args:
+            proposition_set (list): List of propositions.
+            calculate_propositional_confidence (callable): Calculate propositional confidence.
+            build_automaton (callable): Build automaton.
+            frame_duration_sec (int, optional): Second per frame. Defaults to 2.
+            frame_scale (int | None, optional): Scale of frame. Defaults to None.
+            sliding_window_size (int, optional): Size of sliding window. Defaults to 5.
+            save_image (bool, optional): Save image. Defaults to False.
+        """
         frame_window_idx = 0
         frame_counter = 0
         frame_idx = 0
+
         # Calculate the frame skip rate
         frame_window = dict(frame_window_idx=0)
         temp_frame_set = list()
@@ -118,60 +206,19 @@ def build_frame_synchronously(
                     states, transitions = build_automaton(
                         frame_set, propositional_confidence
                     )
-                    frame_window[frame_window_idx].states = states
-                    frame_window[frame_window_idx].transitions = transitions
 
                     # Verification - Build Transition Matrix
-                    transition_matrix = build_trans_matrix(
-                        transitions=transitions, states=states
+                    verification_result = check_automaton(
+                        transitions=transitions,
+                        states=states,
+                        proposition_set=proposition_set,
                     )
-                    state_labeling = build_label_func(states, proposition_set)
-                    transition_matrix
-                    state_labeling
-
-                    frame_window_idx += 1
-                    temp_frame_set.pop(0)
-
-                frame_idx += 1
-            frame_counter += 1
-
-        self._cap.release()
-        self._frame_window = frame_window
-        return frame_window
-
-    def get_frame_by_sliding_window(
-        self,
-        frame_duration_sec: int = 2,
-        frame_scale: int | None = None,
-        sliding_window_size: int = 5,
-    ):
-        frame_window_idx = 0
-        frame_counter = 0
-        frame_idx = 0
-        # Calculate the frame skip rate
-        frame_window = dict(frame_window_idx=0)
-        temp_frame_set = list()
+                    frame_window[frame_window_idx].states = states
+                    frame_window[frame_window_idx].transitions = transitions
+                    frame_window[
+                        frame_window_idx
+                    ].verification_result = verification_result
 
-        while self._cap.isOpened():
-            ret, frame_img = self._cap.read()
-            if not ret:
-                break
-            if (
-                frame_counter
-                % int(frame_duration_sec * self.original_vidoe_fps)
-                == 0
-            ):
-                if frame_scale is not None:
-                    frame_img = self._resize_frame(frame_img, frame_scale)
-                Frame
-                temp_frame_set.append(
-                    Frame(frame_index=frame_idx, frame_image=frame_img)
-                )
-                if len(temp_frame_set) == sliding_window_size:
-                    frame_window[frame_window_idx] = FrameWindow(
-                        frame_window_idx=frame_window_idx,
-                        frame_image_set=temp_frame_set.copy(),
-                    )
                     frame_window_idx += 1
                     temp_frame_set.pop(0)
 
@@ -182,53 +229,21 @@ def get_frame_by_sliding_window(
         self._frame_window = frame_window
         return frame_window
 
-    def get_frame(
-        self,
-        frame_scale: int = 5,
-        second_per_frame: int = 2,
-        return_format: str = "ndarray",
-    ) -> np.ndarray | list:
-        """Get video frames by frame_scale and second_per_frame.
-
-        Args:
-            frame_scale (int, optional): Scale of frame. Defaults to 5.
-            second_per_frame (int, optional): Second per frame. Defaults to 2.
-            return_format (str, optional): Return format. Defaults to "npndarray".
-
-        Returns:
-        any: Video frames.
-        """
-        frames = list()
-        frame_counter = 0
-        frame_per_sec = int(round(self.original_vidoe_fps)) * second_per_frame
-        while self._cap.isOpened():
-            ret, frame_img = self._cap.read()
-            if not ret:
-                break
-            if frame_counter % frame_per_sec == 0:
-                if frame_scale is not None:
-                    frame_img = self._resize_frame(frame_img, frame_scale)
-
-                frames.append(frame_img)
-            if cv2.waitKey(1) & 0xFF == ord("q"):  # on press of q break
-                break
-            frame_counter += 1
-        self._cap.release()
-        cv2.destroyAllWindows()
-        if return_format == "npndarray":
-            self._processed_frames = np.array(frames)
-            return np.array(frames)
-        else:
-            self._processed_frames = frames
-            return frames
-
     def replay_and_save(
         self,
         frames: list[Frame],
         frame_rate=2,
         is_imshow: bool = False,
         output_dir: str | None = None,
-    ):
+    ) -> None:
+        """Replay and save frames.
+
+        Args:
+            frames (list[Frame]): List of frames.
+            frame_rate (int, optional): Frame rate. Defaults to 2.
+            is_imshow (bool, optional): Show image. Defaults to False.
+            output_dir (str | None, optional): Output directory. Defaults to None.
+        """
         if output_dir is None:
             output_dir = os.path.join(self._artifact_dir, "replay_frames")
 
@@ -247,16 +262,3 @@ def replay_and_save(
                 int(1000 / frame_rate)
             )  # wait for specified milliseconds
         cv2.destroyAllWindows
-
-    @property
-    def original_video_num_of_frames(self):
-        """Get number of frames in video."""
-        return self.original_frame_count
-
-    @property
-    def number_of_frames(self):
-        """Get number of frames in video."""
-        if self._processed_frames is None:
-            return self.original_frame_count
-        else:
-            return len(self._processed_frames)