diff --git a/.gitignore b/.gitignore index 3aadeef6..23a67ea4 100644 --- a/.gitignore +++ b/.gitignore @@ -324,6 +324,7 @@ release.tar.gz .env pi-gen /recordings/ +/recording_durations.json !frontend/src/lib/ diff --git a/backend_py/src/routes/recordings.py b/backend_py/src/routes/recordings.py index 71b188b7..8070c65f 100644 --- a/backend_py/src/routes/recordings.py +++ b/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) diff --git a/backend_py/src/server.py b/backend_py/src/server.py index 0e1976ce..890d595f 100644 --- a/backend_py/src/server.py +++ b/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: diff --git a/backend_py/src/services/cameras/device_manager.py b/backend_py/src/services/cameras/device_manager.py index 53cf8d1a..9b4732ca 100644 --- a/backend_py/src/services/cameras/device_manager.py +++ b/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 diff --git a/backend_py/src/services/cameras/pwm/serial_pwm_controller.py b/backend_py/src/services/cameras/pwm/serial_pwm_controller.py index b0e01616..82f4d103 100644 --- a/backend_py/src/services/cameras/pwm/serial_pwm_controller.py +++ b/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: diff --git a/backend_py/src/services/cameras/stream_engines/gstreamer_stream_engine.py b/backend_py/src/services/cameras/stream_engines/gstreamer_stream_engine.py index 6b41f8c1..73b368b9 100644 --- a/backend_py/src/services/cameras/stream_engines/gstreamer_stream_engine.py +++ b/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 "" diff --git a/backend_py/src/services/preferences/settings_manager.py b/backend_py/src/services/preferences/settings_manager.py index a37825d1..29e48178 100644 --- a/backend_py/src/services/preferences/settings_manager.py +++ b/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: diff --git a/backend_py/src/services/recordings/recordings_service.py b/backend_py/src/services/recordings/recordings_service.py index 03ca3386..f594c757 100644 --- a/backend_py/src/services/recordings/recordings_service.py +++ b/backend_py/src/services/recordings/recordings_service.py @@ -8,16 +8,32 @@ import json import logging import os +import shutil import subprocess import threading +import time import uuid import zipfile +from dataclasses import dataclass +from datetime import datetime +from pathlib import Path from backend_py.src.models import RecordingInfo +@dataclass +class Duration: + duration_value: str + size: int # determine uniqueness + + class RecordingsService: BASE_PATH = "" + # a file modified within this window is assumed to still be recording + ACTIVE_MTIME_THRESHOLD_S = 5.0 + DURATION_CACHE_INTERVAL_S = 300 + + DURATION_PATH = "recording_durations.json" def __init__(self, data_dir: str) -> None: self.recordings_path = os.path.join(data_dir, "recordings") @@ -26,13 +42,83 @@ def __init__(self, data_dir: str) -> None: self.logger = logging.getLogger("dwe_os_2.RecordingsService") self.recordings_lock = threading.Lock() - self.durations = {} + self._load_durations() + + self.active_recordings: dict[str, str] = {} self._last_dir_mtime = 0.0 + self._stop_event = threading.Event() + self._cache_thread: threading.Thread | None = None if not os.path.exists(self.recordings_path): os.makedirs(self.recordings_path) + def _load_durations(self) -> None: + self.durations: dict[str, Duration] = {} + path = Path(self.DURATION_PATH) + + try: + with path.open(encoding="utf-8") as f: + raw = json.load(f) + except FileNotFoundError: + return + except (json.JSONDecodeError, UnicodeDecodeError) as e: + self.logger.warning("Duration cache is corrupt (%s); ignoring it.", e) + return + except OSError as e: + self.logger.warning("Could not read duration cache: %s", e) + return + + if not isinstance(raw, dict): + self.logger.warning("Duration cache has unexpected format; ignoring it.") + return + + for file_path, data in raw.items(): + try: + self.durations[file_path] = Duration(**data) + except (TypeError, ValueError) as e: + self.logger.warning("Skipping bad cache entry %r: %s", file_path, e) + + def start(self) -> None: + """Starts the background thread that keeps the durations cache warm""" + if self._cache_thread is not None and self._cache_thread.is_alive(): + return + self._stop_event.clear() + # Daemon so a scan in progress does not keep the process alive + self._cache_thread = threading.Thread( + target=self._cache_durations_loop, daemon=True + ) + self._cache_thread.start() + + def stop(self) -> None: + # Also aborts a scan in progress, so it releases recordings_lock + self._stop_event.set() + self._cache_thread = None + + def _cache_durations_loop(self) -> None: + while not self._stop_event.is_set(): + try: + self.logger.info("Updating recordings cache.") + # get_recordings refreshes the durations cache + self.get_recordings() + except Exception as e: + self.logger.error(f"Failed to cache recording durations: {e}") + self._stop_event.wait(self.DURATION_CACHE_INTERVAL_S) + + def get_disk_stats(self) -> shutil._ntuple_diskusage: + return shutil.disk_usage(self.recordings_path) + + def save_durations(self) -> None: + with open(self.DURATION_PATH, "w") as f: + f.write( + json.dumps( + { + file_path: self.durations[file_path].__dict__ + for file_path in self.durations + } + ) + ) + def get_recordings(self) -> list[RecordingInfo]: with self.recordings_lock: # check last modified @@ -43,32 +129,84 @@ def get_recordings(self) -> list[RecordingInfo]: # A file being written to does not modify the directory, so the # size of an active recording has to be refreshed separately for recording in self.recordings: - recording.size = self._format_size(os.stat(recording.path).st_size) + if self._stop_event.is_set(): + break + try: + file_stat = os.stat(recording.path) + except FileNotFoundError: + continue + + new_size = self._format_size(file_stat.st_size) + if not recording.path.endswith(".dwvo"): + recently_modified = self._is_recently_modified(file_stat) + if ( + recently_modified + or new_size != recording.size + or recording.path in self.active_recordings + ): + self._refresh_duration(recording, recently_modified) + + recording.size = new_size + return self.recordings # otherwise run scan recordings: list[RecordingInfo] = [] for filename in os.listdir(self.recordings_path): - if filename.endswith((".mp4", ".avi", ".dwvo")): + # Each file can take a while with exiftool, so bail out on + # shutdown instead of holding recordings_lock until the end + if self._stop_event.is_set(): + return self.recordings + if filename.endswith((".mp4", ".avi", ".mkv", ".dwvo")): file_path = os.path.join(self.recordings_path, filename) file_stat = os.stat(file_path) recording_info = RecordingInfo( path=file_path, - name=filename.split(".")[0], - format=filename.split(".")[-1], - duration=( - self._get_duration(file_path) - if not filename.endswith(".dwvo") - else "00:00:00" - ), + # Bus IDs in filenames can contain dots, so split on the last + name=os.path.splitext(filename)[0], + format=os.path.splitext(filename)[1][1:], + duration="00:00:00", created=self._epoch_to_readable(file_stat.st_ctime), size=self._format_size(file_stat.st_size), ) + if not filename.endswith(".dwvo"): + self._refresh_duration( + recording_info, self._is_recently_modified(file_stat) + ) recordings.append(recording_info) self.recordings = recordings self._last_dir_mtime = current_mtime - return recordings + + self.save_durations() + + return recordings + + def _is_recently_modified(self, file_stat: os.stat_result) -> bool: + return time.time() - file_stat.st_mtime < self.ACTIVE_MTIME_THRESHOLD_S + + def _refresh_duration( + self, recording: RecordingInfo, recently_modified: bool + ) -> None: + # Caller must hold recordings_lock + duration = self._get_duration(recording.path) + + # A file with a valid duration has been finalized, even if it was just + # written to, so only an unreadable recent file is considered active + if recently_modified and duration == "Unknown": + # The file is incomplete, so show the time elapsed since it started + start = self.active_recordings.setdefault(recording.path, recording.created) + recording.duration = self._elapsed_since(start) + recording.created = datetime.now().strftime("%Y-%m-%d %H:%M:%S") + return + + self.active_recordings.pop(recording.path, None) + recording.duration = duration + + def _elapsed_since(self, created: str) -> str: + start = datetime.strptime(created, "%Y-%m-%d %H:%M:%S") + secs = max(int((datetime.now() - start).total_seconds()), 0) + return f"{secs // 3600:02}:{secs % 3600 // 60:02}:{secs % 60:02}" def _format_size(self, size_bytes: int) -> str: return f"{size_bytes / (1024 * 1024):.2f} MB" @@ -78,10 +216,30 @@ def _epoch_to_readable(self, epoch: float) -> str: return datetime.fromtimestamp(epoch).strftime("%Y-%m-%d %H:%M:%S") + @staticmethod + def _exif_duration_to_seconds(value) -> float: + """Parses exiftool durations: '12.34 s', '0:01:05', or a bare number.""" + if isinstance(value, (int, float)): + return float(value) + value = str(value).strip() + if value.endswith(" s"): + return float(value[:-2]) + secs = 0.0 + for part in value.split(":"): + secs = secs * 60 + float(part) + return secs + + @staticmethod + def _format_seconds(secs: float) -> str: + secs = int(round(secs)) + return f"{secs // 3600:02}:{secs % 3600 // 60:02}:{secs % 60:02}" + def _get_duration(self, file_path: str) -> str: if file_path in self.durations: - self.logger.info(f"Found cached duration: {file_path}") - return self.durations[file_path] + # Only cache if the size is the same, otherwise it was modified + size = os.path.getsize(file_path) + if size == self.durations[file_path].size: + return self.durations[file_path].duration_value # FIXME: We need to change this function to use a better metadata library try: @@ -95,13 +253,23 @@ def _get_duration(self, file_path: str) -> str: if output: data = json.loads(output) - if file_path.endswith(".mp4"): - duration = data[0].get("Duration", "00:00:00") - if "s" in duration: - duration = ( - f"00:00:{round(float(duration.replace(' s', ''))):02}" - ) - self.durations[file_path] = duration + size = os.path.getsize(file_path) + + if file_path.endswith((".mp4", ".mkv")): + raw_duration = data[0].get("Duration") + if not raw_duration: + return "Unknown" + try: + secs = self._exif_duration_to_seconds(raw_duration) + except ValueError: + return "Unknown" + # exiftool reports an unfinalized file as 0 s + if secs == 0: + return "Unknown" + duration = self._format_seconds(secs) + self.durations[file_path] = Duration( + duration_value=duration, size=size + ) return duration totalFrameCount = data[0].get("TotalFrameCount", 0) @@ -112,16 +280,19 @@ def _get_duration(self, file_path: str) -> str: minutes = int((duration % 3600) // 60) seconds = int(duration % 60) duration = f"{hours:02}:{minutes:02}:{seconds:02d}" - self.durations[file_path] = duration + self.durations[file_path] = Duration( + duration_value=duration, size=size + ) return duration - return "00:00:00" + return "Unknown" except FileNotFoundError as e: self.logger.error(f"exiftool was not found: {e}") except json.JSONDecodeError as e: self.logger.error(f"Error decoding output from exiftool: {e}") - except Exception as e: - self.logger.error(f"exiftool had an unknown system error: {e}") + except Exception: + # self.logger.error(f"exiftool had an unknown system error: {e}") + pass return "Unknown" def get_recording(self, filename: str) -> RecordingInfo | None: @@ -136,6 +307,7 @@ def _remove_recording_file(self, filename: str) -> bool: # Caller must hold recordings_lock recording_path = os.path.join(self.recordings_path, filename) self.durations.pop(recording_path, None) + self.active_recordings.pop(recording_path, None) if not os.path.exists(recording_path): return False @@ -168,9 +340,9 @@ def rename_recording( os.rename(old_path, new_path) for recording in self.recordings: if recording.path == old_path: - recording.name = new_name.split(".")[0] + recording.name = os.path.splitext(new_name)[0] recording.path = new_path - recording.format = new_name.split(".")[-1] + recording.format = os.path.splitext(new_name)[1][1:] return self.recordings return None diff --git a/frontend/src/components/dwe/recordings/components/recording-table.tsx b/frontend/src/components/dwe/recordings/components/recording-table.tsx index 928679ab..be811bb0 100644 --- a/frontend/src/components/dwe/recordings/components/recording-table.tsx +++ b/frontend/src/components/dwe/recordings/components/recording-table.tsx @@ -305,7 +305,7 @@ export const RecordingTable = ({ onDoubleClick={(e) => e.stopPropagation()} className="group" > - {recording?.format === "mp4" ? ( + {isPlayable(recording) ? ( ) : ( @@ -314,7 +314,7 @@ export const RecordingTable = ({

- {recording?.format === "mp4" + {isPlayable(recording) ? "Playable in browser" : "Download required to play"}

diff --git a/frontend/src/components/dwe/recordings/recordings.tsx b/frontend/src/components/dwe/recordings/recordings.tsx index 938d6551..3e6b4ed3 100644 --- a/frontend/src/components/dwe/recordings/recordings.tsx +++ b/frontend/src/components/dwe/recordings/recordings.tsx @@ -34,10 +34,10 @@ const Recordings = () => { const containerRef = useRef(null); const [searchQuery, setSearchQuery] = useState(""); const [sortColumn, setSortColumn] = useState( - null, + "created", ); const [sortDirection, setSortDirection] = useState<"asc" | "desc" | null>( - null, + "desc", ); // initial data fetch, then keep sizes of active recordings up to date @@ -318,7 +318,7 @@ const Recordings = () => { {snap.zipJobs && snap.zipJobs.length > 0 && (
diff --git a/frontend/src/components/dwe/recordings/utils/recording-utils.tsx b/frontend/src/components/dwe/recordings/utils/recording-utils.tsx index 388b36d9..67d6d163 100644 --- a/frontend/src/components/dwe/recordings/utils/recording-utils.tsx +++ b/frontend/src/components/dwe/recordings/utils/recording-utils.tsx @@ -36,7 +36,7 @@ export const formatDate = (value: string | null | undefined): string => { export const fullName = (rec: RecordingInfo) => `${rec.name}.${rec.format}`; -export const isPlayable = (rec: RecordingInfo) => rec.format === "mp4"; +export const isPlayable = (rec: RecordingInfo) => rec.format === "mkv" || rec.format === "mp4"; export const getRecordingStreamUrl = (rec: RecordingInfo, baseUrl: string) => `${baseUrl}/api/recordings/${encodeURIComponent(fullName(rec))}`; diff --git a/run_release.py b/run_release.py index a721f168..de267261 100755 --- a/run_release.py +++ b/run_release.py @@ -83,6 +83,7 @@ async def lifespan(app: FastAPI) -> AsyncIterator[None]: sio, app, settings_path=args.settings_path, + pwm_light_mode=True # For #560 only ) FRONTEND_DIR = os.path.abspath("./frontend/dist")