add bitrate parameter to RTSP cameras

rosys/vision/rtsp_camera/jovision_rtsp_interface.py

@@ -43,10 +43,6 @@ def __init__(self, ip: str) -> None:
     def settings_url(self) -> str:
         return f'http://{self.ip}/cgi-bin/jvsweb.cgi'
 
-    @property
-    def headers(self) -> dict[str, str]:
-        return {'Cookie': 'passwdRule=1; username=admin; password=admin;'}
-
     def _send_settings(self, settings: list[JovisionCameraSettings]) -> None:
         """
         Set the settings for all streams
@@ -64,32 +60,47 @@ def _send_settings(self, settings: list[JovisionCameraSettings]) -> None:
                 'digest': '479ead4555b49227dc8812d06970c652',
             },
             'param': {
-                'channelid': 0,
                 'streams': streams,
             },
         }
         params = {
             'cmd': json.dumps(cmd),
             '_': int(time.time() * 1000),  # current time as a timestamp
         }
-        requests.get(self.settings_url, params=params, headers=self.headers, timeout=1)  # type: ignore
+        requests.get(self.settings_url, params=params, timeout=1)  # type: ignore
 
-    def set_fps(self, stream_id, fps) -> None:
+    def set_fps(self, stream_id: int, fps: int) -> None:
         current_settings = self.get_current_settings()
         for settings in current_settings:
             if settings.stream_id == stream_id:
-                settings.fps = fps
+                settings.fps = int(fps)
                 break
         self._send_settings(current_settings)
 
-    def get_fps(self, stream_id) -> int | None:
+    def get_fps(self, stream_id: int) -> int | None:
         current_settings = self.get_current_settings()
         for settings in current_settings:
             if settings.stream_id == stream_id:
                 return settings.fps
 
         return None
 
+    def get_bitrate(self, stream_id: int) -> int | None:
+        current_settings = self.get_current_settings()
+        for settings in current_settings:
+            if settings.stream_id == stream_id:
+                return settings.bitrate
+
+        return None
+
+    def set_bitrate(self, stream_id: int, bitrate: int) -> None:
+        current_settings = self.get_current_settings()
+        for settings in current_settings:
+            if settings.stream_id == stream_id:
+                settings.bitrate = int(bitrate)
+                break
+        self._send_settings(current_settings)
+
     def get_current_settings(self) -> list[JovisionCameraSettings]:
         cmd = {
             'method': 'stream_get_params',
@@ -105,7 +116,7 @@ def get_current_settings(self) -> list[JovisionCameraSettings]:
             'cmd': json.dumps(cmd),
             '_': int(time.time() * 1000),
         }
-        response = requests.get(self.settings_url, params=params, headers=self.headers, timeout=1)  # type: ignore
+        response = requests.get(self.settings_url, params=params, timeout=1)  # type: ignore
 
         return [
             JovisionCameraSettings(
@@ -139,7 +150,7 @@ def get_parameter_ranges(self):
             'cmd': json.dumps(cmd),
             '_': time.time() * 1000,
         }
-        response = requests.get(self.settings_url, params=params, headers=self.headers, timeout=1)
+        response = requests.get(self.settings_url, params=params, timeout=1)
 
         for stream_id, stream in enumerate(response.json()['result']['all']):
             print(f'stream {stream_id}')

rosys/vision/rtsp_camera/rtsp_camera.py

@@ -21,6 +21,7 @@ def __init__(self,
                  streaming: bool = True,
                  fps: int = 5,
                  jovision_profile: int = 1,
+                 bitrate: int = 4096,
                  ip: str | None = None,
                  **kwargs,
                  ) -> None:
@@ -37,9 +38,11 @@ def __init__(self,
         self.ip: str | None = ip
 
         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)
+                                 min_value=0, max_value=1, 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)
+        self._register_parameter('bitrate', self.get_bitrate, self.set_bitrate,
+                                 min_value=32, max_value=8192, step=1, default_value=bitrate)
 
     def to_dict(self) -> dict[str, Any]:
         return super().to_dict() | {
@@ -127,6 +130,16 @@ def get_jovision_profile(self) -> int | None:
 
         return self.device.get_jovision_profile()
 
+    def set_bitrate(self, bitrate: int) -> None:
+        assert self.device is not None
+
+        self.device.set_bitrate(bitrate)
+
+    def get_bitrate(self) -> int | None:
+        assert self.device is not None
+
+        return self.device.get_bitrate()
+
     async def _apply_parameters(self, new_values: dict[str, Any], force_set: bool = False) -> None:
         await super()._apply_parameters(new_values, force_set)
         if self.is_connected:

rosys/vision/rtsp_camera/rtsp_device.py

@@ -201,3 +201,12 @@ def set_jovision_profile(self, profile: int) -> None:
 
     def get_jovision_profile(self) -> int:
         return self.jovision_profile
+
+    def set_bitrate(self, bitrate: int) -> None:
+        if self.settings_interface is not None:
+            self.settings_interface.set_bitrate(stream_id=self.jovision_profile, bitrate=bitrate)
+
+    def get_bitrate(self) -> int | None:
+        if self.settings_interface is not None:
+            return self.settings_interface.get_bitrate(stream_id=self.jovision_profile)
+        return None