From c1d91500b56db8022067054a0c0c7e34b79cbd51 Mon Sep 17 00:00:00 2001 From: Nathan Spencer Date: Fri, 26 Apr 2024 12:25:57 -0600 Subject: [PATCH 1/2] Refactor grpc calls to reduce code --- vivintpy/vivintskyapi.py | 84 +++++++++++++++++++--------------------- 1 file changed, 39 insertions(+), 45 deletions(-) diff --git a/vivintpy/vivintskyapi.py b/vivintpy/vivintskyapi.py index e8e6647..ef2c2b2 100644 --- a/vivintpy/vivintskyapi.py +++ b/vivintpy/vivintskyapi.py @@ -14,6 +14,7 @@ import grpc from aiohttp import ClientResponseError from aiohttp.client import _RequestContextManager +from google.protobuf.message import Message from .const import ( AuthenticationResponse, @@ -139,21 +140,16 @@ async def reboot_camera( self, panel_id: int, device_id: int, device_type: str ) -> None: """Reboot a camera.""" - creds = grpc.ssl_channel_credentials() - assert (cookie := self._get_session_cookie()) - async with grpc.aio.secure_channel(GRPC_ENDPOINT, credentials=creds) as channel: - stub: beam_pb2_grpc.BeamStub = beam_pb2_grpc.BeamStub(channel) # type: ignore - response: beam_pb2.RebootCameraResponse = await stub.RebootCamera( + async def _cb(stub: beam_pb2_grpc.BeamStub, metadata: list[tuple[str, str]]): + return await stub.RebootCamera( beam_pb2.RebootCameraRequest( # pylint: disable=no-member - panel_id=panel_id, - device_id=device_id, - device_type=device_type, + panel_id=panel_id, device_id=device_id, device_type=device_type ), - metadata=[("session", cookie.value)], + metadata=metadata, ) - _LOGGER.debug("Response received: %s", str(response)) + await self._send_grpc(_cb) async def reboot_panel(self, panel_id: int) -> None: """Reboot a panel.""" @@ -204,63 +200,48 @@ async def set_camera_as_doorbell_chime_extender( self, panel_id: int, device_id: int, state: bool ) -> None: """Set the camera to be used as a doorbell chime extender.""" - creds = grpc.ssl_channel_credentials() - assert (cookie := self._get_session_cookie()) - async with grpc.aio.secure_channel(GRPC_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 - panel_id=panel_id, - device_id=device_id, - use_as_doorbell_chime_extender=state, - ), - metadata=[("session", cookie.value)], - ) + async def _cb(stub: beam_pb2_grpc.BeamStub, metadata: list[tuple[str, str]]): + return await stub.SetUseAsDoorbellChimeExtender( + beam_pb2.SetUseAsDoorbellChimeExtenderRequest( # pylint: disable=no-member + panel_id=panel_id, + device_id=device_id, + use_as_doorbell_chime_extender=state, + ), + metadata=metadata, ) - _LOGGER.debug("Response received: %s", str(response)) + await self._send_grpc(_cb) 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(GRPC_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)], - ) + async def _cb(stub: beam_pb2_grpc.BeamStub, metadata: list[tuple[str, str]]): + return await stub.SetCameraPrivacyMode( + beam_pb2.SetCameraPrivacyModeRequest( # pylint: disable=no-member + panel_id=panel_id, device_id=device_id, privacy_mode=state + ), + metadata=metadata, ) - _LOGGER.debug("Response received: %s", str(response)) + await self._send_grpc(_cb) async def set_camera_deter_mode( self, panel_id: int, device_id: int, state: bool ) -> None: """Set the camera deter mode.""" - creds = grpc.ssl_channel_credentials() - assert (cookie := self._get_session_cookie()) - async with grpc.aio.secure_channel(GRPC_ENDPOINT, credentials=creds) as channel: - stub: beam_pb2_grpc.BeamStub = beam_pb2_grpc.BeamStub(channel) # type: ignore - response: beam_pb2.SetDeterOverrideResponse = await stub.SetDeterOverride( + async def _cb(stub: beam_pb2_grpc.BeamStub, metadata: list[tuple[str, str]]): + return await stub.SetDeterOverride( beam_pb2.SetDeterOverrideRequest( # pylint: disable=no-member panel_id=panel_id, device_id=device_id, enabled=state ), - metadata=[("session", cookie.value)], + metadata=metadata, ) - _LOGGER.debug("Response received: %s", str(response)) + await self._send_grpc(_cb) async def set_garage_door_state( self, panel_id: int, partition_id: int, device_id: int, state: int @@ -551,3 +532,16 @@ async def __call( raise VivintSkyApiAuthenticationError(message) resp.raise_for_status() return None + + async def _send_grpc( + self, + callback: Callable[[beam_pb2_grpc.BeamStub, list[tuple[str, str]]], Message], + ): + """Send gRPC.""" + creds = grpc.ssl_channel_credentials() + assert (cookie := self._get_session_cookie()) + + async with grpc.aio.secure_channel(GRPC_ENDPOINT, credentials=creds) as channel: + stub: beam_pb2_grpc.BeamStub = beam_pb2_grpc.BeamStub(channel) # type: ignore + response = await callback(stub, [("session", cookie.value)]) + _LOGGER.debug("Response received: %s", str(response)) From 1a1190cf595300cc27c6caeb8d8b5870294c4a7d Mon Sep 17 00:00:00 2001 From: Nathan Spencer Date: Fri, 26 Apr 2024 12:33:16 -0600 Subject: [PATCH 2/2] Fix mypy issues --- vivintpy/vivintskyapi.py | 28 ++++++++++++++++++---------- 1 file changed, 18 insertions(+), 10 deletions(-) diff --git a/vivintpy/vivintskyapi.py b/vivintpy/vivintskyapi.py index ef2c2b2..cffe118 100644 --- a/vivintpy/vivintskyapi.py +++ b/vivintpy/vivintskyapi.py @@ -14,7 +14,7 @@ import grpc from aiohttp import ClientResponseError from aiohttp.client import _RequestContextManager -from google.protobuf.message import Message +from google.protobuf.message import Message # type: ignore from .const import ( AuthenticationResponse, @@ -141,7 +141,9 @@ async def reboot_camera( ) -> None: """Reboot a camera.""" - async def _cb(stub: beam_pb2_grpc.BeamStub, metadata: list[tuple[str, str]]): + async def _callback( + stub: beam_pb2_grpc.BeamStub, metadata: list[tuple[str, str]] + ) -> Message: return await stub.RebootCamera( beam_pb2.RebootCameraRequest( # pylint: disable=no-member panel_id=panel_id, device_id=device_id, device_type=device_type @@ -149,7 +151,7 @@ async def _cb(stub: beam_pb2_grpc.BeamStub, metadata: list[tuple[str, str]]): metadata=metadata, ) - await self._send_grpc(_cb) + await self._send_grpc(_callback) async def reboot_panel(self, panel_id: int) -> None: """Reboot a panel.""" @@ -201,7 +203,9 @@ async def set_camera_as_doorbell_chime_extender( ) -> None: """Set the camera to be used as a doorbell chime extender.""" - async def _cb(stub: beam_pb2_grpc.BeamStub, metadata: list[tuple[str, str]]): + async def _callback( + stub: beam_pb2_grpc.BeamStub, metadata: list[tuple[str, str]] + ) -> Message: return await stub.SetUseAsDoorbellChimeExtender( beam_pb2.SetUseAsDoorbellChimeExtenderRequest( # pylint: disable=no-member panel_id=panel_id, @@ -211,14 +215,16 @@ async def _cb(stub: beam_pb2_grpc.BeamStub, metadata: list[tuple[str, str]]): metadata=metadata, ) - await self._send_grpc(_cb) + await self._send_grpc(_callback) async def set_camera_privacy_mode( self, panel_id: int, device_id: int, state: bool ) -> None: """Set the camera privacy mode.""" - async def _cb(stub: beam_pb2_grpc.BeamStub, metadata: list[tuple[str, str]]): + async def _callback( + stub: beam_pb2_grpc.BeamStub, metadata: list[tuple[str, str]] + ) -> Message: return await stub.SetCameraPrivacyMode( beam_pb2.SetCameraPrivacyModeRequest( # pylint: disable=no-member panel_id=panel_id, device_id=device_id, privacy_mode=state @@ -226,14 +232,16 @@ async def _cb(stub: beam_pb2_grpc.BeamStub, metadata: list[tuple[str, str]]): metadata=metadata, ) - await self._send_grpc(_cb) + await self._send_grpc(_callback) async def set_camera_deter_mode( self, panel_id: int, device_id: int, state: bool ) -> None: """Set the camera deter mode.""" - async def _cb(stub: beam_pb2_grpc.BeamStub, metadata: list[tuple[str, str]]): + async def _callback( + stub: beam_pb2_grpc.BeamStub, metadata: list[tuple[str, str]] + ) -> Message: return await stub.SetDeterOverride( beam_pb2.SetDeterOverrideRequest( # pylint: disable=no-member panel_id=panel_id, device_id=device_id, enabled=state @@ -241,7 +249,7 @@ async def _cb(stub: beam_pb2_grpc.BeamStub, metadata: list[tuple[str, str]]): metadata=metadata, ) - await self._send_grpc(_cb) + await self._send_grpc(_callback) async def set_garage_door_state( self, panel_id: int, partition_id: int, device_id: int, state: int @@ -536,7 +544,7 @@ async def __call( async def _send_grpc( self, callback: Callable[[beam_pb2_grpc.BeamStub, list[tuple[str, str]]], Message], - ): + ) -> None: """Send gRPC.""" creds = grpc.ssl_channel_credentials() assert (cookie := self._get_session_cookie())