dahua profile setting

rosys/helpers/deprecation.py

@@ -0,0 +1,42 @@
+import warnings
+from collections.abc import Callable
+from functools import wraps
+
+
+def deprecated_param(param_name: str, *, remove_in_version: str | None = None, stacklevel: int = 2) -> Callable:
+    """Mark a function parameter as deprecated."""
+    def decorator(func: Callable) -> Callable:
+        @wraps(func)
+        def wrapper(*args, **kwargs):
+            if param_name in kwargs:
+                warnings.warn(
+                    f'The parameter "{param_name}" is deprecated and will be removed in '
+                    f'{("RoSys " + remove_in_version) if remove_in_version else "a future version"}.',
+                    category=DeprecationWarning,
+                    stacklevel=stacklevel,
+                )
+            return func(*args, **kwargs)
+        return wrapper
+    return decorator
+
+
+def deprecated_function(*, remove_in_version: str | None = None, stacklevel: int = 2) -> Callable:
+    """Mark a function as deprecated."""
+    def decorator(func: Callable) -> Callable:
+        warnings.warn(f'The function "{func.__name__}" is deprecated and will be removed in '
+                      f'{("RoSys " + remove_in_version) if remove_in_version else "a future version"}.',
+                      category=DeprecationWarning,
+                      stacklevel=stacklevel)
+        return func
+    return decorator
+
+
+def deprecated_class(*, remove_in_version: str | None = None, stacklevel: int = 2) -> Callable:
+    """Mark a class as deprecated."""
+    def decorator(cls: type) -> type:
+        warnings.warn(f'The class "{cls.__name__}" is deprecated and will be removed in '
+                      f'{("RoSys " + remove_in_version) if remove_in_version else "a future version"}.',
+                      category=DeprecationWarning,
+                      stacklevel=stacklevel)
+        return cls
+    return decorator

rosys/rosys.py

@@ -6,6 +6,7 @@
 import signal
 import threading
 import time as pytime
+import warnings
 from collections.abc import Awaitable, Callable
 from dataclasses import dataclass
 from typing import Any, ClassVar, Literal
@@ -21,6 +22,8 @@
 from .helpers import is_test as is_test_
 from .persistence.registry import backup, restore, sync_backup
 
+warnings.filterwarnings('once', category=DeprecationWarning, module='rosys')
+
 log = logging.getLogger('rosys.core')
 
 config = Config()

rosys/vision/rtsp_camera/rtsp_camera.py

@@ -4,6 +4,7 @@
 from typing_extensions import Self
 
 from ... import rosys
+from ...helpers.deprecation import deprecated_function, deprecated_param
 from ..camera.configurable_camera import ConfigurableCamera
 from ..camera.transformable_camera import TransformableCamera
 from ..image import Image
@@ -13,13 +14,15 @@
 
 class RtspCamera(ConfigurableCamera, TransformableCamera):
 
+    @deprecated_param('jovision_profile', remove_in_version='0.27.0')
     def __init__(self,
                  *,
                  id: str,  # pylint: disable=redefined-builtin
                  name: str | None = None,
                  connect_after_init: bool = True,
                  fps: int = 5,
-                 jovision_profile: int = 1,
+                 substream: int = 1,
+                 jovision_profile: int | None = None,
                  bitrate: int = 4096,
                  ip: str | None = None,
                  **kwargs,
@@ -34,22 +37,30 @@ def __init__(self,
         self.device: RtspDevice | None = None
         self.ip: str | None = ip
 
+        self.substream = jovision_profile or substream
         self._register_parameter('jovision_profile', self.get_jovision_profile, self.set_jovision_profile,
-                                 min_value=0, max_value=1, step=1, default_value=jovision_profile)
+                                 min_value=0, max_value=1, step=1, default_value=substream)
+        self._register_parameter('substream', self.get_substream, self.set_substream,
+                                 min_value=0, max_value=1, step=1, default_value=substream)
         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() | {
+        parameters = {
             name: param.value for name, param in self._parameters.items()
-        } | {
+        }
+        if 'jovision_profile' in parameters:
+            del parameters['jovision_profile']  # DEPRECATED: 0.27.0
+        return super().to_dict() | parameters | {
             'ip': self.ip,
         }
 
     @classmethod
     def from_dict(cls, data: dict[str, Any]) -> Self:
+        if 'jovision_profile' in data:
+            data['substream'] = data['jovision_profile']
         return cls(**data)
 
     @property
@@ -73,7 +84,7 @@ async def connect(self) -> None:
             return
 
         self.device = RtspDevice(mac=self.id, ip=self.ip,
-                                 jovision_profile=self.parameters['jovision_profile'],
+                                 substream=self.parameters['jovision_profile'],
                                  fps=self.parameters['fps'],
                                  on_new_image_data=self._handle_new_image_data)
 
@@ -113,15 +124,30 @@ async def get_fps(self) -> int | None:
 
         return await self.device.get_fps()
 
+    @deprecated_function(remove_in_version='0.27.0', stacklevel=3)
+    def get_jovision_profile(self) -> int | None:
+        assert self.device is not None
+        profile = self.get_substream()
+        if profile is not None:
+            self.parameters['substream'].value = profile
+        return profile
+
+    @deprecated_function(remove_in_version='0.27.0', stacklevel=3)
     def set_jovision_profile(self, profile: int) -> None:
         assert self.device is not None
+        self.parameters['substream'].value = profile
 
-        self.device.set_jovision_profile(profile)
+        self.device.set_substream(profile)
 
-    def get_jovision_profile(self) -> int | None:
+    def set_substream(self, index: int) -> None:
+        assert self.device is not None
+
+        self.device.set_substream(index)
+
+    def get_substream(self) -> int | None:
         assert self.device is not None
 
-        return self.device.get_jovision_profile()
+        return self.device.get_substream()
 
     async def set_bitrate(self, bitrate: int) -> None:
         assert self.device is not None

rosys/vision/rtsp_camera/rtsp_camera_provider.py

@@ -1,6 +1,7 @@
 import logging
 
 from ... import persistence, rosys
+from ...helpers.deprecation import deprecated_param
 from ..camera_provider import CameraProvider
 from .arp_scan import find_known_cameras
 from .rtsp_camera import RtspCamera
@@ -10,15 +11,17 @@ class RtspCameraProvider(CameraProvider[RtspCamera], persistence.PersistentModul
     """This module collects and provides real RTSP streaming cameras."""
     SCAN_INTERVAL = 10
 
+    @deprecated_param('jovision_profile', remove_in_version='0.27.0')
     def __init__(self, *,
                  frame_rate: int = 6,
-                 jovision_profile: int = 0,
+                 substream: int = 0,
+                 jovision_profile: int | None = None,
                  network_interface: str | None = None,
                  auto_scan: bool = True) -> None:
         super().__init__()
 
         self.frame_rate = frame_rate
-        self.jovision_profile = jovision_profile
+        self.substream = jovision_profile if jovision_profile is not None else substream
         self.network_interface = network_interface
 
         self.log = logging.getLogger('rosys.rtsp_camera_provider')
@@ -49,7 +52,7 @@ async def update_device_list(self) -> None:
         for mac, ip in await find_known_cameras(network_interface=self.network_interface):
             if mac not in self._cameras:
                 self.log.debug('found new camera %s', mac)
-                self.add_camera(RtspCamera(id=mac, fps=self.frame_rate, jovision_profile=self.jovision_profile, ip=ip))
+                self.add_camera(RtspCamera(id=mac, fps=self.frame_rate, substream=self.substream, ip=ip))
             camera = self._cameras[mac]
             if not camera.is_connected:
                 self.log.info('activating authorized camera %s...', camera.id)

rosys/vision/rtsp_camera/rtsp_device.py

@@ -17,13 +17,13 @@
 class RtspDevice:
 
     def __init__(self, mac: str, ip: str, *,
-                 jovision_profile: int, fps: int, on_new_image_data: Callable[[bytes, float], Awaitable | None]) -> None:
+                 substream: int, fps: int, on_new_image_data: Callable[[bytes, float], Awaitable | None]) -> None:
         self._mac = mac
         self._ip = ip
         self.log = logging.getLogger('rosys.vision.rtsp_camera.rtsp_device.' + self._mac)
 
         self._fps = fps
-        self._jovision_profile = jovision_profile
+        self._substream = substream
         self._on_new_image_data = on_new_image_data
 
         self._capture_task: asyncio.Task | None = None
@@ -52,7 +52,7 @@ def authorized(self) -> bool:
 
     @property
     def url(self) -> str:
-        url = mac_to_url(self._mac, self._ip, self._jovision_profile)
+        url = mac_to_url(self._mac, self._ip, self._substream)
         if url is None:
             raise ValueError(f'could not determine RTSP URL for {self._mac}')
         return url
@@ -101,9 +101,6 @@ async def _run_gstreamer(self) -> None:
 
         async def stream() -> AsyncGenerator[bytes, None]:
             url = self.url
-            if 'subtype=0' in url:
-                url = url.replace('subtype=0', 'subtype=1')
-
             self.log.debug('[%s] Starting gstreamer pipeline for %s', self._mac, url)
             # to try: replace avdec_h264 with nvh264dec ! nvvidconv (!videoconvert)
             command = f'gst-launch-1.0 rtspsrc location="{url}" latency=0 protocols=tcp ! rtph264depay ! avdec_h264 ! videoconvert ! jpegenc ! fdsink'
@@ -185,24 +182,24 @@ async def set_fps(self, fps: int) -> None:
         self._fps = fps
 
         if self._settings_interface is not None:
-            await self._settings_interface.set_fps(stream_id=self._jovision_profile, fps=self._fps)
+            await self._settings_interface.set_fps(stream_id=self._substream, fps=self._fps)
 
     async def get_fps(self) -> int | None:
         if self._settings_interface is not None:
-            return await self._settings_interface.get_fps(stream_id=self._jovision_profile)
+            return await self._settings_interface.get_fps(stream_id=self._substream)
         return self._fps
 
-    def set_jovision_profile(self, profile: int) -> None:
-        self._jovision_profile = profile
+    def set_substream(self, index: int) -> None:
+        self._substream = index
 
-    def get_jovision_profile(self) -> int:
-        return self._jovision_profile
+    def get_substream(self) -> int:
+        return self._substream
 
     async def set_bitrate(self, bitrate: int) -> None:
         if self._settings_interface is not None:
-            await self._settings_interface.set_bitrate(stream_id=self._jovision_profile, bitrate=bitrate)
+            await self._settings_interface.set_bitrate(stream_id=self._substream, bitrate=bitrate)
 
     async def get_bitrate(self) -> int | None:
         if self._settings_interface is not None:
-            return await self._settings_interface.get_bitrate(stream_id=self._jovision_profile)
+            return await self._settings_interface.get_bitrate(stream_id=self._substream)
         return None

rosys/vision/rtsp_camera/vendors.py

@@ -16,10 +16,10 @@ def mac_to_vendor(mac: str) -> VendorType:
     return VendorType.OTHER
 
 
-def mac_to_url(mac: str, ip: str, jovision_profile: int = 0) -> str | None:
+def mac_to_url(mac: str, ip: str, substream: int = 0) -> str | None:
     vendor = mac_to_vendor(mac)
     if vendor == VendorType.JOVISION:
-        return f'rtsp://admin:admin@{ip}/profile{jovision_profile}'
+        return f'rtsp://admin:admin@{ip}/profile{substream}'
     if vendor == VendorType.DAHUA:
-        return f'rtsp://admin:Adminadmin@{ip}/cam/realmonitor?channel=1&subtype=0'
+        return f'rtsp://admin:Adminadmin@{ip}/cam/realmonitor?channel=1&subtype={substream}'
     return None