diff --git a/babyping.py b/babyping.py index 1a1affa..0b364ae 100644 --- a/babyping.py +++ b/babyping.py @@ -186,11 +186,32 @@ def select_roi(cap): return roi -def open_camera(index): +def try_open_camera(index): + """Try to open a camera. Returns the VideoCapture or None if unavailable.""" cap = cv2.VideoCapture(index) - if not cap.isOpened(): - cap = cv2.VideoCapture(index, cv2.CAP_AVFOUNDATION) - if not cap.isOpened(): + if cap.isOpened(): + return cap + cap = cv2.VideoCapture(index, cv2.CAP_AVFOUNDATION) + if cap.isOpened(): + return cap + return None + + +def reconnect_camera(index, max_attempts=10, base_delay=2.0): + """Retry opening camera with exponential backoff. Returns cap or None.""" + for attempt in range(max_attempts): + delay = min(base_delay * (2 ** attempt), 60) + print(f" Reconnect attempt {attempt + 1}/{max_attempts} (waiting {delay:.0f}s)...") + time.sleep(delay) + cap = try_open_camera(index) + if cap is not None: + return cap + return None + + +def open_camera(index): + cap = try_open_camera(index) + if cap is None: print(f"Error: Could not open camera at index {index}") sys.exit(1) return cap @@ -231,7 +252,7 @@ def main(): prev_gray = None last_alert_time = 0 consecutive_failures = 0 - max_retries = 3 + max_frame_failures = 30 try: while True: @@ -240,15 +261,21 @@ def main(): ret, frame = cap.read() if not ret: consecutive_failures += 1 - print(f"Warning: Failed to read frame ({consecutive_failures}/{max_retries})") - if consecutive_failures >= max_retries: - print("Error: Camera disconnected. Attempting to reconnect...") + if consecutive_failures == 1: + print("Warning: Dropped frame — camera may be intermittent") + if consecutive_failures >= max_frame_failures: + print(f"Camera lost after {consecutive_failures} dropped frames. Reconnecting...") + send_notification("BabyPing", "Camera disconnected — reconnecting...") cap.release() - time.sleep(2) - cap = open_camera(args.camera) + cap = reconnect_camera(args.camera) + if cap is None: + print("Error: Could not reconnect to camera. Exiting.") + send_notification("BabyPing", "Camera reconnect failed — stopping") + break consecutive_failures = 0 prev_gray = None - print("Reconnected.") + print("Reconnected to camera.") + send_notification("BabyPing", "Camera reconnected") continue consecutive_failures = 0 diff --git a/tests/test_babyping.py b/tests/test_babyping.py index fd13230..c33a1f1 100644 --- a/tests/test_babyping.py +++ b/tests/test_babyping.py @@ -9,7 +9,9 @@ import time sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) -from babyping import apply_night_mode, crop_to_roi, detect_motion, FrameBuffer, offset_contours, parse_args, parse_roi_string, save_snapshot, throttle_fps, SENSITIVITY_THRESHOLDS +from unittest.mock import MagicMock, patch + +from babyping import apply_night_mode, crop_to_roi, detect_motion, FrameBuffer, offset_contours, parse_args, parse_roi_string, reconnect_camera, save_snapshot, throttle_fps, try_open_camera, SENSITIVITY_THRESHOLDS # --- detect_motion tests --- @@ -335,3 +337,72 @@ def test_fps_zero_means_no_throttle(self): throttle_fps(frame_start, 0) after = time.monotonic() assert (after - before) < 0.01 + + +# --- try_open_camera tests --- + +class TestTryOpenCamera: + def test_returns_cap_when_camera_available(self): + mock_cap = MagicMock() + mock_cap.isOpened.return_value = True + with patch("babyping.cv2.VideoCapture", return_value=mock_cap): + result = try_open_camera(0) + assert result is mock_cap + + def test_returns_none_when_camera_unavailable(self): + mock_cap = MagicMock() + mock_cap.isOpened.return_value = False + with patch("babyping.cv2.VideoCapture", return_value=mock_cap): + result = try_open_camera(0) + assert result is None + + def test_tries_avfoundation_fallback(self): + mock_cap_fail = MagicMock() + mock_cap_fail.isOpened.return_value = False + mock_cap_ok = MagicMock() + mock_cap_ok.isOpened.return_value = True + + with patch("babyping.cv2.VideoCapture", side_effect=[mock_cap_fail, mock_cap_ok]) as mock_vc: + result = try_open_camera(0) + assert result is mock_cap_ok + assert mock_vc.call_count == 2 + + +# --- reconnect_camera tests --- + +class TestReconnectCamera: + def test_returns_cap_on_first_try(self): + mock_cap = MagicMock() + mock_cap.isOpened.return_value = True + with patch("babyping.try_open_camera", return_value=mock_cap): + result = reconnect_camera(0, max_attempts=3, base_delay=0.01) + assert result is mock_cap + + def test_retries_with_backoff_then_succeeds(self): + mock_cap = MagicMock() + mock_cap.isOpened.return_value = True + with patch("babyping.try_open_camera", side_effect=[None, None, mock_cap]) as mock_try: + result = reconnect_camera(0, max_attempts=5, base_delay=0.01) + assert result is mock_cap + assert mock_try.call_count == 3 + + def test_returns_none_after_max_attempts(self): + with patch("babyping.try_open_camera", return_value=None): + result = reconnect_camera(0, max_attempts=3, base_delay=0.01) + assert result is None + + def test_backoff_increases_wait_time(self): + with patch("babyping.try_open_camera", return_value=None), \ + patch("babyping.time.sleep") as mock_sleep: + reconnect_camera(0, max_attempts=4, base_delay=1.0) + delays = [call.args[0] for call in mock_sleep.call_args_list] + # Exponential: 1, 2, 4, 8 (capped at 60) + assert delays == [1.0, 2.0, 4.0, 8.0] + + def test_backoff_caps_at_60_seconds(self): + with patch("babyping.try_open_camera", return_value=None), \ + patch("babyping.time.sleep") as mock_sleep: + reconnect_camera(0, max_attempts=8, base_delay=10.0) + delays = [call.args[0] for call in mock_sleep.call_args_list] + # 10, 20, 40, 60, 60, 60, 60, 60 + assert all(d <= 60 for d in delays)