From bf01babc94b02516042b81d597e498768e57fe69 Mon Sep 17 00:00:00 2001 From: Mateo Date: Sat, 27 Jul 2024 07:06:19 +0200 Subject: [PATCH] mesure for each cam (#221) * mesure for each cam * fix test * use not gray image --- pyroengine/core.py | 14 +------------- tests/test_core.py | 35 ++++++++--------------------------- 2 files changed, 9 insertions(+), 40 deletions(-) diff --git a/pyroengine/core.py b/pyroengine/core.py index 68f215c..182d241 100644 --- a/pyroengine/core.py +++ b/pyroengine/core.py @@ -75,7 +75,7 @@ async def capture_camera_image(camera: ReolinkCamera, image_queue: asyncio.Queue # Move camera to the next pose to avoid waiting next_pos_id = camera.cam_poses[(idx + 1) % len(camera.cam_poses)] camera.move_camera("ToPos", idx=int(next_pos_id), speed=50) - if frame is not None: + if frame is not None and is_day_time(None, frame, "ir"): await image_queue.put((cam_id, frame)) await asyncio.sleep(0) # Yield control else: @@ -137,17 +137,6 @@ async def analyze_stream(self, image_queue: asyncio.Queue) -> None: finally: image_queue.task_done() # Mark the task as done - def check_day_time(self) -> None: - """ - Checks and updates the day_time attribute based on the current frame. - """ - try: - frame = self.cameras[0].capture() - if frame is not None: - self.day_time = is_day_time(None, frame, "ir") - except Exception as e: - logging.exception(f"Exception during initial day time check: {e}") - async def run(self, period: int = 30, send_alerts: bool = True) -> None: """ Captures and analyzes all camera streams, then processes alerts. @@ -157,7 +146,6 @@ async def run(self, period: int = 30, send_alerts: bool = True) -> None: send_alerts (bool): Boolean to activate / deactivate alert sending """ try: - self.check_day_time() if self.day_time: image_queue: asyncio.Queue[Any] = asyncio.Queue() diff --git a/tests/test_core.py b/tests/test_core.py index 071fe8a..ef1c4f2 100644 --- a/tests/test_core.py +++ b/tests/test_core.py @@ -16,18 +16,18 @@ def mock_engine(): @pytest.fixture -def mock_cameras(): +def mock_cameras(mock_wildfire_image): camera = MagicMock() - camera.capture.return_value = Image.new("RGB", (100, 100)) # Mock captured image + camera.capture.return_value = mock_wildfire_image # Mock captured image camera.cam_type = "static" camera.ip_address = "192.168.1.1" return [camera] @pytest.fixture -def mock_cameras_ptz(): +def mock_cameras_ptz(mock_wildfire_image): camera = MagicMock() - camera.capture.return_value = Image.new("RGB", (100, 100)) # Mock captured image + camera.capture.return_value = mock_wildfire_image # Mock captured image camera.cam_type = "ptz" camera.cam_poses = [1, 2] camera.ip_address = "192.168.1.1" @@ -94,9 +94,9 @@ async def test_capture_images_ptz(system_controller_ptz): @pytest.mark.asyncio -async def test_analyze_stream(system_controller): +async def test_analyze_stream(system_controller, mock_wildfire_image): queue = asyncio.Queue() - mock_frame = Image.new("RGB", (100, 100)) + mock_frame = mock_wildfire_image await queue.put(("192.168.1.1", mock_frame)) analyze_task = asyncio.create_task(system_controller.analyze_stream(queue)) @@ -118,9 +118,9 @@ async def test_capture_images_method(system_controller): @pytest.mark.asyncio -async def test_analyze_stream_method(system_controller): +async def test_analyze_stream_method(system_controller, mock_wildfire_image): queue = asyncio.Queue() - mock_frame = Image.new("RGB", (100, 100)) + mock_frame = mock_wildfire_image await queue.put(("192.168.1.1", mock_frame)) await queue.put(None) # Signal the end of the stream @@ -129,25 +129,6 @@ async def test_analyze_stream_method(system_controller): system_controller.engine.predict.assert_called_once_with(mock_frame, "192.168.1.1") -def test_check_day_time(system_controller): - with patch("pyroengine.core.is_day_time", return_value=True) as mock_is_day_time: - system_controller.check_day_time() - assert system_controller.day_time is True - mock_is_day_time.assert_called_once() - - with patch("pyroengine.core.is_day_time", return_value=False) as mock_is_day_time: - system_controller.check_day_time() - assert system_controller.day_time is False - mock_is_day_time.assert_called_once() - - with patch("pyroengine.core.is_day_time", side_effect=Exception("Error in is_day_time")) as mock_is_day_time, patch( - "pyroengine.core.logging.exception" - ) as mock_logging_exception: - system_controller.check_day_time() - mock_is_day_time.assert_called_once() - mock_logging_exception.assert_called_once_with("Exception during initial day time check: Error in is_day_time") - - def test_repr_method(system_controller): repr_str = repr(system_controller) # Check if the representation is a string