Merge pull request #18 from swhan0329/refactor/remove-legacy-video-helper

Refactor input capture and harden output writer fallback

Author: swhan0329

app/io/video_source.py

@@ -1,14 +1,29 @@
 from __future__ import annotations
 
+from pathlib import Path
+
 import cv2 as cv
 
-import video
+
+def _resolve_capture_source(video_src: str) -> int | str:
+    raw = str(video_src).strip()
+    path_candidate = Path(raw)
+
+    # Keep existing files as path inputs, even if the filename is numeric.
+    if path_candidate.exists():
+        return raw
+
+    if raw.lstrip("+-").isdigit():
+        return int(raw)
+
+    return raw
 
 
 def open_video_source(video_src: str) -> cv.VideoCapture:
-    """Open a source path or camera index using the existing helper."""
+    """Open a source path or camera index with OpenCV VideoCapture."""
 
-    return video.create_capture(video_src)
+    source = _resolve_capture_source(video_src)
+    return cv.VideoCapture(source)
 
 
 def resolve_fps(capture: cv.VideoCapture, fallback_fps: float) -> float:

app/pipeline.py

@@ -50,14 +50,31 @@ def _create_writer(
     if not output_path:
         return None
 
-    fourcc = cv.VideoWriter_fourcc(
-        *("mp4v" if output_path.lower().endswith(".mp4") else "XVID")
+    output = Path(output_path)
+    output.parent.mkdir(parents=True, exist_ok=True)
+
+    if output.suffix.lower() == ".mp4":
+        codec_candidates = ["mp4v", "avc1", "MJPG", "XVID"]
+    else:
+        codec_candidates = ["XVID", "MJPG", "mp4v", "avc1"]
+
+    attempted: list[str] = []
+    for codec in codec_candidates:
+        attempted.append(codec)
+        fourcc = cv.VideoWriter_fourcc(*codec)
+        writer = cv.VideoWriter(str(output), fourcc, fps, frame_size)
+        if writer.isOpened():
+            if len(attempted) > 1:
+                print(f"Output writer fallback succeeded with codec {codec}")
+            return writer
+        writer.release()
+
+    attempted_str = ", ".join(attempted)
+    print(
+        "Warning: unable to open output file: "
+        f"{output_path} (attempted codecs: {attempted_str})"
     )
-    writer = cv.VideoWriter(output_path, fourcc, fps, frame_size)
-    if not writer.isOpened():
-        print(f"Warning: unable to open output file: {output_path}")
-        return None
-    return writer
+    return None
 
 
 def run_pipeline(

tests/test_video_source.py

@@ -0,0 +1,44 @@
+import os
+import tempfile
+import unittest
+from pathlib import Path
+from unittest.mock import patch
+
+from app.io.video_source import _resolve_capture_source, open_video_source
+
+
+class TestVideoSource(unittest.TestCase):
+    def test_resolve_capture_source_uses_camera_index_for_numeric_string(self):
+        self.assertEqual(_resolve_capture_source("0"), 0)
+        self.assertEqual(_resolve_capture_source("-1"), -1)
+
+    def test_resolve_capture_source_keeps_existing_numeric_filename(self):
+        with tempfile.TemporaryDirectory() as temp_dir:
+            file_path = Path(temp_dir) / "0"
+            file_path.write_text("dummy", encoding="utf-8")
+
+            previous_cwd = os.getcwd()
+            try:
+                os.chdir(temp_dir)
+                self.assertEqual(_resolve_capture_source("0"), "0")
+            finally:
+                os.chdir(previous_cwd)
+
+    def test_resolve_capture_source_keeps_non_numeric_source(self):
+        self.assertEqual(
+            _resolve_capture_source("samples/MNn9qKG2UFI_10s.mp4"),
+            "samples/MNn9qKG2UFI_10s.mp4",
+        )
+        self.assertEqual(
+            _resolve_capture_source("rtsp://127.0.0.1:8554/stream"),
+            "rtsp://127.0.0.1:8554/stream",
+        )
+
+    def test_open_video_source_passes_resolved_source(self):
+        with patch("app.io.video_source.cv.VideoCapture") as mock_capture:
+            open_video_source("2")
+            mock_capture.assert_called_once_with(2)
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_writer_creation.py

@@ -0,0 +1,68 @@
+import tempfile
+import unittest
+from pathlib import Path
+from unittest.mock import patch
+
+from app.pipeline import _create_writer
+
+
+class _FakeWriter:
+    def __init__(self, opened: bool) -> None:
+        self._opened = opened
+        self.released = False
+
+    def isOpened(self) -> bool:
+        return self._opened
+
+    def release(self) -> None:
+        self.released = True
+
+
+class TestWriterCreation(unittest.TestCase):
+    def test_create_writer_falls_back_and_creates_parent_directory(self):
+        with tempfile.TemporaryDirectory() as temp_dir:
+            output_path = Path(temp_dir) / "nested" / "out.mp4"
+            writers = [_FakeWriter(False), _FakeWriter(False), _FakeWriter(True)]
+            calls: list[tuple[str, int, float, tuple[int, int]]] = []
+
+            def fake_video_writer(
+                path: str,
+                fourcc: int,
+                fps: float,
+                size: tuple[int, int],
+            ) -> _FakeWriter:
+                calls.append((path, fourcc, fps, size))
+                return writers[len(calls) - 1]
+
+            with (
+                patch("app.pipeline.cv.VideoWriter_fourcc", side_effect=[1, 2, 3]),
+                patch("app.pipeline.cv.VideoWriter", side_effect=fake_video_writer),
+            ):
+                writer = _create_writer(str(output_path), (1280, 720), 30.0)
+
+            self.assertIs(writer, writers[2])
+            self.assertEqual(len(calls), 3)
+            self.assertTrue(writers[0].released)
+            self.assertTrue(writers[1].released)
+            self.assertTrue(output_path.parent.exists())
+
+    def test_create_writer_returns_none_when_all_codecs_fail(self):
+        with tempfile.TemporaryDirectory() as temp_dir:
+            output_path = Path(temp_dir) / "out.mp4"
+            writers = [_FakeWriter(False), _FakeWriter(False), _FakeWriter(False), _FakeWriter(False)]
+
+            with (
+                patch("app.pipeline.cv.VideoWriter_fourcc", side_effect=[1, 2, 3, 4]),
+                patch("app.pipeline.cv.VideoWriter", side_effect=writers),
+                patch("builtins.print") as mock_print,
+            ):
+                writer = _create_writer(str(output_path), (640, 480), 30.0)
+
+            self.assertIsNone(writer)
+            self.assertTrue(all(item.released for item in writers))
+            printed = " ".join(str(arg) for arg in mock_print.call_args_list[-1][0])
+            self.assertIn("attempted codecs", printed)
+
+
+if __name__ == "__main__":
+    unittest.main()