Feature/split recording

.gitignore

@@ -324,6 +324,7 @@ release.tar.gz
 .env
 pi-gen
 /recordings/
+/recording_durations.json
 
 !frontend/src/lib/
 

backend_py/src/routes/recordings.py

@@ -8,7 +8,6 @@
 
 import asyncio
 import os
-import shutil
 import time
 import uuid
 
@@ -81,7 +80,7 @@ def get_recordings(request: Request) -> list[RecordingInfo]:
 def get_disk_usage(request: Request) -> DiskStatsResponse:
     recordings_service: RecordingsService = request.app.state.recordings_service
 
-    total, used, free = shutil.disk_usage(recordings_service.recordings_path)
+    total, used, free = recordings_service.get_disk_stats()
 
     return DiskStatsResponse(total=total, used=used, free=free)
 

backend_py/src/server.py

@@ -48,6 +48,7 @@ def __init__(
         settings_path: str = ".",
         log_level=logging.INFO,
         is_dev_mode=False,
+        pwm_light_mode=False,
     ) -> None:
         # initialize the app
         self.app = app
@@ -106,12 +107,18 @@ def __init__(
 
         # Serial
         self.serial = SerialPWMController(
-            frequency_offset=self.preferences_manager.get_preferences().frequency_offset
+            frequency_offset=self.preferences_manager.get_preferences().frequency_offset,
+            pwm_light_mode=pwm_light_mode,
         )
 
+        self.recordings_service = RecordingsService(self.data_dir)
+
         # Device Manager
         self.device_manager = DeviceManager(
-            settings_manager=self.settings_manager, sio=self.sio, serial=self.serial
+            settings_manager=self.settings_manager,
+            sio=self.sio,
+            serial=self.serial,
+            recordings_service=self.recordings_service,
         )
 
         self.server_logger = logging.getLogger("dwe_os_2.Server")
@@ -123,8 +130,6 @@ def __init__(
 
         self.system_manager = SystemManager()
 
-        self.recordings_service = RecordingsService(self.data_dir)
-
         # TTYD
         if self.feature_support.ttyd:
             self.ttyd_manager = TTYDManager(is_dev_mode)
@@ -193,6 +198,7 @@ async def serve(self) -> None:
             self.serial.start()
 
         self.device_manager.start_monitoring()
+        self.recordings_service.start()
 
         if self.feature_support.wifi:
             await self.network_wrapper.initialize()
@@ -210,6 +216,7 @@ def shutdown(self) -> None:
         self.server_logger.info("Shutting down")
 
         self.device_manager.stop_monitoring()
+        self.recordings_service.stop()
         self.settings_manager.cleanup()
 
         if self.feature_support.ttyd:

backend_py/src/services/cameras/device_manager.py

@@ -26,6 +26,7 @@
     StreamInfoModel,
     StreamTypeEnum,
 )
+from backend_py.src.services.recordings.recordings_service import RecordingsService
 
 from ..preferences import SettingsManager
 from .device_utils import find_device_with_bus_info, list_diff
@@ -68,11 +69,15 @@ class DeviceManager(events.EventEmitter):
     Class for interfacing with and monitoring devices
     """
 
+    # Recordings are stopped (and cannot be started) below this much free space
+    MIN_RECORDING_FREE_BYTES = 2 * 1024**3
+
     def __init__(
         self,
         sio: socketio.AsyncServer,
         settings_manager: SettingsManager,
         serial: SerialPWMController,
+        recordings_service: RecordingsService,
     ) -> None:
         super().__init__()
 
@@ -84,6 +89,8 @@ def __init__(
         # List of devices with stream errors
         self.stream_errors: list[str] = []
 
+        self.recordings_service = recordings_service
+
         self.serial = serial
 
         self.logger = logging.getLogger("dwe_os_2.cameras.DeviceManager")
@@ -102,6 +109,7 @@ def start_monitoring(self) -> None:
         self._is_monitoring = True
         self._loop = asyncio.get_running_loop()
         asyncio.create_task(self._monitor())
+        asyncio.create_task(self._monitor_disk_usage())
 
     def stop_monitoring(self) -> None:
         """
@@ -223,13 +231,28 @@ def configure_device_stream(self, stream_info: StreamInfoModel) -> bool:
             encode_type, width, height, interval, stream_type, endpoints
         )
 
-        if stream_info.enabled:
+        should_start = stream_info.enabled
+        if (
+            should_start
+            and stream_type == StreamTypeEnum.RECORDING
+            and not self._can_record()
+        ):
+            self.logger.error(
+                "Cannot start recording with less than "
+                f"{int(self.MIN_RECORDING_FREE_BYTES / (1024**3))} "
+                "GB of disk space remaining"
+            )
+            should_start = False
+            # The frontend assumes the stream started, so make it refetch
+            device.emit("updated")
+
+        if should_start:
             device.start_stream()
         else:
             device.stop_stream()
 
         self.settings_manager.save_device(device)
-        return True
+        return should_start == stream_info.enabled
 
     def set_device_nickname(self, bus_info: str, nickname: str) -> bool:
         """
@@ -490,6 +513,42 @@ async def _monitor(self) -> None:
             # get the list of devices and update the internal array
             devices_info = await self._get_devices(devices_info)
 
+    def _can_record(self) -> bool:
+        """
+        Check if there is enough free disk space to record
+        """
+        free = self.recordings_service.get_disk_stats().free
+        return free > self.MIN_RECORDING_FREE_BYTES
+
+    async def _monitor_disk_usage(self) -> None:
+        """
+        Internal code to stop recordings when the disk is almost full
+        """
+        while self._is_monitoring:
+            await asyncio.sleep(5)
+
+            recording_devices = [
+                device
+                for device in self.devices
+                if device.stream.enabled
+                and device.stream.stream_type == StreamTypeEnum.RECORDING
+            ]
+
+            if not recording_devices or self._can_record():
+                continue
+
+            # Logged as an error so the frontend shows it as a toast
+            self.logger.error(
+                "Less than 2 GB of disk space remaining, stopping all recordings"
+            )
+
+            for device in recording_devices:
+                device.stop_stream()
+                # Save so the recording does not resume on restart
+                self.settings_manager.save_device(device)
+
+            await self.sio.emit("device_updated")
+
     async def _emit_stream_error(self, device: str, errors: list | str) -> None:
         """
         Emit a stream_error and make sure it is not due to the device being unplugged

backend_py/src/services/cameras/pwm/serial_pwm_controller.py

@@ -1,5 +1,6 @@
 import asyncio
 import logging
+import time
 
 import serial
 from serial.tools import list_ports
@@ -33,10 +34,13 @@ def _find_pwm_device_path() -> str | None:
 
 
 class SerialPWMController:
-    def __init__(self, baudrate: int = 9600, frequency_offset: float = 0) -> None:
+    def __init__(
+        self, baudrate: int = 9600, frequency_offset: float = 0, pwm_light_mode=False
+    ) -> None:
         self.found_port = False
         self.has_printed_error = False
         self.frequency_offset = frequency_offset
+        self.pwm_light_mode = pwm_light_mode
 
         self.logger = logging.getLogger("dwe_os_2.pwm.SerialPWMController")
         self.baudrate = baudrate
@@ -46,9 +50,14 @@ def __init__(self, baudrate: int = 9600, frequency_offset: float = 0) -> None:
         self._monitor_task: asyncio.Task | None = None
         self._running = False
 
+        # Apply 100% brightness on start
+        if pwm_light_mode:
+            self.apply(60, 100)
+
     def set_frequency_offset(self, frequency_offset: float) -> None:
-        self.frequency_offset = frequency_offset
-        self.apply(self.frequency, self.duty_cycle)
+        if not self.pwm_light_mode:
+            self.frequency_offset = frequency_offset
+            self.apply(self.frequency, self.duty_cycle)
 
     def _disconnect_serial(self) -> None:
         self.found_port = False
@@ -79,6 +88,10 @@ async def _monitor_loop(self) -> None:
                             self.found_port = True
                             self.has_printed_error = False
                             self.logger.info("PWM USB serial connected at %s", path)
+
+                            # Wait until serial connects
+                            time.sleep(4)
+
                             # Perform initial apply
                             self.apply(self.frequency, self.duty_cycle)
                         except serial.SerialException as e:
@@ -119,6 +132,12 @@ def apply(self, frequency: float, duty_cycle: int) -> None:
             self.serial.write(command.encode("utf-8"))
 
     def apply_from_fps(self, fps: int) -> None:
+        if self.pwm_light_mode:
+            self.logger.info(
+                "apply_from_fps was called, but PWM controller is being used for "
+                "lighting. Ignoring request."
+            )
+            return
         self.apply(fps, 30)
 
     def stop(self) -> None:

backend_py/src/services/cameras/stream_engines/gstreamer_stream_engine.py

@@ -57,13 +57,13 @@ def _build_payload(stream: Stream) -> str:
                         f"h264parse ! video/x-h264,width={stream.width},"
                         f"height={stream.height},"
                         f"framerate={stream.interval.denominator}/"
-                        f"{stream.interval.numerator} ! queue ! mp4mux"
+                        f"{stream.interval.numerator} ! queue"
                     )
                 else:
                     return "h264parse ! queue ! rtph264pay config-interval=10 pt=96"
             case StreamEncodeTypeEnum.MJPG:
                 if stream.stream_type == StreamTypeEnum.RECORDING:
-                    return "queue ! avimux"
+                    return "queue"
                 else:
                     return "rtpjpegpay"
             case StreamEncodeTypeEnum.SOFTWARE_H264:
@@ -79,7 +79,7 @@ def _build_payload(stream: Stream) -> str:
                         "h264parse ! video/x-h264,"
                         f"width={stream.width},height={stream.height},"
                         f"framerate={stream.interval.denominator}"
-                        f"/{stream.interval.numerator} ! queue ! mp4mux"
+                        f"/{stream.interval.numerator} ! queue"
                     )
                 else:
                     return (
@@ -101,16 +101,18 @@ def _build_sink(stream: Stream, recording_directory: str) -> str:
                 sink += ",".join(f"{e.host}:{e.port}" for e in stream.endpoints)
                 return sink
             case StreamTypeEnum.RECORDING:
-                extension = (
-                    "avi" if stream.encode_type == StreamEncodeTypeEnum.MJPG else "mp4"
-                )
+                extension = "mkv"
                 timestamp = datetime.now().strftime("%F-%T")
-                unique_filename = (
-                    f"{stream.device_path.split('/')[-1]}_{timestamp}.{extension}"
-                )
+                fixed_bus_info = stream.bus_info.replace(".", "_")
+                unique_filename = f"{fixed_bus_info}__{timestamp}_%02d.{extension}"
                 unique_path = os.path.join(recording_directory, unique_filename)
                 stream.file_path = unique_path
-                return f"filesink location={unique_path} sync=true"
+                return (
+                    f'splitmuxsink location="{unique_path}" '
+                    "muxer-factory=matroskamux "
+                    'muxer-properties="properties,streamable=true" '
+                    "max-size-time=300000000000"
+                )
             case _:
                 return ""
 

backend_py/src/services/preferences/settings_manager.py

@@ -77,7 +77,8 @@ def load_device(self, device: Device, devices: dict[str, Device]) -> None:
             for saved_device in self.settings:
                 if saved_device.bus_info == device.bus_info:
                     self._load_device(device, saved_device, devices)
-                    return
+                    if not isinstance(device, SHDDevice) and device.stream.enabled:
+                        device.start_stream()
 
     def get_saved_device(self, bus_info: str) -> SavedDeviceModel | None:
         for saved_device in self.settings: