Types refactoring

custom_components/gsm_call/hardware/at_dialer.py

@@ -2,10 +2,12 @@
 # License, v. 2.0. If a copy of the MPL was not distributed with this
 # file, You can obtain one at https://mozilla.org/MPL/2.0/.
 
-from asyncio import StreamReader, StreamWriter, sleep
-from typing import Tuple
+import asyncio
+
+from homeassistant.exceptions import HomeAssistantError
 
 from ..const import _LOGGER
+from ..modem import READ_LIMIT, Modem
 
 
 class ATDialer:
@@ -14,24 +16,25 @@ class ATDialer:
     def __init__(self, call_duration):
         self.call_duration = call_duration
 
-    async def dial(self, modem: Tuple[StreamReader, StreamWriter], phone_number: str):
-        reader, writer = modem
-
+    async def dial(self, modem: Modem, phone_number: str):
         _LOGGER.debug(f"Dialing +{phone_number}...")
-        writer.write(f"{self.at_command}+{phone_number};\r\n".encode())
+        modem.writer.write(f"{self.at_command}+{phone_number};\r\n".encode())
 
-        await sleep(1)
+        await asyncio.sleep(1)
         _LOGGER.debug("Reading from modem...")
-        buf = await reader.read(128)
+        buf = await modem.reader.read(READ_LIMIT)
         reply = buf.decode().strip()
         _LOGGER.debug(f"Modem replied with ${reply}")
 
-        if "ERROR" in reply:
-            raise Exception("Modem replied with an unknown error")
+        if "ERROR" in reply or "NO CARRIER" in reply:
+            raise HomeAssistantError("Modem replied with an unknown error")
+
+        if "BUSY" in reply:
+            raise HomeAssistantError("Busy")
 
         sleep_duration = self.call_duration + 5
         _LOGGER.info(f"Ringing for {sleep_duration} seconds...")
-        await sleep(sleep_duration)
+        await asyncio.sleep(sleep_duration)
 
         _LOGGER.debug("Hanging up...")
-        writer.write(b"AT+CHUP\r\n")
+        modem.writer.write(b"AT+CHUP\r\n")

custom_components/gsm_call/modem.py

@@ -0,0 +1,21 @@
+# This Source Code Form is subject to the terms of the Mozilla Public
+# License, v. 2.0. If a copy of the MPL was not distributed with this
+# file, You can obtain one at https://mozilla.org/MPL/2.0/.
+
+from asyncio import StreamReader, StreamWriter
+from typing import Tuple
+
+import serial_asyncio_fast as serial_asyncio
+
+READ_LIMIT = 2**16  # 64 KiB
+
+
+class Modem:
+    reader: StreamReader
+    writer: StreamWriter
+    serial: serial_asyncio.serial.Serial
+
+    def __init__(self, connection: Tuple[StreamReader, StreamWriter]):
+        self.reader = connection[0]
+        self.writer = connection[1]
+        self.serial = self.writer.transport.get_extra_info("serial")

custom_components/gsm_call/notify.py

@@ -5,8 +5,6 @@
 from __future__ import annotations
 
 import re
-from asyncio import StreamReader, StreamWriter
-from typing import Tuple
 
 import homeassistant.helpers.config_validation as cv
 import serial_asyncio_fast as serial_asyncio
@@ -29,6 +27,7 @@
 from custom_components.gsm_call.hardware.at_dialer import ATDialer
 from custom_components.gsm_call.hardware.at_tone_dialer import ATToneDialer
 from custom_components.gsm_call.hardware.zte_dialer import ZTEDialer
+from custom_components.gsm_call.modem import READ_LIMIT, Modem
 
 SUPPORTED_HARDWARE = {
     "atd": ATDialer,
@@ -66,7 +65,7 @@ def get_service(
 
 
 class GsmCallNotificationService(BaseNotificationService):
-    modem: Tuple[StreamReader, StreamWriter] | None = None
+    modem: Modem | None = None
 
     def __init__(self, device_path, dialer):
         self.device_path = device_path
@@ -98,21 +97,24 @@ async def async_send_message(self, message="", **kwargs):
 
     async def connect(self):
         _LOGGER.debug(f"Connecting to {self.device_path}...")
-        GsmCallNotificationService.modem = await serial_asyncio.open_serial_connection(
-            url=self.device_path,
-            baudrate=75600,
-            bytesize=serial_asyncio.serial.EIGHTBITS,
-            parity=serial_asyncio.serial.PARITY_NONE,
-            stopbits=serial_asyncio.serial.STOPBITS_ONE,
-            dsrdtr=True,
-            rtscts=True,
+        GsmCallNotificationService.modem = Modem(
+            await serial_asyncio.open_serial_connection(
+                url=self.device_path,
+                baudrate=75600,
+                bytesize=serial_asyncio.serial.EIGHTBITS,
+                parity=serial_asyncio.serial.PARITY_NONE,
+                stopbits=serial_asyncio.serial.STOPBITS_ONE,
+                dsrdtr=True,
+                rtscts=True,
+                limit=READ_LIMIT,
+            )
         )
 
     async def terminate(self):
         if GsmCallNotificationService.modem is None:
             return
 
         _LOGGER.debug("Closing connection to the modem...")
-        GsmCallNotificationService.modem[1].close()
-        await GsmCallNotificationService.modem[1].wait_closed()
+        GsmCallNotificationService.modem.writer.close()
+        await GsmCallNotificationService.modem.writer.wait_closed()
         GsmCallNotificationService.modem = None