Merge pull request #840 from roflcoopter/feature/v3.0.0b11-fixes

v3.0.0b11 bugfixes

docker/Dockerfile

@@ -148,12 +148,6 @@ RUN \
   && useradd --uid 911 --user-group --create-home abc \
   && mkdir -p /home/abc/bin /segments
 
-VOLUME /config
-VOLUME /recordings
-VOLUME /segments
-VOLUME /snapshots
-VOLUME /thumbnails
-
 ENTRYPOINT ["/init"]
 
 COPY docker/ffprobe_wrapper /home/abc/bin/ffprobe

rootfs/etc/services.d/viseron/finish

@@ -4,7 +4,11 @@ define VISERON_RESTART_EXIT_CODE 100
 define SIGNAL_EXIT_CODE 256
 define SIGTERM 15
 
+# Log the exit code and time
 foreground { s6-echo "[viseron-finish] Viseron exit code ${1}" }
+backtick -D "unknown time" date { /bin/date }
+importas -i date date
+foreground { s6-echo "[viseron-finish] Shutdown completed at ${date}" }
 
 # Exit without stopping the supervisor so the Viseron service restarts on its own
 if { s6-test ${1} -ne ${VISERON_RESTART_EXIT_CODE} }

tests/components/codeprojectai/test_object_detector.py

@@ -87,7 +87,7 @@ def test_object_detector_init(vis: MockViseron, config):
     """
     _ = MockComponent(COMPONENT, vis)
     _ = MockCamera(vis, identifier=CAMERA_IDENTIFIER)
-    with patch("codeprojectai.core.CodeProjectAIObject"):
+    with patch("viseron.components.codeprojectai.object_detector.CodeProjectAIObject"):
         detector = ObjectDetector(vis, config["codeprojectai"], CAMERA_IDENTIFIER)
         assert detector._image_resolution == (  # pylint: disable=protected-access
             640,
@@ -108,7 +108,7 @@ def test_preprocess(vis: Viseron, config):
     """
     _ = MockComponent(COMPONENT, vis)
     _ = MockCamera(vis, identifier=CAMERA_IDENTIFIER)
-    with patch("codeprojectai.core.CodeProjectAIObject"):
+    with patch("viseron.components.codeprojectai.object_detector.CodeProjectAIObject"):
         detector = ObjectDetector(vis, config["codeprojectai"], CAMERA_IDENTIFIER)
         frame = np.zeros((480, 640, 3), dtype=np.uint8)
         processed = detector.preprocess(frame)
@@ -125,7 +125,7 @@ def test_postprocess(vis: Viseron, config):
     """
     _ = MockComponent(COMPONENT, vis)
     _ = MockCamera(vis, identifier=CAMERA_IDENTIFIER)
-    with patch("codeprojectai.core.CodeProjectAIObject"):
+    with patch("viseron.components.codeprojectai.object_detector.CodeProjectAIObject"):
         detector = ObjectDetector(vis, config["codeprojectai"], CAMERA_IDENTIFIER)
         detections = [
             {
@@ -142,7 +142,7 @@ def test_postprocess(vis: Viseron, config):
         assert isinstance(objects[0], DetectedObject)
 
 
-@patch("codeprojectai.core.CodeProjectAIObject.detect")
+@patch("viseron.components.codeprojectai.object_detector.CodeProjectAIObject.detect")
 def test_return_objects_success(mock_detect, vis: Viseron, config):
     """
     Test the return_objects method of the ObjectDetector class for successful detection.
@@ -203,7 +203,7 @@ def test_object_detector_init_no_image_size(vis: Viseron, config, mock_detected_
         config (dict): The configuration dictionary.
         mock_detected_object (MagicMock): Mocked DetectedObject class.
     """
-    with patch("codeprojectai.core.CodeProjectAIObject"):
+    with patch("viseron.components.codeprojectai.object_detector.CodeProjectAIObject"):
         # Set non-square image resolution
         config["codeprojectai"]["object_detector"]["image_size"] = None
 
@@ -255,7 +255,7 @@ def test_postprocess_square_resolution(vis: Viseron, config, mock_detected_objec
         config (dict): The configuration dictionary.
         mock_detected_object (MagicMock): Mocked DetectedObject class.
     """
-    with patch("codeprojectai.core.CodeProjectAIObject"):
+    with patch("viseron.components.codeprojectai.object_detector.CodeProjectAIObject"):
         # Set square image resolution
         config["codeprojectai"]["object_detector"]["image_size"] = 640
 

tests/components/ffmpeg/test_stream.py

@@ -31,6 +31,7 @@
     DEFAULT_CODEC,
     DEFAULT_FPS,
     DEFAULT_HEIGHT,
+    DEFAULT_PASSWORD,
     DEFAULT_PROTOCOL,
     DEFAULT_RECORDER_AUDIO_CODEC,
     DEFAULT_STREAM_FORMAT,
@@ -48,7 +49,7 @@
 
 from tests.common import MockCamera, return_any
 
-CONFIG = {
+CONFIG: dict[str, Any] = {
     CONFIG_HOST: "test_host",
     CONFIG_PORT: 1234,
     CONFIG_PATH: "/",
@@ -211,22 +212,31 @@ def test_get_encoder_audio_codec(
                 stream.get_encoder_audio_codec(stream_audio_codec) == expected_audio_cmd
             )
 
-    def test_get_stream_url(self) -> None:
+    @pytest.mark.parametrize(
+        "username, password, expected_url",
+        [
+            (
+                "test_username",
+                "test_password",
+                "rtsp://test_username:test_password@test_host:1234/",
+            ),
+            ("admin", "", "rtsp://admin:@test_host:1234/"),
+            (DEFAULT_USERNAME, DEFAULT_PASSWORD, "rtsp://test_host:1234/"),
+        ],
+    )
+    def test_get_stream_url(self, username, password, expected_url) -> None:
         """Test that the correct stream url is returned."""
         mocked_camera = MockCamera(identifier="test_camera_identifier")
+        config = dict(CONFIG)
+        config[CONFIG_USERNAME] = username
+        config[CONFIG_PASSWORD] = password
+
         with patch.object(
             Stream, "__init__", MagicMock(spec=Stream, return_value=None)
         ):
             stream = Stream(CONFIG, mocked_camera, "test_camera_identifier")
-            stream._config = CONFIG  # pylint: disable=protected-access
-            assert (
-                stream.get_stream_url(CONFIG)
-                == "rtsp://test_username:test_password@test_host:1234/"
-            )
-            stream._config[  # pylint: disable=protected-access
-                CONFIG_USERNAME
-            ] = DEFAULT_USERNAME
-            assert stream.get_stream_url(CONFIG) == "rtsp://test_host:1234/"
+            stream._config = config  # pylint: disable=protected-access
+            assert stream.get_stream_url(config) == expected_url
 
     def test_get_stream_information(self):
         """Test that the correct stream information is returned."""

tests/components/storage/test__init__.py

@@ -6,7 +6,7 @@
 import tempfile
 from pathlib import Path
 from typing import TYPE_CHECKING
-from unittest.mock import MagicMock, Mock
+from unittest.mock import MagicMock, Mock, patch
 
 import pytest
 
@@ -133,7 +133,8 @@ class TestStorage:
 
     def setup_method(self, vis: Viseron) -> None:
         """Set up the test."""
-        self._storage = Storage(vis, MagicMock())
+        with patch("viseron.components.storage.CleanupManager"):
+            self._storage = Storage(vis, MagicMock())
 
     def test_search_file(self) -> None:
         """Test the search_file method."""

tests/conftest.py

@@ -28,8 +28,10 @@ class MockViseron(Viseron):
 
     def __init__(self) -> None:
         super().__init__()
-        self.register_domain = Mock(side_effect=self.register_domain)  # type: ignore
-        self.mocked_register_domain = self.register_domain  # type: ignore
+        self.register_domain = Mock(  # type: ignore[method-assign]
+            side_effect=self.register_domain,
+        )
+        self.mocked_register_domain = self.register_domain
 
 
 @pytest.fixture

viseron/__init__.py

@@ -69,6 +69,7 @@
 )
 from viseron.states import States
 from viseron.types import SupportedDomains
+from viseron.watchdog.process_watchdog import ProcessWatchDog
 from viseron.watchdog.subprocess_watchdog import SubprocessWatchDog
 from viseron.watchdog.thread_watchdog import ThreadWatchDog
 
@@ -190,8 +191,7 @@ def setup_viseron() -> Viseron:
     else:
         vis.critical_components_config_store.save(config)
 
-    end = timer()
-    LOGGER.info("Viseron initialized in %.1f seconds", end - start)
+    LOGGER.info("Viseron initialized in %.1f seconds", timer() - start)
     return vis
 
 
@@ -218,10 +218,11 @@ def __init__(self) -> None:
         self._domain_register_lock = threading.Lock()
         self.data[REGISTERED_DOMAINS] = {}
 
-        self._thread_watchdog = ThreadWatchDog()
-        self._subprocess_watchdog = SubprocessWatchDog()
         self.background_scheduler = BackgroundScheduler(timezone="UTC", daemon=True)
         self.background_scheduler.start()
+        self._thread_watchdog = ThreadWatchDog(self)
+        self._subprocess_watchdog = SubprocessWatchDog(self)
+        self._process_watchdog = ProcessWatchDog(self)
 
         self.storage: Storage | None = None
 
@@ -265,7 +266,7 @@ def register_signal_handler(self, viseron_signal, callback):
             return False
 
         return self.data[DATA_STREAM_COMPONENT].subscribe_data(
-            f"viseron/signal/{viseron_signal}", callback
+            VISERON_SIGNALS[viseron_signal], callback, stage=viseron_signal
         )
 
     def listen_event(self, event: str, callback, ioloop=None) -> Callable[[], None]:
@@ -482,15 +483,14 @@ def get_registered_identifiers(self, domain: SupportedDomains):
 
     def shutdown(self) -> None:
         """Shut down Viseron."""
+        start = timer()
         LOGGER.info("Initiating shutdown")
 
         if self.data.get(DATA_STREAM_COMPONENT, None):
             data_stream: DataStream = self.data[DATA_STREAM_COMPONENT]
 
-        self._thread_watchdog.stop()
-        self._subprocess_watchdog.stop()
         try:
-            self.background_scheduler.shutdown()
+            self.background_scheduler.shutdown(wait=False)
         except SchedulerNotRunningError as err:
             LOGGER.warning(f"Failed to shutdown scheduler: {err}")
 
@@ -504,7 +504,7 @@ def shutdown(self) -> None:
             self, data_stream, VISERON_SIGNAL_STOPPING
         )
 
-        LOGGER.info("Shutdown complete")
+        LOGGER.info("Shutdown complete in %.1f seconds", timer() - start)
 
     def add_entity(self, component: str, entity: Entity):
         """Add entity to states registry."""
@@ -543,18 +543,31 @@ def wait_for_threads_and_processes_to_exit(
     stage: Literal["shutdown", "last_write", "stopping"],
 ) -> None:
     """Wait for all threads and processes to exit."""
+    LOGGER.debug(f"Sending signal for stage {stage}")
     vis.shutdown_stage = stage
     data_stream.publish_data(VISERON_SIGNALS[stage])
-
+    time.sleep(0.1)  # Wait for signal to be processed
     LOGGER.debug(f"Waiting for threads and processes to exit in stage {stage}")
 
     def join(
         thread_or_process: threading.Thread
         | multiprocessing.Process
         | multiprocessing.process.BaseProcess,
     ) -> None:
-        thread_or_process.join(timeout=10)
-        time.sleep(0.5)  # Wait for process to exit properly
+        start_time = time.time()
+        LOGGER.debug(f"Waiting for {thread_or_process.name} to exit")
+        try:
+            thread_or_process.join(timeout=5)
+        except RuntimeError:
+            LOGGER.debug(f"Failed to join {thread_or_process.name}")
+            time.sleep(0.1)
+            thread_or_process.join(timeout=5)
+        LOGGER.debug(
+            f"Finished waiting for {thread_or_process.name} "
+            f"after {time.time() - start_time:.2f}s"
+        )
+
+        time.sleep(0.1)  # Wait for process to exit properly
         if thread_or_process.is_alive():
             LOGGER.error(f"{thread_or_process.name} did not exit in time")
             if isinstance(thread_or_process, multiprocessing.Process):

viseron/__main__.py

@@ -1,8 +1,12 @@
 """Start Viseron."""
 from __future__ import annotations
 
+import multiprocessing as mp
+import os
 import signal
 import sys
+import threading
+from threading import Timer
 
 from viseron import Viseron, setup_viseron
 
@@ -15,6 +19,16 @@ def signal_term(*_) -> None:
         if viseron:
             viseron.shutdown()
 
+        def shutdown_failed():
+            print("Shutdown failed. Exiting forcefully.")
+            print(f"Active threads: {threading.enumerate()}")
+            print(f"Active processes: {mp.active_children()}")
+            os.kill(os.getpid(), signal.SIGKILL)
+
+        shutdown_timer = Timer(2, shutdown_failed, args=())
+        shutdown_timer.daemon = True
+        shutdown_timer.start()
+
     # Listen to signals
     signal.signal(signal.SIGTERM, signal_term)
     signal.signal(signal.SIGINT, signal_term)

viseron/components/codeprojectai/object_detector.py

@@ -39,7 +39,7 @@ def __init__(self, vis: Viseron, config, camera_identifier) -> None:
         )
 
         self._ds_config = config
-        self._detector = cpai.CodeProjectAIObject(
+        self._detector = CodeProjectAIObject(
             ip=config[CONFIG_HOST],
             port=config[CONFIG_PORT],
             timeout=config[CONFIG_TIMEOUT],
@@ -109,3 +109,21 @@ def return_objects(self, frame):
             return []
 
         return self.postprocess(detections)
+
+
+class CodeProjectAIObject(cpai.CodeProjectAIObject):
+    """CodeProject.AI object detection."""
+
+    def __init__(self, ip, port, timeout, min_confidence, custom_model):
+        super().__init__(ip, port, timeout, min_confidence, custom_model)
+
+    def detect(self, image_bytes: bytes):
+        """Process image_bytes and detect."""
+        response = cpai.process_image(
+            url=self._url_detect,
+            image_bytes=image_bytes,
+            min_confidence=self.min_confidence,
+            timeout=self.timeout,
+        )
+        LOGGER.debug("CodeProject.AI response: %s", response)
+        return response["predictions"]

viseron/components/darknet/__init__.py

@@ -7,10 +7,11 @@
 import os
 import pwd
 from abc import ABC, abstractmethod
-from queue import Queue
+from queue import Empty, Queue
 from typing import Any
 
 import cv2
+import numpy as np
 import voluptuous as vol
 
 from viseron import Viseron
@@ -272,7 +273,7 @@ def spawn_subprocess(self) -> RestartablePopen:
             stderr=self._log_pipe,
         )
 
-    def preprocess(self, frame):
+    def preprocess(self, frame) -> np.ndarray:
         """Pre process frame before detection."""
         return cv2.resize(
             frame,
@@ -421,11 +422,17 @@ def work_output(self, item) -> None:
         """Put result into queue."""
         pop_if_full(self._result_queues[item["camera_identifier"]], item)
 
-    def preprocess(self, frame):
+    def preprocess(self, frame) -> bytes:
         """Pre process frame before detection."""
         return letterbox_resize(frame, self.model_width, self.model_height).tobytes()
 
-    def detect(self, frame, camera_identifier, result_queue, min_confidence):
+    def detect(
+        self,
+        frame: np.ndarray,
+        camera_identifier: str,
+        result_queue,
+        min_confidence: float,
+    ):
         """Perform detection."""
         self._result_queues[camera_identifier] = result_queue
         pop_if_full(
@@ -436,7 +443,10 @@ def detect(self, frame, camera_identifier, result_queue, min_confidence):
                 "min_confidence": min_confidence,
             },
         )
-        item = result_queue.get()
+        try:
+            item = result_queue.get(timeout=3)
+        except Empty:
+            return None
         return item["result"]
 
     def post_process(self, detections, camera_resolution):