Provide alternative camera urls (#1316)

kasa/__init__.py

@@ -40,6 +40,7 @@ from kasa.interfaces.thermostat import Thermostat, ThermostatState
 from kasa.module import Module
 from kasa.protocols import BaseProtocol, IotProtocol, SmartProtocol
 from kasa.protocols.iotprotocol import _deprecated_TPLinkSmartHomeProtocol  # noqa: F401
+from kasa.smartcam.modules.camera import StreamResolution
 from kasa.transports import BaseTransport
 
 __version__ = version("python-kasa")
@@ -75,6 +76,7 @@ __all__ = [
     "DeviceFamily",
     "ThermostatState",
     "Thermostat",
+    "StreamResolution",
 ]
 
 from . import iot

kasa/smartcam/modules/camera.py

@@ -4,6 +4,7 @@ from __future__ import annotations
 
 import base64
 import logging
+from enum import StrEnum
 from urllib.parse import quote_plus
 
 from ...credentials import Credentials
@@ -15,6 +16,14 @@ from ..smartcammodule import SmartCamModule
 _LOGGER = logging.getLogger(__name__)
 
 LOCAL_STREAMING_PORT = 554
+ONVIF_PORT = 2020
+
+
+class StreamResolution(StrEnum):
+    """Class for stream resolution."""
+
+    HD = "HD"
+    SD = "SD"
 
 
 class Camera(SmartCamModule):
@@ -64,7 +73,12 @@ class Camera(SmartCamModule):
 
         return None
 
-    def stream_rtsp_url(self, credentials: Credentials | None = None) -> str | None:
+    def stream_rtsp_url(
+        self,
+        credentials: Credentials | None = None,
+        *,
+        stream_resolution: StreamResolution = StreamResolution.HD,
+    ) -> str | None:
         """Return the local rtsp streaming url.
 
         :param credentials: Credentials for camera account.
@@ -73,17 +87,27 @@ class Camera(SmartCamModule):
         :return: rtsp url with escaped credentials or None if no credentials or
             camera is off.
         """
-        if not self.is_on:
+        streams = {
+            StreamResolution.HD: "stream1",
+            StreamResolution.SD: "stream2",
+        }
+        if (stream := streams.get(stream_resolution)) is None:
             return None
-        dev = self._device
+
         if not credentials:
             credentials = self._get_credentials()
 
         if not credentials or not credentials.username or not credentials.password:
             return None
+
         username = quote_plus(credentials.username)
         password = quote_plus(credentials.password)
-        return f"rtsp://{username}:{password}@{dev.host}:{LOCAL_STREAMING_PORT}/stream1"
+
+        return f"rtsp://{username}:{password}@{self._device.host}:{LOCAL_STREAMING_PORT}/{stream}"
+
+    def onvif_url(self) -> str | None:
+        """Return the onvif url."""
+        return f"http://{self._device.host}:{ONVIF_PORT}/onvif/device_service"
 
     async def set_state(self, on: bool) -> dict:
         """Set the device state."""

tests/smartcam/modules/test_camera.py

@@ -4,15 +4,13 @@ from __future__ import annotations
 
 import base64
 import json
-from datetime import UTC, datetime
 from unittest.mock import patch
 
 import pytest
-from freezegun.api import FrozenDateTimeFactory
 
-from kasa import Credentials, Device, DeviceType, Module
+from kasa import Credentials, Device, DeviceType, Module, StreamResolution
 
-from ..conftest import camera_smartcam, device_smartcam, hub_smartcam
+from ...conftest import camera_smartcam, device_smartcam
 
 
 @device_smartcam
@@ -37,6 +35,16 @@ async def test_stream_rtsp_url(dev: Device):
     url = camera_module.stream_rtsp_url(Credentials("foo", "bar"))
     assert url == "rtsp://foo:bar@127.0.0.123:554/stream1"
 
+    url = camera_module.stream_rtsp_url(
+        Credentials("foo", "bar"), stream_resolution=StreamResolution.HD
+    )
+    assert url == "rtsp://foo:bar@127.0.0.123:554/stream1"
+
+    url = camera_module.stream_rtsp_url(
+        Credentials("foo", "bar"), stream_resolution=StreamResolution.SD
+    )
+    assert url == "rtsp://foo:bar@127.0.0.123:554/stream2"
+
     with patch.object(dev.config, "credentials", Credentials("bar", "foo")):
         url = camera_module.stream_rtsp_url()
     assert url == "rtsp://bar:foo@127.0.0.123:554/stream1"
@@ -75,49 +83,12 @@ async def test_stream_rtsp_url(dev: Device):
         url = camera_module.stream_rtsp_url()
     assert url is None
 
-    # Test with camera off
-    await camera_module.set_state(False)
-    await dev.update()
-    url = camera_module.stream_rtsp_url(Credentials("foo", "bar"))
-    assert url is None
-    with patch.object(dev.config, "credentials", Credentials("bar", "foo")):
-        url = camera_module.stream_rtsp_url()
-    assert url is None
 
+@camera_smartcam
+async def test_onvif_url(dev: Device):
+    """Test the onvif url."""
+    camera_module = dev.modules.get(Module.Camera)
+    assert camera_module
 
-@device_smartcam
+    url = camera_module.onvif_url()
-async def test_alias(dev):
+    assert url == "http://127.0.0.123:2020/onvif/device_service"
-    test_alias = "TEST1234"
-    original = dev.alias
-
-    assert isinstance(original, str)
-    await dev.set_alias(test_alias)
-    await dev.update()
-    assert dev.alias == test_alias
-
-    await dev.set_alias(original)
-    await dev.update()
-    assert dev.alias == original
-
-
-@hub_smartcam
-async def test_hub(dev):
-    assert dev.children
-    for child in dev.children:
-        assert "Cloud" in child.modules
-        assert child.modules["Cloud"].data
-        assert child.alias
-        await child.update()
-        assert "Time" not in child.modules
-        assert child.time
-
-
-@device_smartcam
-async def test_device_time(dev: Device, freezer: FrozenDateTimeFactory):
-    """Test a child device gets the time from it's parent module."""
-    fallback_time = datetime.now(UTC).astimezone().replace(microsecond=0)
-    assert dev.time != fallback_time
-    module = dev.modules[Module.Time]
-    await module.set_time(fallback_time)
-    await dev.update()
-    assert dev.time == fallback_time

tests/smartcam/test_smartcamdevice.py

@@ -0,0 +1,61 @@
+"""Tests for smart camera devices."""
+
+from __future__ import annotations
+
+from datetime import UTC, datetime
+
+import pytest
+from freezegun.api import FrozenDateTimeFactory
+
+from kasa import Device, DeviceType, Module
+
+from ..conftest import device_smartcam, hub_smartcam
+
+
+@device_smartcam
+async def test_state(dev: Device):
+    if dev.device_type is DeviceType.Hub:
+        pytest.skip("Hubs cannot be switched on and off")
+
+    state = dev.is_on
+    await dev.set_state(not state)
+    await dev.update()
+    assert dev.is_on is not state
+
+
+@device_smartcam
+async def test_alias(dev):
+    test_alias = "TEST1234"
+    original = dev.alias
+
+    assert isinstance(original, str)
+    await dev.set_alias(test_alias)
+    await dev.update()
+    assert dev.alias == test_alias
+
+    await dev.set_alias(original)
+    await dev.update()
+    assert dev.alias == original
+
+
+@hub_smartcam
+async def test_hub(dev):
+    assert dev.children
+    for child in dev.children:
+        assert "Cloud" in child.modules
+        assert child.modules["Cloud"].data
+        assert child.alias
+        await child.update()
+        assert "Time" not in child.modules
+        assert child.time
+
+
+@device_smartcam
+async def test_device_time(dev: Device, freezer: FrozenDateTimeFactory):
+    """Test a child device gets the time from it's parent module."""
+    fallback_time = datetime.now(UTC).astimezone().replace(microsecond=0)
+    assert dev.time != fallback_time
+    module = dev.modules[Module.Time]
+    await module.set_time(fallback_time)
+    await dev.update()
+    assert dev.time == fallback_time