Use zigpy for making serial connections

Author: puddly

zigpy_znp/uart.py

@@ -1,26 +1,17 @@
+from __future__ import annotations
+
 import typing
 import asyncio
 import logging
-import warnings
 
-import serial
+import zigpy.serial
 
 import zigpy_znp.config as conf
 import zigpy_znp.frames as frames
 import zigpy_znp.logger as log
 from zigpy_znp.types import Bytes
 from zigpy_znp.exceptions import InvalidFrame
 
-with warnings.catch_warnings():
-    warnings.filterwarnings(
-        action="ignore",
-        module="serial_asyncio",
-        message='"@coroutine" decorator is deprecated',
-        category=DeprecationWarning,
-    )
-    import serial_asyncio  # noqa: E402
-
-
 LOGGER = logging.getLogger(__name__)
 
 
@@ -29,12 +20,14 @@ class BufferTooShort(Exception):
 
 
 class ZnpMtProtocol(asyncio.Protocol):
-    def __init__(self, api):
+    def __init__(self, api, *, url: str | None = None) -> None:
         self._buffer = bytearray()
         self._api = api
         self._transport = None
         self._connected_event = asyncio.Event()
 
+        self.url = url
+
     def close(self) -> None:
         """Closes the port."""
 
@@ -47,7 +40,7 @@ def close(self) -> None:
             self._transport.close()
             self._transport = None
 
-    def connection_lost(self, exc: typing.Optional[Exception]) -> None:
+    def connection_lost(self, exc: Exception | None) -> None:
         """Connection lost."""
 
         if exc is not None:
@@ -56,7 +49,7 @@ def connection_lost(self, exc: typing.Optional[Exception]) -> None:
         if self._api is not None:
             self._api.connection_lost(exc)
 
-    def connection_made(self, transport: serial_asyncio.SerialTransport) -> None:
+    def connection_made(self, transport: asyncio.BaseTransport) -> None:
         """Opened serial port."""
         self._transport = transport
         LOGGER.debug("Opened %s serial port", transport.serial.name)
@@ -179,13 +172,11 @@ async def connect(config: conf.ConfigType, api) -> ZnpMtProtocol:
 
     LOGGER.debug("Connecting to %s at %s baud", port, baudrate)
 
-    _, protocol = await serial_asyncio.create_serial_connection(
+    _, protocol = await zigpy.serial.create_serial_connection(
         loop=loop,
         protocol_factory=lambda: ZnpMtProtocol(api),
         url=port,
         baudrate=baudrate,
-        parity=serial.PARITY_NONE,
-        stopbits=serial.STOPBITS_ONE,
         xonxoff=(flow_control == "software"),
         rtscts=(flow_control == "hardware"),
     )