From 1e0ca799bc516503918b6957eb95dbf069f3644c Mon Sep 17 00:00:00 2001 From: "Steven B." <51370195+sdb9696@users.noreply.github.com> Date: Fri, 25 Oct 2024 18:30:21 +0100 Subject: [PATCH] Add stream_rtsp_url to camera module (#1197) --- kasa/experimental/modules/camera.py | 30 ++++++++++++++-- kasa/experimental/sslaestransport.py | 10 +++--- kasa/tests/smartcamera/test_smartcamera.py | 42 ++++++++++++++++++++-- 3 files changed, 74 insertions(+), 8 deletions(-) diff --git a/kasa/experimental/modules/camera.py b/kasa/experimental/modules/camera.py index 76701b52..ecd7fff7 100644 --- a/kasa/experimental/modules/camera.py +++ b/kasa/experimental/modules/camera.py @@ -2,10 +2,15 @@ from __future__ import annotations +from urllib.parse import quote_plus + +from ...credentials import Credentials from ...device_type import DeviceType from ...feature import Feature from ..smartcameramodule import SmartCameraModule +LOCAL_STREAMING_PORT = 554 + class Camera(SmartCameraModule): """Implementation of device module.""" @@ -31,11 +36,32 @@ class Camera(SmartCameraModule): @property def is_on(self) -> bool: """Return the device id.""" - return self.data["lens_mask_info"]["enabled"] == "on" + return self.data["lens_mask_info"]["enabled"] == "off" + + def stream_rtsp_url(self, credentials: Credentials | None = None) -> str | None: + """Return the local rtsp streaming url. + + :param credentials: Credentials for camera account. + These could be different credentials to tplink cloud credentials. + If not provided will use tplink credentials if available + :return: rtsp url with escaped credentials or None if no credentials or + camera is off. + """ + if not self.is_on: + return None + dev = self._device + if not credentials: + credentials = dev.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" async def set_state(self, on: bool) -> dict: """Set the device state.""" - params = {"enabled": "on" if on else "off"} + # Turning off enables the privacy mask which is why value is reversed. + params = {"enabled": "off" if on else "on"} return await self._device._query_setter_helper( "setLensMaskConfig", self.QUERY_MODULE_NAME, "lens_mask_info", params ) diff --git a/kasa/experimental/sslaestransport.py b/kasa/experimental/sslaestransport.py index f095a11e..2a5d12e2 100644 --- a/kasa/experimental/sslaestransport.py +++ b/kasa/experimental/sslaestransport.py @@ -120,11 +120,13 @@ class SslAesTransport(BaseTransport): self._seq: int | None = None self._pwd_hash: str | None = None self._username: str | None = None + self._password: str | None = None if self._credentials != Credentials() and self._credentials: self._username = self._credentials.username + self._password = self._credentials.password elif self._credentials_hash: ch = json_loads(base64.b64decode(self._credentials_hash.encode())) - self._pwd_hash = ch["pwd"] + self._password = ch["pwd"] self._username = ch["un"] self._local_nonce: str | None = None @@ -140,10 +142,10 @@ class SslAesTransport(BaseTransport): """The hashed credentials used by the transport.""" if self._credentials == Credentials(): return None - if self._credentials_hash: + if not self._credentials and self._credentials_hash: return self._credentials_hash - if self._pwd_hash and self._credentials: - ch = {"un": self._credentials.username, "pwd": self._pwd_hash} + if (cred := self._credentials) and cred.password and cred.username: + ch = {"un": cred.username, "pwd": cred.password} return base64.b64encode(json_dumps(ch).encode()).decode() return None diff --git a/kasa/tests/smartcamera/test_smartcamera.py b/kasa/tests/smartcamera/test_smartcamera.py index 3e12dcfb..1185943a 100644 --- a/kasa/tests/smartcamera/test_smartcamera.py +++ b/kasa/tests/smartcamera/test_smartcamera.py @@ -3,13 +3,14 @@ from __future__ import annotations from datetime import datetime, timezone +from unittest.mock import patch import pytest from freezegun.api import FrozenDateTimeFactory -from kasa import Device, DeviceType, Module +from kasa import Credentials, Device, DeviceType, Module -from ..conftest import device_smartcamera, hub_smartcamera +from ..conftest import camera_smartcamera, device_smartcamera, hub_smartcamera @device_smartcamera @@ -23,6 +24,43 @@ async def test_state(dev: Device): assert dev.is_on is not state +@camera_smartcamera +async def test_stream_rtsp_url(dev: Device): + camera_module = dev.modules.get(Module.Camera) + assert camera_module + + await camera_module.set_state(True) + await dev.update() + assert camera_module.is_on + url = camera_module.stream_rtsp_url(Credentials("foo", "bar")) + assert url == "rtsp://foo:bar@127.0.0.123:554/stream1" + + with patch.object( + dev.protocol._transport, "_credentials", Credentials("bar", "foo") + ): + url = camera_module.stream_rtsp_url() + assert url == "rtsp://bar:foo@127.0.0.123:554/stream1" + + with patch.object(dev.protocol._transport, "_credentials", Credentials("bar", "")): + url = camera_module.stream_rtsp_url() + assert url is None + + with patch.object(dev.protocol._transport, "_credentials", Credentials("", "Foo")): + 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.protocol._transport, "_credentials", Credentials("bar", "foo") + ): + url = camera_module.stream_rtsp_url() + assert url is None + + @device_smartcamera async def test_alias(dev): test_alias = "TEST1234"