Merge pull request #13 from Y-B-Class-Projects/add-ci-test

Add ci test

Author: Ian-I-Chien

.github/workflows/ci.yaml

@@ -0,0 +1,39 @@
+name: Code Quality and Tests
+
+on:
+  push:
+    branches:
+      - main
+      - '**'
+  pull_request:
+    branches:
+      - main
+
+jobs:
+  pr_checks:
+    runs-on: ubuntu-latest
+
+    steps:
+      - name: Checkout code
+        uses: actions/checkout@v2
+
+      - name: Set up Python
+        uses: actions/setup-python@v2
+        with:
+          python-version: '3.12'
+
+      - name: Install dependencies
+        run: |
+          pip install -r requirements.txt
+
+      - name: Download YOLOv7 pose model
+        run: |
+          curl -L -o yolov7-w6-pose.pt https://github.com/WongKinYiu/yolov7/releases/download/v0.1/yolov7-w6-pose.pt
+
+      - name: Check Python script for errors
+        run: |
+          CI_MODE=1 python video.py
+
+      - name: Run Black to check code style
+        run: |
+          black --check .

.gitignore

@@ -28,6 +28,8 @@ data/images/*
 !data/images/zidane.jpg
 !data/images/bus.jpg
 !data/*.sh
+!fall_dataset/ci_videos/**
+!.github/workflows/ci.yaml
 
 results*.csv
 

config.py

@@ -4,13 +4,15 @@
 
 load_dotenv()
 
+
 def get_env_int(var_name, default):
     try:
         return int(os.getenv(var_name, default))
     except ValueError:
         print(f"Warning: {var_name} is not a valid int, using default {default}")
         return default
 
+
 def get_env_float(var_name, default):
     try:
         return float(os.getenv(var_name, default))

fall_core.py

@@ -24,7 +24,7 @@ def compute_center_velocity(pose_start, pose_end, fps, window_size):
     com_end = compute_center_of_mass(pose_end)
     dx = com_end[0] - com_start[0]
     dy = com_end[1] - com_start[1]
-    distance = math.sqrt(dx ** 2 + dy ** 2)
+    distance = math.sqrt(dx**2 + dy**2)
     time_elapsed = (window_size - 1) / fps
     velocity = distance / time_elapsed
     return min(velocity, 300.0), dy
@@ -50,7 +50,7 @@ def find_most_similar_pose(reference_pose, candidate_poses):
     best_pose = None
     for pose in candidate_poses:
         cx, cy = compute_center_of_mass(pose)
-        dist = (ref_cx - cx)**2 + (ref_cy - cy)**2
+        dist = (ref_cx - cx) ** 2 + (ref_cy - cy) ** 2
         if dist < min_dist:
             min_dist = dist
             best_pose = pose
@@ -101,7 +101,7 @@ def prepare_vid_out(video_path, vid_cap, output_dir):
 
     vid_write_image = letterbox(first_frame, 960, stride=64, auto=True)[0]
     resize_height, resize_width = vid_write_image.shape[:2]
-    video_name = os.path.basename(video_path).split('.')[0] + "_output.mp4"
+    video_name = os.path.basename(video_path).split(".")[0] + "_output.mp4"
     out_video_name = os.path.join(output_dir, video_name)
     print(f"[INFO] Writing output to: {out_video_name}")
 
@@ -114,7 +114,9 @@ def prepare_vid_out(video_path, vid_cap, output_dir):
     return out
 
 
-def fall_detection(pose_window, window_size, fps, v_thresh, aspect_ratio_thresh, dy_thresh):
+def fall_detection(
+    pose_window, window_size, fps, v_thresh, aspect_ratio_thresh, dy_thresh
+):
     if len(pose_window) < window_size:
         return False, None, None, None
 
@@ -131,14 +133,14 @@ def fall_detection(pose_window, window_size, fps, v_thresh, aspect_ratio_thresh,
         f"ar={ar_delta:.2f}/{aspect_ratio_thresh:.2f}"
     )
     print(f"[TRACE] {debug_text}")
-    
+
     cond_speed_drop = v > v_thresh and dy > dy_thresh
     cond_down_flat = dy > dy_thresh and ar_delta > aspect_ratio_thresh
 
     if cond_speed_drop or cond_down_flat:
         tag = (
-            ("SpeedDrop " if cond_speed_drop else "") +
-            ("DownFlat " if cond_down_flat else "")
+            ("SpeedDrop " if cond_speed_drop else "")
+            + ("DownFlat " if cond_down_flat else "")
         ).strip()
 
         xmin = pose_end[2] - pose_end[4] / 2
@@ -171,8 +173,10 @@ def falling_alarm(image, bbox):
         lineType=cv2.LINE_AA,
     )
 
+
 def draw_fps(frame, prev_time):
     import time
+
     curr_time = time.time()
     fps = 1 / (curr_time - prev_time)
     cv2.putText(
@@ -186,4 +190,3 @@ def draw_fps(frame, prev_time):
         cv2.LINE_AA,
     )
     return frame, curr_time
-

fall_dataset/ci_videos/video_1.mp4

@@ -1,6 +1,10 @@
 from fall_core import (
-    get_pose_model, get_pose, prepare_image,
-    fall_detection, falling_alarm, draw_fps,
+    get_pose_model,
+    get_pose,
+    prepare_image,
+    fall_detection,
+    falling_alarm,
+    draw_fps,
 )
 import cv2
 import time
@@ -32,17 +36,22 @@ def process_realtime_camera():
                     FPS,
                     V_THRESH,
                     ASPECT_RATIO_THRESH,
-                    DY_THRESH
+                    DY_THRESH,
                 )
                 if debug_text:
                     _image = cv2.putText(
-                        _image,  f"{tag}: {debug_text}", (10, 30), cv2.FONT_HERSHEY_SIMPLEX,
-                        0.7, (0, 255, 0), 2
+                        _image,
+                        f"{tag}: {debug_text}",
+                        (10, 30),
+                        cv2.FONT_HERSHEY_SIMPLEX,
+                        0.7,
+                        (0, 255, 0),
+                        2,
                     )
                 if is_fall:
                     falling_alarm(_image, bbox)
 
-        cv2.imshow("Real-Time Fall Detection", _image[:,:,::-1])
+        cv2.imshow("Real-Time Fall Detection", _image[:, :, ::-1])
         if cv2.waitKey(1) & 0xFF == ord("q"):
             break
 

realtime.py

@@ -41,4 +41,6 @@ thop~=0.1.1.post2209072238
 # roboflow
 
 telepot~=12.7
-torchmetrics~=0.10.2
+torchmetrics~=0.10.2
+
+black==24.8.0

requirements.txt

@@ -1,6 +1,10 @@
 from fall_core import (
-    get_pose_model, get_pose, prepare_image, fall_detection,
-    falling_alarm, prepare_vid_out
+    get_pose_model,
+    get_pose,
+    prepare_image,
+    fall_detection,
+    falling_alarm,
+    prepare_vid_out,
 )
 import cv2
 import os
@@ -31,29 +35,38 @@ def process_video_file(video_path, output_dir):
         if len(output) > 0:
             pose_window.append(output)
             if len(pose_window) == WINDOW_SIZE:
-                is_fall, bbox, debug_text, tag= fall_detection(
+                is_fall, bbox, debug_text, tag = fall_detection(
                     pose_window,
                     WINDOW_SIZE,
                     FPS,
                     V_THRESH,
                     ASPECT_RATIO_THRESH,
-                    DY_THRESH
+                    DY_THRESH,
                 )
                 # debug
                 _image = cv2.putText(
-                    _image,  f"{tag}: {debug_text}", (10, 30), cv2.FONT_HERSHEY_SIMPLEX,
-                    0.7, (0, 255, 0), 2
+                    _image,
+                    f"{tag}: {debug_text}",
+                    (10, 30),
+                    cv2.FONT_HERSHEY_SIMPLEX,
+                    0.7,
+                    (0, 255, 0),
+                    2,
                 )
                 if is_fall:
                     falling_alarm(_image, bbox)
-        vid_out.write(_image[:,:,::-1])
+        vid_out.write(_image[:, :, ::-1])
 
     vid_out.release()
     vid_cap.release()
 
 
 if __name__ == "__main__":
-    videos_path = "fall_dataset/videos"
+    if os.environ.get("CI_MODE") == "1":
+        videos_path = "fall_dataset/ci_videos"
+        print("[CI MODE] Only running on CI test videos...")
+    else:
+        videos_path = "fall_dataset/videos"
     output_dir = "output_videos"
     os.makedirs(output_dir, exist_ok=True)