Merge pull request #159 from natekspencer/dev

Refactor gRPC calls to reduce code

vivintpy/vivintskyapi.py

@@ -14,6 +14,7 @@
 import grpc
 from aiohttp import ClientResponseError
 from aiohttp.client import _RequestContextManager
+from google.protobuf.message import Message # type: ignore
 
 from .const import (
  AuthenticationResponse,
@@ -139,21 +140,18 @@ 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 _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,
+ 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(_callback)
 
  async def reboot_panel(self, panel_id: int) -> None:
  """Reboot a panel."""
@@ -204,63 +202,54 @@ 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 _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,
+ device_id=device_id,
+ use_as_doorbell_chime_extender=state,
+ ),
+ metadata=metadata,
  )
 
- _LOGGER.debug("Response received: %s", str(response))
+ 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."""
- 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 _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
+ ),
+ metadata=metadata,
  )
 
- _LOGGER.debug("Response received: %s", str(response))
+ 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."""
- 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 _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
  ),
- metadata=[("session", cookie.value)],
+ metadata=metadata,
  )
 
- _LOGGER.debug("Response received: %s", str(response))
+ await self._send_grpc(_callback)
 
  async def set_garage_door_state(
  self, panel_id: int, partition_id: int, device_id: int, state: int
@@ -551,3 +540,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],
+ ) -> None:
+ """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))