diff --git a/pyproject.toml b/pyproject.toml index f7eeb7a..2b68a42 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -1,6 +1,6 @@ [tool.poetry] name = "vivintpy" -version = "2023.3.2" +version = "2023.3.3" description = "Python library for interacting with a Vivint security and smart home system." authors = ["Nathan Spencer "] license = "MIT" diff --git a/tests/test_version.py b/tests/test_version.py index fdfd877..f5eafcf 100644 --- a/tests/test_version.py +++ b/tests/test_version.py @@ -4,4 +4,4 @@ def test_version() -> None: """Test version.""" - assert __version__ == "2023.3.2" + assert __version__ == "2023.3.3" diff --git a/vivintpy/__init__.py b/vivintpy/__init__.py index 4c16f01..85cb00b 100644 --- a/vivintpy/__init__.py +++ b/vivintpy/__init__.py @@ -1,2 +1,2 @@ """Provide a package for vivintpy.""" -__version__ = "2023.3.2" +__version__ = "2023.3.3" diff --git a/vivintpy/devices/camera.py b/vivintpy/devices/camera.py index 47a5903..8b0474c 100644 --- a/vivintpy/devices/camera.py +++ b/vivintpy/devices/camera.py @@ -160,11 +160,17 @@ async def get_direct_rtsp_url( ) async def set_as_doorbell_chime_extender(self, state: bool) -> None: - """Set camera's use as doorbell chime extender.""" + """Set use as doorbell chime extender.""" await self.vivintskyapi.set_camera_as_doorbell_chime_extender( self.alarm_panel.id, self.id, state ) + async def set_privacy_mode(self, state: bool) -> None: + """Set privacy mode.""" + await self.vivintskyapi.set_camera_privacy_mode( + self.alarm_panel.id, self.id, state + ) + def handle_pubnub_message(self, message: dict) -> None: """Handle a pubnub message addressed to this camera.""" super().handle_pubnub_message(message) diff --git a/vivintpy/vivintskyapi.py b/vivintpy/vivintskyapi.py index 9995b1c..dea50a0 100644 --- a/vivintpy/vivintskyapi.py +++ b/vivintpy/vivintskyapi.py @@ -5,7 +5,8 @@ import logging import ssl from collections.abc import Callable -from typing import Any +from http.cookies import Morsel, SimpleCookie +from typing import Any, cast import aiohttp import certifi @@ -29,9 +30,9 @@ _LOGGER = logging.getLogger(__name__) -VIVINT_API_ENDPOINT = "https://www.vivintsky.com/api" -VIVINT_BEAM_ENDPOINT = "beam.vivintsky.com:443" -VIVINT_MFA_ENDPOINT = ( +API_ENDPOINT = "https://www.vivintsky.com/api" +BEAM_ENDPOINT = "beam.vivintsky.com:443" +MFA_ENDPOINT = ( "https://www.vivintsky.com/platform-user-api/v0/platformusers/2fa/validate" ) @@ -54,10 +55,14 @@ def __init__( self.__has_custom_client_session = client_session is not None self.__mfa_pending = False + def _get_session_cookie(self) -> Morsel | None: + """Get the session cookie.""" + cookie = self.__client_session.cookie_jar.filter_cookies(API_ENDPOINT) + return cast(SimpleCookie, cookie).get("s") + def is_session_valid(self) -> bool: """Return the state of the current session.""" - cookies = self.__client_session.cookie_jar.filter_cookies(VIVINT_API_ENDPOINT) - return cookies.get("s") is not None + return self._get_session_cookie() is not None async def connect(self) -> dict: """Connect to VivintSky Cloud Service.""" @@ -78,10 +83,7 @@ async def disconnect(self) -> None: async def verify_mfa(self, code: str) -> None: """Verify multi-factor authentication code.""" - resp = await self.__post( - VIVINT_MFA_ENDPOINT, - data=json.dumps({"code": code}), - ) + resp = await self.__post(MFA_ENDPOINT, data=json.dumps({"code": code})) if resp is not None: self.__mfa_pending = False @@ -182,18 +184,9 @@ async def set_camera_as_doorbell_chime_extender( ) -> None: """Set the camera to be used as a doorbell chime extender.""" creds = grpc.ssl_channel_credentials() - metad = [ - ( - "session", - self.__client_session.cookie_jar.filter_cookies(VIVINT_API_ENDPOINT) - .get("s") - .value, - ) - ] + assert (cookie := self._get_session_cookie()) - async with grpc.aio.secure_channel( - VIVINT_BEAM_ENDPOINT, credentials=creds - ) as channel: + async with grpc.aio.secure_channel(BEAM_ENDPOINT, credentials=creds) as channel: stub: beam_pb2_grpc.BeamStub = beam_pb2_grpc.BeamStub(channel) # type: ignore response: beam_pb2.SetUseAsDoorbellChimeExtenderResponse = await stub.SetUseAsDoorbellChimeExtender( beam_pb2.SetUseAsDoorbellChimeExtenderRequest( # pylint: disable=no-member @@ -201,7 +194,29 @@ async def set_camera_as_doorbell_chime_extender( device_id=device_id, use_as_doorbell_chime_extender=state, ), - metadata=metad, + metadata=[("session", cookie.value)], + ) + + _LOGGER.debug("Response received: %s", str(response)) + + async def set_camera_privacy_mode( + self, panel_id: int, device_id: int, state: bool + ) -> None: + """Set the camera privacy mode.""" + creds = grpc.ssl_channel_credentials() + assert (cookie := self._get_session_cookie()) + + async with grpc.aio.secure_channel(BEAM_ENDPOINT, credentials=creds) as channel: + stub: beam_pb2_grpc.BeamStub = beam_pb2_grpc.BeamStub(channel) # type: ignore + response: beam_pb2.SetCameraPrivacyModeResponse = ( + await stub.SetCameraPrivacyMode( + beam_pb2.SetCameraPrivacyModeRequest( # pylint: disable=no-member + panel_id=panel_id, + device_id=device_id, + privacy_mode=state, + ), + metadata=[("session", cookie.value)], + ) ) _LOGGER.debug("Response received: %s", str(response)) @@ -460,13 +475,13 @@ async def __call( if self.__client_session.closed: raise VivintSkyApiError("The client session has been closed") - is_mfa_request = path == VIVINT_MFA_ENDPOINT + is_mfa_request = path == MFA_ENDPOINT if self.__mfa_pending and not is_mfa_request: raise VivintSkyApiMfaRequiredError(AuthenticationResponse.MFA_REQUIRED) resp = await method( - path if is_mfa_request else f"{VIVINT_API_ENDPOINT}/{path}", + path if is_mfa_request else f"{API_ENDPOINT}/{path}", headers=headers, params=params, data=data,