diff --git a/backend_py/src/models/cameras.py b/backend_py/src/models/cameras.py index b1db3d5b..fe3315fd 100644 --- a/backend_py/src/models/cameras.py +++ b/backend_py/src/models/cameras.py @@ -161,6 +161,8 @@ class StreamModel(BaseModel): height: int interval: IntervalModel enabled: bool + recording_interval: int + recording_length: int class Config: from_attributes = True @@ -230,6 +232,8 @@ class StreamInfoModel(BaseModel): encode_type: StreamEncodeTypeEnum enabled: bool endpoints: list[StreamEndpointModel] + recording_interval: int + recording_length: int class Config: from_attributes = True diff --git a/backend_py/src/models/saved_cameras.py b/backend_py/src/models/saved_cameras.py index 8f94b735..774a7c14 100644 --- a/backend_py/src/models/saved_cameras.py +++ b/backend_py/src/models/saved_cameras.py @@ -35,6 +35,9 @@ class SavedStreamModel(BaseModel): height: int interval: IntervalModel enabled: bool + # Defaults keep settings saved before interval recording loadable + recording_interval: int = 0 + recording_length: int = 0 class Config: # use_enum_values = True diff --git a/backend_py/src/server.py b/backend_py/src/server.py index 0e1976ce..1c6344ac 100644 --- a/backend_py/src/server.py +++ b/backend_py/src/server.py @@ -111,7 +111,10 @@ def __init__( # 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, + serial_mode_fps=False, # Spelunky ) self.server_logger = logging.getLogger("dwe_os_2.Server") diff --git a/backend_py/src/services/cameras/device_manager.py b/backend_py/src/services/cameras/device_manager.py index 53cf8d1a..ab1c0a9e 100644 --- a/backend_py/src/services/cameras/device_manager.py +++ b/backend_py/src/services/cameras/device_manager.py @@ -37,6 +37,8 @@ from .exceptions import DeviceNotFoundException from .pwm.serial_pwm_controller import SerialPWMController +LIGHT_PWM_FREQ = 60 + def todict(obj, classkey=None) -> Any: if isinstance(obj, dict): @@ -73,6 +75,9 @@ def __init__( sio: socketio.AsyncServer, settings_manager: SettingsManager, serial: SerialPWMController, + # Defines whether or not the serial communication is + # used to set the FPS or just duty cycle + serial_mode_fps=True, ) -> None: super().__init__() @@ -85,6 +90,7 @@ def __init__( self.stream_errors: list[str] = [] self.serial = serial + self.serial_mode_fps = serial_mode_fps self.logger = logging.getLogger("dwe_os_2.cameras.DeviceManager") @@ -143,6 +149,13 @@ def create_device(self, device_info: DeviceInfo) -> Device | None: lambda _: self._append_stream_error(DeviceModel.model_validate(device)), ) + # if not self.serial_mode_fps: + # device.stream_runner.on("stream_active", + # lambda: self._set_light(True)) + # + # device.stream_runner.on("stream_inactive", + # lambda: self._set_light(False)) + # Hack to allow shd to save follower and leader settings on removal device.on("save", lambda: self.settings_manager.save_device(device)) @@ -155,11 +168,21 @@ def create_device(self, device_info: DeviceInfo) -> Device | None: ) # Only followers will update PWM frequency - if self.serial and device.can_follow: + if self.serial and device.can_follow and self.serial_mode_fps: device.on("pwm_frequency", self.serial.apply_from_fps) return device + def _set_light(self, enabled: bool) -> None: + """ + Sends a signal to the PWM controller assuming it is connected to a light + """ + + self.logger.info(f"Applying value to light: {enabled}") + + if self.serial: + self.serial.apply(LIGHT_PWM_FREQ, 100 if enabled else 0) + def _append_stream_error(self, device: DeviceModel) -> None: """ Helper function to append a gst error @@ -218,9 +241,18 @@ def configure_device_stream(self, stream_info: StreamInfoModel) -> bool: encode_type: StreamEncodeTypeEnum = stream_info.encode_type stream_type: StreamTypeEnum = stream_info.stream_type endpoints = stream_info.endpoints + recording_interval = stream_info.recording_interval + recording_length = stream_info.recording_length device.configure_stream( - encode_type, width, height, interval, stream_type, endpoints + encode_type, + width, + height, + interval, + stream_type, + recording_interval, + recording_length, + endpoints, ) if stream_info.enabled: diff --git a/backend_py/src/services/cameras/drivers/device.py b/backend_py/src/services/cameras/drivers/device.py index 25043d9d..8822acb1 100644 --- a/backend_py/src/services/cameras/drivers/device.py +++ b/backend_py/src/services/cameras/drivers/device.py @@ -216,6 +216,8 @@ def configure_stream( height: int, interval: IntervalModel, stream_type: StreamTypeEnum, + recording_interval: int, + recording_length: int, stream_endpoints: list[StreamEndpointModel] | None = None, ) -> None: if stream_endpoints is None: @@ -249,6 +251,8 @@ def configure_stream( self.stream.endpoints = stream_endpoints self.stream.encode_type = encode_type self.stream.stream_type = stream_type + self.stream.recording_interval = recording_interval + self.stream.recording_length = recording_length # Update the pwm frequency with the new fps # TODO: This should be on a command bus or something, not emitted from @@ -314,6 +318,8 @@ def load_settings(self, saved_device: SavedDeviceModel) -> None: saved_device.stream.height, saved_device.stream.interval, saved_device.stream.stream_type, + saved_device.stream.recording_interval, + saved_device.stream.recording_length, saved_device.stream.endpoints, ) diff --git a/backend_py/src/services/cameras/drivers/shd/shd.py b/backend_py/src/services/cameras/drivers/shd/shd.py index bf8d6575..d8dc18bb 100644 --- a/backend_py/src/services/cameras/drivers/shd/shd.py +++ b/backend_py/src/services/cameras/drivers/shd/shd.py @@ -231,6 +231,8 @@ def start_stream(self) -> None: self.stream.height, self.stream.interval, self.stream.stream_type, + self.stream.recording_interval, + self.stream.recording_length, [], ) 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..562cf929 100644 --- a/backend_py/src/services/cameras/pwm/serial_pwm_controller.py +++ b/backend_py/src/services/cameras/pwm/serial_pwm_controller.py @@ -110,12 +110,12 @@ def apply(self, frequency: float, duty_cycle: int) -> None: # have the correct clock frequency self.frequency = frequency self.duty_cycle = duty_cycle - if not self.found_port: - return command = f"{frequency + self.frequency_offset},{duty_cycle}\n" + self.logger.info(f"Sending command {command.strip()}") + if not self.found_port: + return if self.serial: - self.logger.info(f"Sending command {command.strip()}") self.serial.write(command.encode("utf-8")) def apply_from_fps(self, fps: int) -> 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..138de6db 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 @@ -130,43 +130,105 @@ def __init__( self._lock = threading.RLock() self.started = False + # Interval recording + self.recording_length = streams[0].recording_length + self.recording_interval = streams[0].recording_interval + + self._scheduler: threading.Thread | None = None + + self.has_recording_stream = any( + stream.stream_type == StreamTypeEnum.RECORDING for stream in self.streams + ) + + # 0 means interval recording is disabled + if ( + self.has_recording_stream + and self.recording_interval != 0 + and self.recording_length != 0 + and self.recording_interval <= self.recording_length + ): + self.emit_error( + "Recording interval must be greater than the recording length" + ) + + # Behavior: + # - Thread waits until stop thread is set + # - If it is set, we stop the recording prematurely + # - If it is not set, we continue with the interval + # Not relevant for non recording streams + self._stop_flag = threading.Event() + def start(self) -> None: + self.stop() + with self._lock: self.logger.info( "Starting stream for devices: " f"{', '.join([stream.device_path for stream in self.streams])}" ) - if self.started: - self.stop() self.started = True + + self._stop_flag.clear() + self._scheduler = threading.Thread(target=self._schedule_loop, daemon=True) + self._scheduler.start() + + def _schedule_loop(self) -> None: + if ( + self.has_recording_stream + and self.recording_interval != 0 + and self.recording_length != 0 + ): + while True: + self.emit("stream_active") + self.logger.info("Starting recording interval!") + + self._run_pipeline() + + if self._stop_flag.wait(self.recording_length): + break + + self.emit("stream_inactive") + self.logger.info("Stopping recording interval!") + self._terminate_process() + + if self._stop_flag.wait( + self.recording_interval - self.recording_length + ): + break + self._terminate_process() + else: + # Simply run it self._run_pipeline() def _run_pipeline(self) -> None: - pipeline_str = self._construct_pipeline() - self.logger.info(pipeline_str) - has_recording_stream = any( - stream.stream_type == StreamTypeEnum.RECORDING for stream in self.streams - ) - self._process = subprocess.Popen( - [ - "gst-launch-1.0", - f"{'-e' if has_recording_stream else ''}", # EOS on shutdown - *pipeline_str.split(" "), - ], - stdout=subprocess.DEVNULL, - stderr=subprocess.PIPE, - text=True, - ) - self._error_thread = threading.Thread(target=self._monitor_stderr) - self._error_thread.start() - - def stop(self) -> None: with self._lock: - if not self.started or not self._process: + if self._stop_flag.is_set(): return - self.logger.info("Stopping stream") - self.started = False + pipeline_str = self._construct_pipeline() + self.logger.info(pipeline_str) + has_recording_stream = any( + stream.stream_type == StreamTypeEnum.RECORDING + for stream in self.streams + ) + self._process = subprocess.Popen( + [ + "gst-launch-1.0", + f"{'-e' if has_recording_stream else ''}", # EOS on shutdown + *pipeline_str.split(" "), + ], + stdout=subprocess.DEVNULL, + stderr=subprocess.PIPE, + text=True, + ) + self._error_thread = threading.Thread(target=self._monitor_stderr) + self._error_thread.start() + + def _terminate_process(self) -> None: + self.logger.info("Trying to stop process") + with self._lock: + if not self._process: + return # For recording streams, send EOS to properly finalize the file has_recording_stream = any( @@ -181,10 +243,6 @@ def stop(self) -> None: else: self._process.terminate() self._process.wait(timeout=5) - except subprocess.TimeoutExpired: - self.logger.warning("Shutdown timed out, force killing...") - self._process.kill() - self._process.wait() except Exception as e: self.logger.error(f"Error during stop: {e}") self._process.kill() @@ -193,48 +251,56 @@ def stop(self) -> None: self._process.stderr.close() self._process = None + def stop(self) -> None: + with self._lock: + if not self.started: + return + + self.logger.info("Stopping stream") + self.started = False + + self._stop_flag.set() + + self._terminate_process() + + scheduler, self._scheduler = self._scheduler, None + + if scheduler and scheduler is not threading.current_thread(): + scheduler.join() + def _construct_pipeline(self) -> str: parts = [GStreamerPipelineBuilder.build(s) for s in self.streams] return " ".join(parts) def _monitor_stderr(self) -> None: - error_block = [] - - if not self._process or not self._process.stderr: + process = self._process + if not process or not process.stderr: self.logger.error( - "Unable to monitor stderr for process. " - "Is the GStreamer process running?" + "Unable to monitor stderr. Is the GStreamer process running?" ) return - for stderr_line in iter(self._process.stderr.readline, ""): - line_stripped = stderr_line.strip() - - # Log all stderr output but only stop on actual errors + error_block = [] + for line in iter(process.stderr.readline, ""): + stripped = line.strip() if any( - error_keyword in line_stripped.lower() - for error_keyword in ["error", "failed", "warning", "critical"] + k in stripped.lower() + for k in ("error", "failed", "warning", "critical") ): - error_block.append(line_stripped) + error_block.append(stripped) - if self._process: - self._process.wait() - return_code = self._process.returncode + return_code = process.wait() - if self.started and return_code != 0: - self.logger.error( - f"GStreamer process crashed with return code: {return_code}" - ) - - for error in error_block: - self.logger.error(error) - - # Construct error message - error_msg = f"Process exited with code {return_code}." - - self.emit_error(error_msg) + with self._lock: + # If _terminate_process() already cleared/replaced it, we stopped it + intentional = self._process is not process + if intentional or not self.started or return_code == 0: + return + self.started = False + self._process = None + self._stop_flag.set() - # Reset state - with self._lock: - self.started = False - self._process = None + self.logger.error(f"GStreamer process crashed with return code: {return_code}") + for error in error_block: + self.logger.error(error) + self.emit_error(f"Process exited with code {return_code}.") diff --git a/backend_py/src/services/cameras/stream_engines/stream.py b/backend_py/src/services/cameras/stream_engines/stream.py index 77c98aa1..284368d2 100644 --- a/backend_py/src/services/cameras/stream_engines/stream.py +++ b/backend_py/src/services/cameras/stream_engines/stream.py @@ -26,6 +26,8 @@ class Stream: default_factory=lambda: IntervalModel(numerator=1, denominator=30) ) enabled: bool = False + recording_interval: int = 0 + recording_length: int = 0 # Configuration specific software_h264_bitrate: int = 5000 diff --git a/backend_py/src/services/cameras/stream_runner.py b/backend_py/src/services/cameras/stream_runner.py index beb92683..b3644839 100644 --- a/backend_py/src/services/cameras/stream_runner.py +++ b/backend_py/src/services/cameras/stream_runner.py @@ -63,6 +63,10 @@ def start(self) -> None: self.engine.on("frame_drop", lambda: self.emit("frame_drop")) + # Forward over the interval emissions + self.engine.on("stream_active", lambda: self.emit("stream_active")) + self.engine.on("stream_inactive", lambda: self.emit("stream_inactive")) + self.started = True # We don't need to catch exceptions, maybe remove later try: 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..369007e6 100644 --- a/backend_py/src/services/recordings/recordings_service.py +++ b/backend_py/src/services/recordings/recordings_service.py @@ -80,7 +80,7 @@ def _epoch_to_readable(self, epoch: float) -> str: def _get_duration(self, file_path: str) -> str: if file_path in self.durations: - self.logger.info(f"Found cached duration: {file_path}") + # self.logger.info(f"Found cached duration: {file_path}") return self.durations[file_path] # FIXME: We need to change this function to use a better metadata library @@ -120,8 +120,9 @@ def _get_duration(self, file_path: str) -> str: 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 subprocess.CalledProcessError: + # self.logger.warning(f"exiftool had an unknown system error: {e}") + pass return "Unknown" def get_recording(self, filename: str) -> RecordingInfo | None: diff --git a/frontend/src/components/dwe/cameras/stream/interval-recording.tsx b/frontend/src/components/dwe/cameras/stream/interval-recording.tsx new file mode 100644 index 00000000..78e62b74 --- /dev/null +++ b/frontend/src/components/dwe/cameras/stream/interval-recording.tsx @@ -0,0 +1,131 @@ +import { + Accordion, + AccordionContent, + AccordionItem, + AccordionTrigger, +} from "@/components/ui/accordion"; +import { Input } from "@/components/ui/input"; +import { useDeviceStore } from "@/store/devices"; +import { useEffect, useState } from "react"; + +export const IntervalRecording = ({ + bus_id, + disabled, + onChange, +}: { + bus_id: string; + disabled: boolean; + onChange: (recording_interval: number, recording_length: number) => void; +}) => { + const isStreamLoading = useDeviceStore( + (state) => state.isStreamLoading[bus_id] ?? false, + ); + + // A value of 0 means the device records normally (no interval) + const savedInterval = useDeviceStore( + (state) => state.devices[bus_id]?.stream.recording_interval ?? 0, + ); + const savedLength = useDeviceStore( + (state) => state.devices[bus_id]?.stream.recording_length ?? 0, + ); + + const [enabled] = useState(true); + const [intervalSeconds, setIntervalSeconds] = useState(savedInterval); + const [durationSeconds, setDurationSeconds] = useState(savedLength); + + // Keep the inputs in sync with the device when it is updated elsewhere + useEffect(() => setIntervalSeconds(savedInterval), [savedInterval]); + useEffect(() => setDurationSeconds(savedLength), [savedLength]); + + const isDisabled = disabled || isStreamLoading; + const isContinuous = intervalSeconds === 0 || durationSeconds === 0; + const isValid = isContinuous || durationSeconds < intervalSeconds; + + const parseSeconds = (value: string) => Math.max(0, parseInt(value) || 0); + + const submit = () => { + if (!isValid) return; + if (intervalSeconds === savedInterval && durationSeconds === savedLength) + return; + onChange(intervalSeconds, durationSeconds); + }; + + return ( + + + + Interval Recording + + +
+
+ {/**/} + {/**/} +
+ +
+
+ + + setIntervalSeconds(parseSeconds(e.target.value)) + } + onBlur={submit} + /> +
+ +
+ + + setDurationSeconds(parseSeconds(e.target.value)) + } + onBlur={submit} + /> +
+
+ + {isContinuous ? ( +
+ Records continuously. Set both values above 0 to record on an + interval. +
+ ) : isValid ? ( +
+ Records for {durationSeconds} second + {durationSeconds === 1 ? "" : "s"} every {intervalSeconds}{" "} + second{intervalSeconds === 1 ? "" : "s"}. +
+ ) : ( +
+ Recording duration must be shorter than the interval. +
+ )} +
+ + + + ); +}; diff --git a/frontend/src/components/dwe/cameras/stream/stream.tsx b/frontend/src/components/dwe/cameras/stream/stream.tsx index f8a402e0..e31f3583 100644 --- a/frontend/src/components/dwe/cameras/stream/stream.tsx +++ b/frontend/src/components/dwe/cameras/stream/stream.tsx @@ -22,6 +22,7 @@ import { import { useCallback, useEffect, useState } from "react"; import { components } from "@/schemas/dwe_os_2"; import { FollowerList } from "./follower-list"; +import { IntervalRecording } from "./interval-recording"; import { CameraControls } from "../camera-controls"; import { toast } from "sonner"; @@ -138,7 +139,7 @@ export const CameraStream = ({ bus_id }: { bus_id: string }) => { width: device.stream.width, height: device.stream.height, interval: { - numerator: 1, + numerator: 1, denominator: parseInt(newFps), }, }, @@ -196,6 +197,14 @@ export const CameraStream = ({ bus_id }: { bus_id: string }) => { {canLead(device) && } + {device.stream.stream_type === "RECORDING" && ( + { + configureStream(device.bus_info, { + recording_interval, recording_length + }); + }} /> + )} +