Merge pull request #1280 from roflcoopter/feature/ffmpeg-reload

feat(ffmpeg): support config reload

Author: roflcoopter

viseron/components/ffmpeg/camera.py

@@ -329,7 +329,9 @@ def __init__(
     ) -> None:
         # Add password to SensitiveInformationFilter.
         # It is done in AbstractCamera but since we are calling Stream before
-        # super().__init__ we need to do it here as well
+        # super().__init__ we need to do it here as well.
+        # For this reason we dont have to clear the SensitiveInformationFilter
+        # on unload since AbstractCamera will do it in its unload method
         if config[CONFIG_PASSWORD]:
             SensitiveInformationFilter.add_sensitive_string(config[CONFIG_PASSWORD])
             SensitiveInformationFilter.add_sensitive_string(
@@ -350,14 +352,20 @@ def __init__(
         ] = mp.Queue(maxsize=2)
         self._capture_frames = mp.Event()
         self._thread_stuck = False
-        self.resolution = None
+        self.resolution = self.stream.width, self.stream.height
         self.decode_error = mp.Event()
 
         if cv2.ocl.haveOpenCL():
             cv2.ocl.setUseOpenCL(True)
         self._recorder = Recorder(vis, config, self)
 
-        self.initialize_camera()
+        self._check_segment_process_thread: RestartableThread | None = None
+
+        self._logger.debug(
+            f"Resolution: {self.resolution[0]}x{self.resolution[1]} "
+            f"@ {self.stream.fps} FPS"
+        )
+        self._logger.debug(f"Camera {self.name} initialized")
 
     def _create_frame_reader(self) -> tuple[RestartableProcess, RestartableThread]:
         """Return a frame reader thread."""
@@ -391,7 +399,7 @@ def _start_recording_only(self) -> None:
         self._logger.debug("Starting recording only mode")
 
         def check_segment_process() -> None:
-            while self.is_on:
+            while self._capture_frames.is_set():
                 time.sleep(1)
                 if (
                     self.stream.segment_process
@@ -402,26 +410,14 @@ def check_segment_process() -> None:
                 self.connected = False
             self.connected = False
 
-        RestartableThread(
+        self._check_segment_process_thread = RestartableThread(
             name="viseron.camera." + self.identifier + ".segment_check",
             target=check_segment_process,
             daemon=True,
             register=True,
-        ).start()
-
-        self.stream.record_only()
-
-    def initialize_camera(self) -> None:
-        """Start processing of camera frames."""
-        self._logger.debug(f"Initializing camera {self.name}")
-
-        self.resolution = self.stream.width, self.stream.height
-        self._logger.debug(
-            f"Resolution: {self.resolution[0]}x{self.resolution[1]} "
-            f"@ {self.stream.fps} FPS"
         )
-
-        self._logger.debug(f"Camera {self.name} initialized")
+        self._check_segment_process_thread.start()
+        self.stream.record_only()
 
     def read_frames(
         self,
@@ -545,12 +541,12 @@ def calculate_output_fps(self, scanners: list[FrameIntervalCalculator]) -> None:
 
     def _start_camera(self) -> None:
         """Start capturing frames from camera."""
+        self._capture_frames.set()
         if self._config[CONFIG_RECORD_ONLY]:
             self._start_recording_only()
             return
 
         self._logger.debug("Starting capture thread")
-        self._capture_frames.set()
         if not self._frame_reader or not self._frame_reader.is_alive():
             self._logger.debug("Creating new frame reader")
             self._frame_reader, self._frame_relay = self._create_frame_reader()
@@ -559,13 +555,14 @@ def _start_camera(self) -> None:
 
     def _stop_camera(self) -> None:
         """Release the connection to the camera."""
-        self._logger.debug("Stopping capture thread")
         self._capture_frames.clear()
         if self._frame_relay:
+            self._logger.debug("Stopping capture thread")
             self._frame_relay.stop()
             self._frame_relay.join(timeout=5)
 
         if self._frame_reader:
+            self._logger.debug("Stopping frame reader process")
             self._frame_reader.stop()
             self._frame_reader.join(timeout=5)
             if self._frame_reader.is_alive():
@@ -575,6 +572,13 @@ def _stop_camera(self) -> None:
                 self._frame_reader = None
                 self.stream.close_pipe()
 
+        if self._config[CONFIG_RECORD_ONLY] and self._check_segment_process_thread:
+            self._logger.debug("Stopping record-only process")
+            self._check_segment_process_thread.stop()
+            self._check_segment_process_thread.join(timeout=5)
+            self._check_segment_process_thread = None
+            self.stream.close_pipe()
+
     def start_recorder(
         self,
         shared_frame: SharedFrame,
@@ -598,12 +602,12 @@ def output_fps(self, fps: int) -> None:
         self.stream.output_fps = fps
 
     @property
-    def resolution(self):
+    def resolution(self) -> tuple[int, int]:
         """Return stream resolution."""
         return self._resolution
 
     @resolution.setter
-    def resolution(self, resolution) -> None:
+    def resolution(self, resolution: tuple[int, int]) -> None:
         """Set stream resolution."""
         self._resolution = resolution
 

viseron/components/ffmpeg/stream.py

@@ -1,4 +1,5 @@
 """Class to interact with an FFmpeg stream."""
+
 from __future__ import annotations
 
 import json
@@ -107,14 +108,14 @@ def __init__(
 
         self._camera: Camera = camera
 
-        self._pipe: sp.Popen | None = None
+        self._pipe: RestartablePopen | None = None
         self.segment_process: RestartablePopen | None = None
         self._log_pipe: LogPipe | None = None
         self._ffprobe = FFprobe(config, camera_identifier, attempt)
 
         self._mainstream = self.get_stream_information(config)
         self._substream = None
-        if config.get(CONFIG_SUBSTREAM, None):
+        if config.get(CONFIG_SUBSTREAM):
             self._substream = self.get_stream_information(config[CONFIG_SUBSTREAM])
 
         self._output_fps = self.fps
@@ -131,7 +132,7 @@ def __init__(
         self.create_symlink(self.segments_alias)
 
     @property
-    def output_args(self):
+    def output_args(self) -> list:
         """Return FFmpeg output args."""
         return [
             "-f",
@@ -152,7 +153,7 @@ def segments_alias(self) -> str:
         return f"ffmpeg_{self._camera_identifier}_seg"
 
     @staticmethod
-    def create_symlink(alias) -> None:
+    def create_symlink(alias: str) -> None:
         """Create a symlink to FFmpeg executable.
 
         This is done to know which FFmpeg command belongs to which camera.
@@ -194,12 +195,12 @@ def fps(self) -> int:
         return self._mainstream.fps
 
     @property
-    def output_fps(self):
+    def output_fps(self) -> int:
         """Return stream output FPS."""
         return self._output_fps
 
     @output_fps.setter
-    def output_fps(self, fps) -> None:
+    def output_fps(self, fps: int) -> None:
         self._output_fps = fps
 
     def get_stream_url(self, stream_config: dict[str, Any]) -> str:
@@ -212,8 +213,7 @@ def get_stream_url(self, stream_config: dict[str, Any]) -> str:
 
         protocol = (
             stream_config[CONFIG_PROTOCOL]
-            if stream_config[CONFIG_PROTOCOL]
-            else STREAM_FORMAT_MAP[stream_config[CONFIG_STREAM_FORMAT]]["protocol"]
+            or STREAM_FORMAT_MAP[stream_config[CONFIG_STREAM_FORMAT]]["protocol"]
         )
 
         return (
@@ -242,11 +242,21 @@ def get_stream_information(
                 stream_url, stream_config
             )
 
-        width = stream_config[CONFIG_WIDTH] if stream_config[CONFIG_WIDTH] else width
+        width = (
+            stream_config[CONFIG_WIDTH]  # noqa: FURB110
+            if stream_config[CONFIG_WIDTH]
+            else width
+        )
         height = (
-            stream_config[CONFIG_HEIGHT] if stream_config[CONFIG_HEIGHT] else height
+            stream_config[CONFIG_HEIGHT]  # noqa: FURB110
+            if stream_config[CONFIG_HEIGHT]
+            else height
+        )
+        fps = (
+            stream_config[CONFIG_FPS]  # noqa: FURB110
+            if stream_config[CONFIG_FPS]
+            else fps
         )
-        fps = stream_config[CONFIG_FPS] if stream_config[CONFIG_FPS] else fps
         codec = (
             stream_config[CONFIG_CODEC]
             if stream_config[CONFIG_CODEC] != DEFAULT_CODEC
@@ -276,36 +286,38 @@ def get_stream_information(
         )
 
     @staticmethod
-    def get_decoder_codec(stream_config: dict[str, Any], stream_codec: str):
+    def get_decoder_codec(
+        stream_config: dict[str, Any], stream_codec: str
+    ) -> list[str]:
         """Return decoder codec set in config or from predefined codec map."""
         if stream_config[CONFIG_CODEC] and stream_config[CONFIG_CODEC] != DEFAULT_CODEC:
             return ["-c:v", stream_config[CONFIG_CODEC]]
 
         codec = None
         codec_map = None
-        if stream_codec:
-            if stream_config[CONFIG_STREAM_FORMAT] in ["rtsp", "rtmp"]:
-                if os.getenv(ENV_RASPBERRYPI3) == "true":
-                    codec_map = HWACCEL_RPI3_DECODER_CODEC_MAP
-                elif os.getenv(ENV_RASPBERRYPI4) == "true":
-                    codec_map = HWACCEL_RPI4_DECODER_CODEC_MAP
-                elif os.getenv(ENV_JETSON_NANO) == "true":
-                    codec_map = HWACCEL_JETSON_NANO_DECODER_CODEC_MAP
-                elif os.getenv(ENV_CUDA_SUPPORTED) == "true":
-                    codec_map = HWACCEL_CUDA_DECODER_CODEC_MAP
-                if codec_map:
-                    codec = codec_map.get(stream_codec, None)
+        if stream_codec and stream_config[CONFIG_STREAM_FORMAT] in ["rtsp", "rtmp"]:
+            if os.getenv(ENV_RASPBERRYPI3) == "true":
+                codec_map = HWACCEL_RPI3_DECODER_CODEC_MAP
+            elif os.getenv(ENV_RASPBERRYPI4) == "true":
+                codec_map = HWACCEL_RPI4_DECODER_CODEC_MAP
+            elif os.getenv(ENV_JETSON_NANO) == "true":
+                codec_map = HWACCEL_JETSON_NANO_DECODER_CODEC_MAP
+            elif os.getenv(ENV_CUDA_SUPPORTED) == "true":
+                codec_map = HWACCEL_CUDA_DECODER_CODEC_MAP
+            if codec_map:
+                codec = codec_map.get(stream_codec, None)
+
         if codec:
             return ["-c:v", codec]
         return []
 
-    def get_encoder_codec(self):
+    def get_encoder_codec(self) -> list[str]:
         """Return encoder codec set in config."""
         return ["-c:v", self._config[CONFIG_RECORDER][CONFIG_RECORDER_CODEC]]
 
     def stream_command(
         self, stream_config: dict[str, Any], stream_codec: str, stream_url: str
-    ):
+    ) -> list[str]:
         """Return FFmpeg input stream."""
         if stream_config[CONFIG_INPUT_ARGS]:
             input_args = stream_config[CONFIG_INPUT_ARGS]
@@ -378,7 +390,7 @@ def recorder_audio_filter_args(self) -> list[str] | list:
             ]
         return []
 
-    def segment_args(self):
+    def segment_args(self) -> list[str]:
         """Generate FFmpeg segment args."""
         return (
             [
@@ -413,7 +425,7 @@ def segment_args(self):
             ]
         )
 
-    def filter_args(self):
+    def filter_args(self) -> list[str]:
         """Return filter arguments."""
         filters = self._config[CONFIG_VIDEO_FILTERS].copy()
         if self.output_fps < self.fps:
@@ -426,7 +438,7 @@ def filter_args(self):
             ]
         return []
 
-    def build_segment_command(self):
+    def build_segment_command(self) -> list[str]:
         """Return command for writing segments only from main stream.
 
         Only used when a substream is configured.
@@ -446,7 +458,7 @@ def build_segment_command(self):
             + self.segment_args()
         )
 
-    def build_command(self):
+    def build_command(self) -> list[str]:
         """Return full FFmpeg command."""
         if self._substream:
             if self._config[CONFIG_SUBSTREAM][CONFIG_RAW_COMMAND]:
@@ -478,7 +490,7 @@ def build_command(self):
             + self.output_args
         )
 
-    def pipe(self):
+    def pipe(self) -> RestartablePopen:
         """Return subprocess pipe for FFmpeg."""
         try:
             if self._log_pipe:
@@ -519,8 +531,19 @@ def start_pipe(self) -> None:
 
     def close_pipe(self) -> None:
         """Close FFmpeg pipe."""
+        self._logger.debug("Closing pipe")
         if self.segment_process:
-            self.segment_process.terminate()
+            self._logger.debug("Terminating segment process")
+            try:
+                self.segment_process.terminate()
+                try:
+                    self.segment_process.communicate(timeout=5)
+                except sp.TimeoutExpired:
+                    self._logger.debug("FFmpeg did not terminate, killing instead.")
+                    self.segment_process.kill()
+                    self.segment_process.communicate()
+            except (AttributeError, OSError) as error:
+                self._logger.error("Failed to close segment process: %s", error)
 
         if self._pipe:
             try:
@@ -541,21 +564,22 @@ def close_pipe(self) -> None:
         except OSError as error:
             self._logger.error("Failed to close log pipe: %s", error)
 
-    def poll(self):
+    def poll(self) -> int | None:
         """Poll pipe."""
         if self._pipe:
             return self._pipe.poll()
+        return None
 
     def read(self) -> bytes | None:
         """Return a single frame from FFmpeg pipe."""
         try:
             if self._pipe and self._pipe.stdout:
                 return self._pipe.stdout.read(self.frame_bytes_size)
-        except Exception as err:  # pylint: disable=broad-except
-            self._logger.error(f"Error reading frame from pipe: {err}")
+        except Exception:  # pylint: disable=broad-except
+            self._logger.exception("Error reading frame from pipe")
         return None
 
-    def record_only(self):
+    def record_only(self) -> None:
         """Record only the stream."""
         self._logger.debug(
             f"Recording only stream: {' '.join(self.build_segment_command())}"
@@ -689,7 +713,7 @@ def run_ffprobe(
                 stdout,
             ) from error
 
-        if output.get("error", None):
+        if output.get("error"):
             raise FFprobeError(
                 output,
             )