drone input and output

__init__.py

@@ -0,0 +1,3 @@
+"""
+Drone and video input module for ParcelMapper.
+"""

test_video.py

@@ -0,0 +1,147 @@
+from pathlib import Path
+
+import cv2
+
+from drone.video import (
+    get_video_info,
+    read_sampled_frames,
+)
+
+
+VIDEO_PATH = Path(
+    "data/input/drone_demo.mp4"
+)
+
+OUTPUT_FOLDER = Path(
+    "data/output/sample_frames"
+)
+
+EVERY_N_FRAMES = 15
+
+NUMBER_OF_TEST_FRAMES = 5
+
+
+def main():
+
+    print()
+    print("ParcelMapper - Drone Video Test")
+    print("--------------------------------")
+
+    if not VIDEO_PATH.exists():
+
+        print(
+            f"ERROR: Video not found at "
+            f"{VIDEO_PATH}"
+        )
+
+        print(
+            "Put an MP4 video inside "
+            "data/input/ and rename it "
+            "drone_demo.mp4"
+        )
+
+        return
+
+    print()
+    print("Video found!")
+
+    info = get_video_info(VIDEO_PATH)
+
+    print()
+    print("VIDEO INFORMATION")
+    print("-----------------")
+
+    print(
+        f"Resolution: "
+        f"{info['width']} x {info['height']}"
+    )
+
+    print(
+        f"FPS: "
+        f"{info['fps']:.2f}"
+    )
+
+    print(
+        f"Total frames: "
+        f"{info['total_frames']}"
+    )
+
+    print(
+        f"Duration: "
+        f"{info['duration_seconds']:.2f} seconds"
+    )
+
+    OUTPUT_FOLDER.mkdir(
+        parents=True,
+        exist_ok=True
+    )
+
+    print()
+    print(
+        f"Reading every "
+        f"{EVERY_N_FRAMES}th frame..."
+    )
+
+    saved_frames = 0
+
+    for frame_number, frame in read_sampled_frames(
+        VIDEO_PATH,
+        every_n=EVERY_N_FRAMES
+    ):
+
+        height, width = frame.shape[:2]
+
+        print(
+            f"Frame {frame_number}: "
+            f"{width} x {height}"
+        )
+
+        output_file = (
+            OUTPUT_FOLDER
+            / f"frame_{frame_number}.jpg"
+        )
+
+        success = cv2.imwrite(
+            str(output_file),
+            frame
+        )
+
+        if success:
+
+            print(
+                f"Saved -> {output_file}"
+            )
+
+        else:
+
+            print(
+                f"Could not save "
+                f"{output_file}"
+            )
+
+        saved_frames += 1
+
+        if saved_frames >= NUMBER_OF_TEST_FRAMES:
+            break
+
+    print()
+    print("--------------------------------")
+
+    if saved_frames == 0:
+
+        print(
+            "No frames were extracted."
+        )
+
+    else:
+
+        print(
+            f"SUCCESS: Extracted "
+            f"{saved_frames} test frames."
+        )
+
+    print("--------------------------------")
+
+
+if __name__ == "__main__":
+    main()

video.py

@@ -0,0 +1,110 @@
+import cv2
+
+
+def read_video(path):
+    """
+    Open a video file and return its frames one by one.
+
+    Parameters
+    ----------
+    path : str
+        Path to the video file.
+
+    Yields
+    ------
+    frame
+        One OpenCV image/frame at a time.
+    """
+
+    cap = cv2.VideoCapture(str(path))
+
+    if not cap.isOpened():
+        raise FileNotFoundError(
+            f"Could not open video: {path}"
+        )
+
+    try:
+        while True:
+
+            success, frame = cap.read()
+
+            if not success:
+                break
+
+            yield frame
+
+    finally:
+        cap.release()
+
+
+def read_sampled_frames(path, every_n=15):
+    """
+    Read only every Nth frame from the video.
+
+    Example:
+    every_n=15 means:
+    frame 15
+    frame 30
+    frame 45
+    ...
+
+    This avoids sending every video frame to the AI.
+    """
+
+    if every_n < 1:
+        raise ValueError(
+            "every_n must be at least 1"
+        )
+
+    frame_number = 0
+
+    for frame in read_video(path):
+
+        frame_number += 1
+
+        if frame_number % every_n != 0:
+            continue
+
+        yield frame_number, frame
+
+
+def get_video_info(path):
+    """
+    Read useful information about a video.
+    """
+
+    cap = cv2.VideoCapture(str(path))
+
+    if not cap.isOpened():
+        raise FileNotFoundError(
+            f"Could not open video: {path}"
+        )
+
+    fps = cap.get(cv2.CAP_PROP_FPS)
+
+    total_frames = int(
+        cap.get(cv2.CAP_PROP_FRAME_COUNT)
+    )
+
+    width = int(
+        cap.get(cv2.CAP_PROP_FRAME_WIDTH)
+    )
+
+    height = int(
+        cap.get(cv2.CAP_PROP_FRAME_HEIGHT)
+    )
+
+    cap.release()
+
+    if fps > 0:
+        duration_seconds = total_frames / fps
+    else:
+        duration_seconds = 0
+
+    return {
+        "fps": fps,
+        "total_frames": total_frames,
+        "width": width,
+        "height": height,
+        "duration_seconds": duration_seconds,
+    }