feat: frame window to have states and transitions

Author: minkyu-choi07

main.py

@@ -24,11 +24,10 @@
             video_path=sample_video_path,
             artifact_dir=config.VERSION_AND_PATH.ARTIFACTS_PATH,
         ),
-        proposition_set=["face", "baby"],
+        proposition_set=["face", "person", "car"],
+        artifact_dir=config.VERSION_AND_PATH.ARTIFACTS_PATH,
     )
-    states, transitions, prev_states = frame2automaton.build_automaton(
-        is_annotation=False
-    )
-    print(states)
-    print(transitions)
+
+    frame_window_automata = frame2automaton.start()
+
     print("Development is in progress.")

ns_vfs/data/frame.py

@@ -14,3 +14,28 @@ class Frame:
 class FrameWindow:
     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)
+
+    def get_propositional_confidence(self):
+        propositional_confidence = [
+            []
+            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]
+                )
+                idx += 1
+        self.propositional_confidence = propositional_confidence
+        return propositional_confidence
+
+    def update_states(self, states):
+        self.states = states
+        return states

ns_vfs/processor/video_processor.py

@@ -9,6 +9,7 @@
 import numpy as np
 
 from ns_vfs.data.frame import Frame, FrameWindow
+from ns_vfs.verification import build_label_func, build_trans_matrix
 
 
 class VideoProcessor(abc.ABC):
@@ -50,20 +51,22 @@ def _resize_frame(self, frame_img, frame_scale):
             ),
         )
 
-    def get_synchronous_frame(
+    def build_frame_synchronously(
         self,
         proposition_set: list,
         calculate_propositional_confidence: callable,
+        build_automaton: callable,
         frame_duration_sec: int = 2,
         frame_scale: int | None = None,
         sliding_window_size: int = 5,
+        save_image: bool = False,
     ):
         frame_window_idx = 0
         frame_counter = 0
         frame_idx = 0
         # Calculate the frame skip rate
         frame_window = dict(frame_window_idx=0)
-        frame_set = list()
+        temp_frame_set = list()
 
         while self._cap.isOpened():
             ret, frame_img = self._cap.read()
@@ -84,21 +87,50 @@ def get_synchronous_frame(
                 )
                 # Calculate propositional confidence
                 for proposition in proposition_set:
+                    propositional_confidence = (
+                        calculate_propositional_confidence(
+                            proposition=proposition,
+                            frame_img=frame_img,
+                            is_annotation=save_image,
+                        )
+                    )
                     frame.propositional_probability[
                         str(proposition)
-                    ] = calculate_propositional_confidence(
-                        proposition=proposition, frame_img=frame_img
-                    )
-                # Build State
+                    ] = propositional_confidence
+                temp_frame_set.append(frame)
+
+                if len(temp_frame_set) == sliding_window_size:
+                    if save_image:
+                        self.replay_and_save(frames=temp_frame_set)
 
-                frame_set.append(frame)
-                if len(frame_set) == sliding_window_size:
+                    # Process frame window
+                    frame_set = temp_frame_set.copy()
                     frame_window[frame_window_idx] = FrameWindow(
                         frame_window_idx=frame_window_idx,
-                        frame_image_set=frame_set.copy(),
+                        frame_image_set=frame_set,
                     )
+
+                    propositional_confidence = frame_window[
+                        frame_window_idx
+                    ].get_propositional_confidence()
+
+                    # Build State & Compute Probability
+                    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
+                    )
+                    state_labeling = build_label_func(states, proposition_set)
+                    transition_matrix
+                    state_labeling
+
                     frame_window_idx += 1
-                    frame_set.pop(0)
+                    temp_frame_set.pop(0)
 
                 frame_idx += 1
             frame_counter += 1
@@ -118,7 +150,7 @@ def get_frame_by_sliding_window(
         frame_idx = 0
         # Calculate the frame skip rate
         frame_window = dict(frame_window_idx=0)
-        frame_set = list()
+        temp_frame_set = list()
 
         while self._cap.isOpened():
             ret, frame_img = self._cap.read()
@@ -132,16 +164,16 @@ def get_frame_by_sliding_window(
                 if frame_scale is not None:
                     frame_img = self._resize_frame(frame_img, frame_scale)
                 Frame
-                frame_set.append(
+                temp_frame_set.append(
                     Frame(frame_index=frame_idx, frame_image=frame_img)
                 )
-                if len(frame_set) == sliding_window_size:
+                if len(temp_frame_set) == sliding_window_size:
                     frame_window[frame_window_idx] = FrameWindow(
                         frame_window_idx=frame_window_idx,
-                        frame_image_set=frame_set.copy(),
+                        frame_image_set=temp_frame_set.copy(),
                     )
                     frame_window_idx += 1
-                    frame_set.pop(0)
+                    temp_frame_set.pop(0)
 
                 frame_idx += 1
             frame_counter += 1
@@ -192,7 +224,7 @@ def get_frame(
 
     def replay_and_save(
         self,
-        frames: list[np.ndarray],
+        frames: list[Frame],
         frame_rate=2,
         is_imshow: bool = False,
         output_dir: str | None = None,
@@ -206,7 +238,7 @@ def replay_and_save(
             shutil.rmtree(output_dir)
             os.makedirs(output_dir)
         for idx, frame in enumerate(frames):
-            plt.imshow(cv2.cvtColor(frame, cv2.COLOR_BGR2RGB))
+            plt.imshow(cv2.cvtColor(frame.frame_image, cv2.COLOR_BGR2RGB))
             plt.axis("off")  # hide the axis values
             plt.savefig(os.path.join(output_dir, f"frame_{idx}.png"))
             if is_imshow:

ns_vfs/state.py

@@ -6,34 +6,34 @@ def __init__(
         self.state_index = index
         self.frame_index = frame_index
         self.proposition_set = proposition_set
-        self.proposition_status_set = (
-            proposition_status_set  # TTT, TFT, FTT, etc.
+        self.current_proposition_combination = (
+            proposition_status_set  # "initial", TTT, TFT, FTT, etc.
         )
-        self.proposition_true_label_list = self._build_label_list(
+        self.current_descriptive_label = self._get_descriptive_label(
             label=proposition_status_set
         )
         self.probability = 1
 
     def __repr__(self):
-        return f"{self.state_index} {self.proposition_true_label_list} {self.frame_index} {self.probability}"
+        return f"{self.state_index} {self.current_descriptive_label} {self.frame_index} {self.probability}"
 
     def __str__(self):
         return f"{self.__repr__()}"
 
-    def _build_label_list(self, label):
+    def _get_descriptive_label(self, label):
         labels = []
         for i in range(len(self.proposition_set)):
             if label[i] == "T":
                 labels.append(self.proposition_set[i])
         return labels
 
-    def update(self, frame_index, proposition_status_set):
+    def update(self, frame_index, proposition_combinations):
         self.frame_index = frame_index
-        self.proposition_status_set = (
-            proposition_status_set  # TTT, TFT, FTT, etc.
+        self.current_proposition_combination = (
+            proposition_combinations  # TTT, TFT, FTT, etc.
         )
-        self.proposition_true_label_list = self._build_label_list(
-            label=proposition_status_set
+        self.current_descriptive_label = self._get_descriptive_label(
+            label=proposition_combinations
         )
         self.probability = 1
 
@@ -46,8 +46,8 @@ def compute_probability(self, probabilities):
                 -> [[0.1, 0.2, 0.3], [0.4, 0.5, 0.6]].
         """
         probability = 1
-        for i in range(len(self.proposition_status_set)):
-            if self.proposition_status_set[i] == "T":
+        for i in range(len(self.current_proposition_combination)):
+            if self.current_proposition_combination[i] == "T":
                 probability *= probabilities[i][self.frame_index]
             else:
                 probability *= 1 - probabilities[i][self.frame_index]