DBC22-6169: access webcam database to retrieve is_on property

revert back

revert back

DBC22-6169: check camera is on status every 5 seconds

removed migration files no need

debug test 1

debug test 2

debug test 3

debug test 4

debug test 5

debug test 6

debug test 7

debug test 8

debug test 9

debug test 10

fixed backup and restore logic

removed comments unused

fixed failed test cases

fixed failed test case 1

fixed sonarcloud alerts 1

fixed failed test case 2

fixed failed test case 2

added new test cases for processor

added more test case to improve coverage

improve test coverage 1

improve test coverage 2

improve test coverage 2

improve test coverage 3

improve test coverage 4

improve test coverage 5

improve test coverage 6

improve test coverage 7

removed unused import

Author: bcgov-brwang

compose/backend/leader-election.py

@@ -50,6 +50,8 @@
 
 huey_process: subprocess.Popen | None = None
 
+monitor_process = None
+
 # ---------------------------------------------------------------------------
 # Lease helpers
 # ---------------------------------------------------------------------------
@@ -132,6 +134,17 @@ def _start_huey():
     )
     logger.info("Huey running (PID %d)", huey_process.pid)
 
+def _start_monitor():
+    global monitor_process
+
+    monitor_process = subprocess.Popen(
+        ["python", "manage.py", "monitor_camera_status"],
+        stdout=sys.stdout,
+        stderr=sys.stderr,
+        cwd="/app/backend",
+    )
+
+    logger.info("Monitor started (PID %d)", monitor_process.pid)
 
 def _stop_huey(timeout: int = 30):
     global huey_process
@@ -189,6 +202,7 @@ def main():
                 # No lease exists yet — race to create it
                 if _create_lease(api):
                     leading = True
+                    _start_monitor()
                     _start_huey()
 
             elif _i_hold_lease(lease):
@@ -202,7 +216,8 @@ def main():
                         # We hold the lease (e.g. after a container restart) 
                         # but haven't started the process yet.
                         logger.info("Already hold lease after restart — starting Huey")
-                        leading = True
+                        leading = True 
+                        _start_monitor()
                         _start_huey()
                     elif not _huey_alive():
                         # Huey was running, but now it's not.
@@ -225,6 +240,7 @@ def main():
                     )
                     logger.info("Takeover successful — we are the new leader")
                     leading = True
+                    _start_monitor()
                     _start_huey()
                 except ApiException as e:
                     logger.warning("Takeover failed (lost race?): %s", e)

compose/backend/start_tasks.sh

@@ -4,4 +4,5 @@ set -o errexit
 set -o pipefail
 set -o nounset
 
+python manage.py monitor_camera_status &
 python manage.py run_huey

src/backend/apps/consumer/processor.py

@@ -27,6 +27,7 @@
 from botocore.config import Config
 from django.contrib.gis.geos import Point
 from django.db import close_old_connections, connection
+import shutil
 
 
 tf = TimezoneFinder()
@@ -159,7 +160,7 @@ async def consume_queue(queue, name: str):
             try:
                 await process_message(message)
             except Exception as e:
-                logger.error(f"Error processing message from {name}: {e}")
+                logging.exception(f"Error processing message from {name}: {e}")
 
 async def consume_from(rb_url: str, name: str):
     while not stop_event.is_set():
@@ -380,7 +381,7 @@ def watermark(webcam: any, image_data: bytes, tz: str, timestamp: str) -> bytes:
         return buffer.read()
 
     except Exception as e:
-        logger.error(f"Error processing image from camera: {e}")
+        logging.exception(f"Error processing image from camera: {e}")
         return None
 
 def blank_out_image(webcam: any, image_data: bytes, tz: str, timestamp: str) -> bytes:
@@ -426,7 +427,7 @@ def blank_out_image(webcam: any, image_data: bytes, tz: str, timestamp: str) ->
         return buffer.read()
 
     except Exception as e:
-        logger.error(f"Error processing image from camera: {e}")
+        logging.exception(f"Error processing image from camera: {e}")
         return None
 
 
@@ -441,7 +442,7 @@ def save_original_image_to_pvc(camera_id: str, image_bytes: bytes):
         with open(filepath, "wb") as f:
             f.write(image_bytes)
     except Exception as e:
-        logger.error(f"Error saving original image to PVC {filepath}: {e}")
+        logging.exception(f"Error saving original image to PVC {filepath}: {e}")
     logger.info(f"Original image saved to PVC at {filepath}")
 
 def save_watermarked_image_to_pvc(camera_id: str, image_bytes: bytes, timestamp: str, is_on: bool):  
@@ -460,23 +461,44 @@ def save_watermarked_image_to_pvc(camera_id: str, image_bytes: bytes, timestamp:
         else:
             logger.info(f"Blank out image saved to PVC at {filepath}")
     except Exception as e:
-        logger.error(f"Error saving image to PVC {filepath}: {e}")
+        logging.exception(f"Error saving image to PVC {filepath}: {e}")
 
-def save_watermarked_image_to_drivebc_pvc(camera_id: str, image_bytes: bytes, is_on: bool):  
-    os.makedirs(os.path.dirname(f'{DRIVEBC_PVC_WATERMARKED_PATH}'), exist_ok=True)
+def check_backup_exists(camera_id: str) -> bool:
+    backup_dir = os.path.join(DRIVEBC_PVC_WATERMARKED_PATH, "backup")
+    backup_filepath = os.path.join(backup_dir, f"{camera_id}.jpg")
+    return os.path.exists(backup_filepath)
+
+def save_watermarked_image_to_drivebc_pvc(camera_id: str, image_bytes: bytes, is_on: bool):
+    save_dir = DRIVEBC_PVC_WATERMARKED_PATH
+    backup_dir = os.path.join(save_dir, "backup")
 
-    save_dir = os.path.join(DRIVEBC_PVC_WATERMARKED_PATH)
     os.makedirs(save_dir, exist_ok=True)
+    os.makedirs(backup_dir, exist_ok=True)
+
     filename = f"{camera_id}.jpg"
     filepath = os.path.join(save_dir, filename)
+    backup_filepath = os.path.join(backup_dir, filename)
 
     try:
+        if os.path.exists(filepath) and not is_on:
+            if not check_backup_exists(camera_id):
+                logger.info(
+                    f"Backing up existing image for camera {camera_id} to {backup_filepath}"
+                )
+                shutil.move(filepath, backup_filepath)
+            else:
+                logger.debug(
+                    f"Backup image already exists for camera {camera_id}. Skipping backup."
+                )
+        # Save the new image
         with open(filepath, "wb") as f:
             f.write(image_bytes)
+
         if is_on:
-            logger.info(f"Watermarked image saved to drivebc PVC at {filepath}")
+            logger.info(f"Watermarked image saved to DriveBC PVC at {filepath}")
+
     except Exception as e:
-        logger.error(f"Error saving image to drivebc PVC {filepath}: {e}")
+        logging.exception(f"Error saving image to DriveBC PVC {filepath}: {e}")
 
 def delete_watermarked_image_from_pvc(camera_id: str):
     save_dir = os.path.join(PVC_WATERMARKED_PATH, camera_id)
@@ -491,7 +513,7 @@ def delete_watermarked_image_from_pvc(camera_id: str):
             if os.path.isfile(filepath):
                 os.remove(filepath)
     except Exception as e:
-        logger.error(f"Error deleting watermarked images from PVC {save_dir}: {e}")
+        logging.exception(f"Error deleting watermarked images from PVC {save_dir}: {e}")
 
 async def get_images_within(camera_id: str, hours: int = 720) -> list:
     cutoff = datetime.now(timezone.utc) - timedelta(hours=hours)

src/backend/apps/consumer/tasks.py

@@ -1,7 +1,7 @@
 import logging
 import io
 
-from apps.consumer.processor import process_camera_rows, watermark, blank_out_image, save_watermarked_image_to_pvc, save_watermarked_image_to_drivebc_pvc, delete_watermarked_image_from_pvc, delete_offline_webcam_records
+from apps.consumer.processor import check_backup_exists, process_camera_rows, blank_out_image, save_watermarked_image_to_drivebc_pvc, delete_watermarked_image_from_pvc, delete_offline_webcam_records
 from datetime import datetime
 
 from apps.webcam.models import Webcam
@@ -38,8 +38,9 @@ def generate_offline_camera_images():
                 delete_watermarked_image_from_pvc(camera_id)
                 # Delete all the records from image index table for offline cams
                 async_to_sync(delete_offline_webcam_records)(camera_id)
-                # Save blank image for current image displaying
-                save_watermarked_image_to_drivebc_pvc(camera_id, watermarked, False)
+                if not check_backup_exists(camera_id):
+                    # Save blank image for current image displaying
+                    save_watermarked_image_to_drivebc_pvc(camera_id, watermarked, False)
             # Update postgres db is_on status to False for offline cameras
             Webcam.objects.filter(id=camera['id']).update(is_on=False)
         else:

src/backend/apps/consumer/tests/test_processor.py

@@ -1,4 +1,5 @@
 import os
+from PIL import Image
 from unittest.mock import patch, MagicMock, AsyncMock
 from datetime import datetime
 import io
@@ -11,6 +12,7 @@
 from apps.consumer.processor import (
     process_camera_rows,
     get_timezone,
+    process_message,
     watermark,
     verify_image,
     push_to_s3,
@@ -536,6 +538,160 @@ async def run_test():
 
         mock_logger.info.assert_called()
 
+class TestProcessMessage(TestCase):
+
+    @patch("apps.consumer.processor.logger")
+    @patch("apps.consumer.processor.handle_image_message", new_callable=AsyncMock)
+    @patch("apps.consumer.processor.safe_db_call")
+    @patch("apps.consumer.processor.generate_local_timestamp")
+    @patch("apps.consumer.processor.calculate_camera_status")
+    @patch("apps.consumer.processor.sync_to_async")
+    async def test_process_message_success(
+        self,
+        mock_sync_to_async,
+        mock_calculate_status,
+        mock_generate_local_timestamp,
+        mock_safe_db_call,
+        mock_handle_image,
+        mock_logger,
+    ):
+        message = MagicMock()
+        message.headers = {
+            "filename": "12345_20250101.jpg",
+            "timestamp": "2025-01-01T00:00:00Z",
+        }
+        message.body = b"image"
+
+        process_cm = AsyncMock()
+        message.process.return_value = process_cm
+
+        def sync_wrapper(func):
+            if func == mock_calculate_status:
+                return AsyncMock(return_value=True)
+            return AsyncMock(return_value="local_timestamp")
+
+        mock_sync_to_async.side_effect = sync_wrapper
+
+        await process_message(message)
+
+        mock_handle_image.assert_awaited_once_with(
+            "12345",
+            b"image",
+            "local_timestamp",
+            True,
+        )
+
+        mock_logger.info.assert_called_once_with(
+            "Processed message for camera %s.",
+            "12345",
+        )
+
+    @patch("apps.consumer.processor.logger")
+    @patch(
+        "apps.consumer.processor.handle_image_message",
+        new_callable=AsyncMock,
+        side_effect=asyncio.TimeoutError,
+    )
+    @patch("apps.consumer.processor.sync_to_async")
+    async def test_process_message_timeout(
+        self,
+        mock_sync_to_async,
+        mock_handle_image,
+        mock_logger,
+    ):
+        message = MagicMock()
+        message.headers = {
+            "filename": "123.jpg",
+            "timestamp": "2025",
+        }
+        message.body = b""
+
+        message.process.return_value = AsyncMock()
+
+        mock_sync_to_async.side_effect = [
+            AsyncMock(return_value=True),
+            AsyncMock(return_value="local"),
+        ]
+
+        await process_message(message)
+
+        mock_logger.error.assert_called_once_with(
+            "Processing timeout for camera 123"
+        )
+
+class TestWatermark(TestCase):
+
+    def create_image(self, width=640, height=480):
+        img = Image.new("RGB", (width, height), color="red")
+        buf = io.BytesIO()
+        img.save(buf, format="JPEG")
+        return buf.getvalue()
+    
+    def test_returns_none_when_image_is_none(self):
+        webcam = {"dbc_mark": "TEST"}
+
+        result = watermark(
+            webcam,
+            None,
+            "America/Vancouver",
+            "20250720120000000000",
+        )
+
+        self.assertIsNone(result)
+
+    def test_successfully_watermarks_image(self):
+        webcam = {"dbc_mark": "DriveBC"}
+
+        result = watermark(
+            webcam,
+            self.create_image(),
+            "America/Vancouver",
+            "20250720120000000000",
+        )
+
+        self.assertIsNotNone(result)
+
+        img = Image.open(io.BytesIO(result))
+
+        # original height = 480
+        self.assertEqual(img.size, (640, 498))
+
+    def test_missing_dbc_mark(self):
+        webcam = {}
+
+        result = watermark(
+            webcam,
+            self.create_image(),
+            "America/Vancouver",
+            "20250720120000000000",
+        )
+
+        self.assertIsNotNone(result)
+
+    def test_invalid_timestamp_returns_none(self):
+        webcam = {"dbc_mark": "DriveBC"}
+
+        result = watermark(
+            webcam,
+            self.create_image(),
+            "America/Vancouver",
+            "bad timestamp",
+        )
+
+        self.assertIsNone(result)
+
+    def test_invalid_image_returns_none(self):
+        webcam = {"dbc_mark": "DriveBC"}
+
+        result = watermark(
+            webcam,
+            b"not an image",
+            "America/Vancouver",
+            "20250720120000000000",
+        )
+
+        self.assertIsNone(result)
+
 def reset_stop_event():
     import asyncio
     from apps.consumer import processor