properly update settings in rtsp devices (#159)

* use new list of tuple return type of scan_for_cameras

* add motec_settings_interface

* add mjpeg camera tests

* add parameters to mjpeg camera

* require control_port to not be None

* implement async interface (with cooldowns)

* mark tests as asyncio

* remove usage of nicegui background task
this is necessary for usage without nicegui (for example in tests) since background_task depends on the nicegui core loop

* assume that getter and setter functions can be async
(and make them async in mjpeg_camera)

* ensure fixed event loop

* catch errors resulting from race conditions with a disconnect

* add docstrings

* remove unused import

* fix async errors in rtsp camera

* code review

* check if result is awaitable

* simplify start_capture_task

* tiny fix

* remove deprecated event_loop fixture

* fix missing value in call to setter

* fix rtsp_camera test

* make setter return type None or Awaitable[None]

* fix missing rename to polling_interval

* fix capture_task creation

* give ip to camera directly

* improve RtspDevice logging

* explicitly save if the camera should stream

* properly update settings used in the gstreamer pipeline

* fix issues left from merge

* remove debug print

* remove extra code left from merging

* use abstract setters and getters in RtspDevice instead of single set method
add jovision_profile field to RtspDevice

* always set jovision profile and type it as int

* fix imports

* remove setLevel(DEBUG)

* swap parameters

* handle case of process being None after wait

* remove unnecessary type hint

Co-authored-by: Falko Schindler <falko@zauberzeug.com>

* remove backslash

Co-authored-by: Falko Schindler <falko@zauberzeug.com>

* remove jovision_profile field

* calculate new url every time gstreamer is restarted

* add assertion to make mypy happy

* define url as a property

* add missing return statement

---------

Co-authored-by: Falko Schindler <falko@zauberzeug.com>

rosys/vision/rtsp_camera/rtsp_camera.py

@@ -34,13 +34,12 @@ def __init__(self,
         self.log = logging.getLogger(f'rosys.vision.rtsp_camera.{self.id}')
 
         self.device: RtspDevice | None = None
-        self.jovision_profile: int = jovision_profile
         self.ip: str | None = ip
 
-        self._register_parameter('fps', self.get_fps, self.set_fps,
-                                 min_value=1, max_value=30, step=1, default_value=fps)
         self._register_parameter('jovision_profile', self.get_jovision_profile, self.set_jovision_profile,
                                  min_value=1, max_value=2, step=1, default_value=jovision_profile)
+        self._register_parameter('fps', self.get_fps, self.set_fps,
+                                 min_value=1, max_value=30, step=1, default_value=fps)
 
     def to_dict(self) -> dict[str, Any]:
         return super().to_dict() | {
@@ -73,7 +72,7 @@ async def connect(self) -> None:
             self.log.error('no IP address provided for camera %s', self.id)
             return
 
-        self.device = RtspDevice(mac=self.id, ip=self.ip, jovision_profile=self.jovision_profile)
+        self.device = RtspDevice(mac=self.id, ip=self.ip, jovision_profile=self.parameters['jovision_profile'])
 
         await self._apply_all_parameters()
 
@@ -109,23 +108,24 @@ async def capture_image(self) -> None:
 
     def set_fps(self, fps: int) -> None:
         assert self.device is not None
-        assert self.device.settings_interface is not None
-        self.device.settings_interface.set_fps(stream_id=self.jovision_profile, fps=fps)
+
+        self.device.set_fps(fps)
         self.polling_interval = 1.0 / fps
 
     def get_fps(self) -> int | None:
         assert self.device is not None
-        assert self.device.settings_interface is not None
-        fps = self.device.settings_interface.get_fps(stream_id=self.jovision_profile)
-        return fps
+
+        return self.device.get_fps()
 
     def set_jovision_profile(self, profile: int) -> None:
         assert self.device is not None
-        self.jovision_profile = profile
+
+        self.device.set_jovision_profile(profile)
 
     def get_jovision_profile(self) -> int | None:
         assert self.device is not None
-        return self.jovision_profile
+
+        return self.device.get_jovision_profile()
 
     async def _apply_parameters(self, new_values: dict[str, Any], force_set: bool = False) -> None:
         await super()._apply_parameters(new_values, force_set)

rosys/vision/rtsp_camera/rtsp_device.py

@@ -17,9 +17,12 @@ class RtspDevice:
 
     def __init__(self, mac: str, ip: str, jovision_profile: int) -> None:
         self.log = logging.getLogger('rosys.vision.rtsp_camera.rtsp_device')
-        self.log.setLevel(logging.DEBUG)
 
         self.mac = mac
+        self.ip = ip
+
+        self.fps: int
+        self.jovision_profile = jovision_profile
 
         self.capture_task: asyncio.Task | None = None
         self.capture_process: Process | None = None
@@ -31,7 +34,7 @@ def __init__(self, mac: str, ip: str, jovision_profile: int) -> None:
         self.settings_interface: JovisionInterface | None = None
         if vendor_type == VendorType.JOVISION:
             self.settings_interface = JovisionInterface(ip)
-            self.fps = self.settings_interface.get_fps(stream_id=jovision_profile)
+            self.fps = self.settings_interface.get_fps(stream_id=jovision_profile) or 10
         else:
             self.log.warning('[%s] No settings interface for vendor type %s', self.mac, vendor_type)
             self.log.warning('[%s] Using default fps of 10', self.mac)
@@ -40,14 +43,20 @@ def __init__(self, mac: str, ip: str, jovision_profile: int) -> None:
         url = mac_to_url(mac, ip, jovision_profile)
         if url is None:
             raise ValueError(f'could not determine RTSP URL for {mac}')
-        self.url = url
-        self.log.info('[%s] Starting VideoStream for %s', self.mac, self.url)
+        self.log.info('[%s] Starting VideoStream for %s', self.mac, url)
         self._start_gstreamer_task()
 
     @property
     def authorized(self) -> bool:
         return self._authorized
 
+    @property
+    def url(self) -> str:
+        url = mac_to_url(self.mac, self.ip, self.jovision_profile)
+        if url is None:
+            raise ValueError(f'could not determine RTSP URL for {self.mac}')
+        return url
+
     def capture(self) -> bytes | None:
         image = self._image_buffer
         self._image_buffer = None
@@ -63,7 +72,7 @@ async def shutdown(self) -> None:
                 self.log.warning('[%s] Timeout while waiting for gstreamer process to terminate', self.mac)
             else:
                 self.log.debug('[%s] Successfully shut down process (code %s)',
-                               self.mac, self.capture_process.returncode)
+                               self.mac, self.capture_process.returncode if self.capture_process.returncode is not None else 'None')
                 self.capture_process = None
         if self.capture_task is not None and not self.capture_task.done():
             self.log.debug('[%s] Cancelling gstreamer task', self.mac)
@@ -84,18 +93,19 @@ def _start_gstreamer_task(self) -> None:
         if self.capture_task is not None and not self.capture_task.done():
             self.log.warning('[%s] capture task already running', self.mac)
             return
-        self.capture_task = background_tasks.create(self._run_gstreamer(self.url), name=f'capture {self.mac}')
+        self.capture_task = background_tasks.create(self._run_gstreamer(), name=f'capture {self.mac}')
 
     async def restart_gstreamer(self) -> None:
         await self.shutdown()
         self._start_gstreamer_task()
 
-    async def _run_gstreamer(self, url: str) -> None:
+    async def _run_gstreamer(self) -> None:
         if self.capture_process is not None and self.capture_process.returncode is None:
             self.log.warning('[%s] capture process already running', self.mac)
             return
 
-        async def stream(url: str) -> AsyncGenerator[bytes, None]:
+        async def stream() -> AsyncGenerator[bytes, None]:
+            url = self.url
             if 'subtype=0' in url:
                 url = url.replace('subtype=0', 'subtype=1')
 
@@ -168,9 +178,26 @@ async def stream(url: str) -> AsyncGenerator[bytes, None]:
                 if 'Unauthorized' in error_message:
                     self._authorized = False
 
-        async for image in stream(url):
+        async for image in stream():
             self._image_buffer = image
 
         self.log.info('[%s] stream ended', self.mac)
 
         self.capture_task = None
+
+    def set_fps(self, fps: int) -> None:
+        self.fps = fps
+
+        if self.settings_interface is not None:
+            self.settings_interface.set_fps(stream_id=self.jovision_profile, fps=self.fps)
+
+    def get_fps(self) -> int | None:
+        if self.settings_interface is not None:
+            return self.settings_interface.get_fps(stream_id=self.jovision_profile)
+        return self.fps
+
+    def set_jovision_profile(self, profile: int) -> None:
+        self.jovision_profile = profile
+
+    def get_jovision_profile(self) -> int:
+        return self.jovision_profile