Migrate serial backend from pyserial to serialx

pymonoprice/__init__.py

@@ -3,10 +3,9 @@
 import asyncio
 import logging
 import re
-import serial
+import serialx
 from dataclasses import dataclass
 from functools import wraps
-from serial_asyncio_fast import create_serial_connection, SerialTransport
 from threading import RLock
 from typing import TYPE_CHECKING
 
@@ -123,13 +122,15 @@ def __init__(self, port_url: str, lock: RLock) -> None:
         Monoprice amplifier interface
         """
         self._lock = lock
-        self._port = serial.serial_for_url(port_url, do_not_open=True)
-        self._port.baudrate = 9600
-        self._port.stopbits = serial.STOPBITS_ONE
-        self._port.bytesize = serial.EIGHTBITS
-        self._port.parity = serial.PARITY_NONE
-        self._port.timeout = TIMEOUT
-        self._port.write_timeout = TIMEOUT
+        self._port = serialx.serial_for_url(
+            port_url,
+            baudrate=9600,
+            stopbits=serialx.StopBits.ONE,
+            byte_size=8,
+            parity=serialx.Parity.NONE,
+            read_timeout=TIMEOUT,
+            write_timeout=TIMEOUT,
+        )
         self._port.open()
 
     def _send_request(self, request: bytes) -> None:
@@ -157,7 +158,7 @@ def _process_request(self, request: bytes, num_eols_to_read: int = 1) -> str:
         while True:
             c = self._port.read(1)
             if not c:
-                raise serial.SerialTimeoutException(
+                raise serialx.SerialTimeoutException(
                     "Connection timed out! Last received bytes {}".format(
                         [hex(a) for a in result]
                     )
@@ -394,12 +395,13 @@ def __init__(self) -> None:
         super().__init__()
         self._lock = asyncio.Lock()
         self._tasks: set[asyncio.Task[None]] = set()
-        self._transport: SerialTransport | None = None
+        self._transport: serialx.SerialTransport | None = None
         self._connected = asyncio.Event()
         self.q: asyncio.Queue[bytes] = asyncio.Queue()
 
     def connection_made(self, transport: asyncio.BaseTransport) -> None:
-        self._transport = transport  # type: ignore[assignment]
+        assert isinstance(transport, serialx.SerialTransport)
+        self._transport = transport
         self._connected.set()
         _LOGGER.debug("port opened %s", self._transport)
 
@@ -520,7 +522,7 @@ async def get_async_monoprice(port_url: str) -> MonopriceAsync:
     lock = asyncio.Lock()
 
     loop = asyncio.get_running_loop()
-    _, protocol = await create_serial_connection(
+    _, protocol = await serialx.create_serial_connection(
         loop, MonopriceProtocol, port_url, baudrate=9600
     )
     return MonopriceAsync(protocol, lock)  # type: ignore[arg-type]

pyproject.toml

@@ -11,8 +11,7 @@ license = { text = "MIT" }
 authors = [{ name = "On Freund", email = "onfreund@gmail.com" }]
 requires-python = ">=3.12"
 dependencies = [
-    "pyserial>=3.4",
-    "pyserial-asyncio-fast>=0.16",
+    "serialx>=0.7.0",
 ]
 classifiers = [
     "Development Status :: 4 - Beta",
@@ -51,6 +50,3 @@ warn_unreachable = true
 warn_unused_configs = true
 warn_unused_ignores = true
 
-[[tool.mypy.overrides]]
-module = ["serial", "serial_asyncio_fast"]
-ignore_missing_imports = true

requirements.txt

@@ -1,2 +1 @@
-pyserial>=3.4
-pyserial-asyncio-fast>=0.16
+serialx>=1.4.1

tests/test_monoprice.py

@@ -1,8 +1,7 @@
 import unittest
 
-import serial
-
 import pymonoprice
+import serialx
 from pymonoprice import (get_monoprice, get_async_monoprice, ZoneStatus)
 from tests import create_dummy_port
 import asyncio
@@ -324,7 +323,7 @@ def test_restore_zone(self):
         self.assertEqual(0, len(self.responses))
 
     def test_timeout(self):
-        with self.assertRaises(serial.SerialTimeoutException):
+        with self.assertRaises(serialx.SerialTimeoutException):
             self.monoprice.set_source(3, 3)