diff --git a/kasa/__init__.py b/kasa/__init__.py index d4a5022e..ee52eb3a 100755 --- a/kasa/__init__.py +++ b/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 diff --git a/kasa/smartcam/modules/camera.py b/kasa/smartcam/modules/camera.py index 815db62b..e96794c2 100644 --- a/kasa/smartcam/modules/camera.py +++ b/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.""" diff --git a/tests/smartcam/test_smartcamera.py b/tests/smartcam/modules/test_camera.py similarity index 57% rename from tests/smartcam/test_smartcamera.py rename to tests/smartcam/modules/test_camera.py index ccb4fbc1..ebc08101 100644 --- a/tests/smartcam/test_smartcamera.py +++ b/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 -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 + url = camera_module.onvif_url() + assert url == "http://127.0.0.123:2020/onvif/device_service" diff --git a/tests/smartcam/test_smartcamdevice.py b/tests/smartcam/test_smartcamdevice.py new file mode 100644 index 00000000..438737eb --- /dev/null +++ b/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