Wrap PySerial inside busio.UART for Linux platforms

src/adafruit_blinka/microcontroller/generic_linux/uart.py

@@ -0,0 +1,167 @@
+# SPDX-FileCopyrightText: 2026 Melissa LeBlanc-Williams for Adafruit Industries
+#
+# SPDX-License-Identifier: MIT
+"""Generic Linux UART class wrapping PySerial"""
+
+import os
+
+import serial
+
+
+class UART:
+    """UART class for generic Linux using PySerial.
+
+    Wraps a ``serial.Serial`` instance so that CircuitPython UART code
+    runs unchanged on Linux / SBC boards.
+    """
+
+    # Map of known port IDs to device paths, by platform.
+    # If the port entry in uartPorts is a string it is used directly;
+    # otherwise we try common symlink / device conventions.
+    _PORT_SEARCH_PATTERNS = (
+        "/dev/serial{}",
+        "/dev/ttyS{}",
+        "/dev/ttyAMA{}",
+    )
+
+    # pylint: disable=too-many-arguments
+    def __init__(
+        self,
+        port_id,
+        baudrate=9600,
+        bits=8,
+        parity=None,
+        stop=1,
+        timeout=1,
+        receiver_buffer_size=64,  # pylint: disable=unused-argument
+    ):
+        device = self._resolve_device(port_id)
+
+        # Translate CircuitPython parity values to pyserial constants.
+        if parity is None:
+            ser_parity = serial.PARITY_NONE
+        elif parity == 0:
+            ser_parity = serial.PARITY_EVEN
+        elif parity == 1:
+            ser_parity = serial.PARITY_ODD
+        else:
+            raise ValueError("Invalid parity: {}".format(parity))
+
+        stop_map = {1: serial.STOPBITS_ONE, 2: serial.STOPBITS_TWO}
+        ser_stop = stop_map.get(stop)
+        if ser_stop is None:
+            raise ValueError("Invalid stop bits: {}".format(stop))
+
+        byte_size_map = {
+            5: serial.FIVEBITS,
+            6: serial.SIXBITS,
+            7: serial.SEVENBITS,
+            8: serial.EIGHTBITS,
+        }
+        ser_bytesize = byte_size_map.get(bits)
+        if ser_bytesize is None:
+            raise ValueError("Invalid bits: {}".format(bits))
+
+        # PySerial timeout is in seconds (float); CircuitPython also uses
+        # seconds as of CP 4.0+.  Blinka's older MicroPython path passed
+        # timeout in *milliseconds* to machine.UART; we accept seconds here.
+        self._serial = serial.Serial(
+            device,
+            baudrate=baudrate,
+            bytesize=ser_bytesize,
+            parity=ser_parity,
+            stopbits=ser_stop,
+            timeout=timeout,
+            write_timeout=timeout,
+        )
+
+    # ----- helpers -----
+
+    @classmethod
+    def _resolve_device(cls, port_id):
+        """Turn a port identifier into a ``/dev/`` path.
+
+        *port_id* may be:
+        - A string that is already a device path (e.g. ``"/dev/serial0"``).
+        - An integer that will be resolved via common naming conventions.
+        """
+        if isinstance(port_id, str) and os.path.exists(port_id):
+            return port_id
+
+        if isinstance(port_id, int):
+            for pattern in cls._PORT_SEARCH_PATTERNS:
+                path = pattern.format(port_id)
+                if os.path.exists(path):
+                    return path
+
+        raise RuntimeError(
+            "Could not find UART device for port {!r}.  "
+            "Make sure the serial port is enabled.".format(port_id)
+        )
+
+    # ----- CircuitPython-compatible API -----
+
+    def deinit(self):
+        """Close the serial port."""
+        if self._serial is not None:
+            self._serial.close()
+            self._serial = None
+
+    def read(self, nbytes=None):
+        """Read up to *nbytes* bytes.  Returns ``None`` when no data is
+        available (matching CircuitPython behaviour, not pyserial's ``b""``).
+        """
+        if nbytes is None:
+            # Read whatever is available; if nothing, wait up to timeout.
+            data = self._serial.read(self._serial.in_waiting or 1)
+        else:
+            data = self._serial.read(nbytes)
+        return data if data else None
+
+    def readinto(self, buf, nbytes=None):
+        """Read bytes into *buf*.  Returns number of bytes read or ``None``."""
+        if nbytes is None:
+            nbytes = len(buf)
+        data = self._serial.read(nbytes)
+        if not data:
+            return None
+        n = len(data)
+        buf[:n] = data
+        return n
+
+    def readline(self):
+        """Read a line (up to ``\\n``).  Returns ``None`` on timeout with no data."""
+        data = self._serial.readline()
+        return data if data else None
+
+    def write(self, buf):
+        """Write bytes from *buf*.  Returns the number of bytes written."""
+        return self._serial.write(buf)
+
+    @property
+    def baudrate(self):
+        """The current baudrate."""
+        return self._serial.baudrate
+
+    @baudrate.setter
+    def baudrate(self, value):
+        self._serial.baudrate = value
+
+    @property
+    def in_waiting(self):
+        """The number of bytes in the input buffer, available to be read."""
+        return self._serial.in_waiting
+
+    @property
+    def timeout(self):
+        """Read timeout in seconds (float)."""
+        return self._serial.timeout
+
+    @timeout.setter
+    def timeout(self, value):
+        self._serial.timeout = value
+        self._serial.write_timeout = value
+
+    def reset_input_buffer(self):
+        """Discard any unread data in the input buffer."""
+        self._serial.reset_input_buffer()

src/busio.py

@@ -522,7 +522,7 @@ def write_readinto(
 class UART(Lockable):
     """
     Busio UART Class for CircuitPython Compatibility. Used
-    for MicroPython and a few other non-Linux boards.
+    for MicroPython, Linux (via PySerial), and other boards.
     """
 
     class Parity(Enum):
@@ -541,15 +541,15 @@ def __init__(
         bits=8,
         parity=None,
         stop=1,
-        timeout=1000,
+        timeout=1,
         receiver_buffer_size=64,
         flow=None,
     ):
         if detector.board.any_embedded_linux:
-            raise RuntimeError(
-                "busio.UART not supported on this platform. Please use pyserial instead."
+            from adafruit_blinka.microcontroller.generic_linux.uart import (
+                UART as _UART,
             )
-        if detector.board.binho_nova:
+        elif detector.board.binho_nova:
             from adafruit_blinka.microcontroller.nova.uart import UART as _UART
         elif detector.board.greatfet_one:
             from adafruit_blinka.microcontroller.nxp_lpc4330.uart import UART as _UART
@@ -560,7 +560,8 @@ def __init__(
 
         from microcontroller.pin import uartPorts
 
-        self.baudrate = baudrate
+        self._baudrate = baudrate
+        self._timeout = timeout
 
         if flow is not None:  # default 0
             raise NotImplementedError(
@@ -577,7 +578,27 @@ def __init__(
         else:
             raise ValueError("Invalid parity")
 
-        if detector.chip.id in (ap_chip.RP2040, ap_chip.RP2350):
+        if detector.board.any_embedded_linux:
+            # check tx and rx have hardware support
+            for portId, portTx, portRx in uartPorts:
+                if portTx == tx and portRx == rx:
+                    self._uart = _UART(
+                        portId,
+                        baudrate=baudrate,
+                        bits=bits,
+                        parity=parity,
+                        stop=stop,
+                        timeout=timeout,
+                        receiver_buffer_size=receiver_buffer_size,
+                    )
+                    break
+            else:
+                raise ValueError(
+                    "No Hardware UART on (tx,rx)={}\nValid UART ports: {}".format(
+                        (tx, rx), uartPorts
+                    )
+                )
+        elif detector.chip.id in (ap_chip.RP2040, ap_chip.RP2350):
             self._uart = _UART(
                 tx,
                 rx,
@@ -609,9 +630,18 @@ def __init__(
 
     def deinit(self):
         """Deinitialization"""
-        if detector.board.binho_nova:
-            self._uart.deinit()
-        self._uart = None
+        if self._uart is not None:
+            if hasattr(self._uart, "deinit"):
+                self._uart.deinit()
+            self._uart = None
+
+    def __enter__(self):
+        """No-op used by Context Managers."""
+        return self
+
+    def __exit__(self, *args):
+        """Automatically deinitializes the hardware when exiting a context."""
+        self.deinit()
 
     def read(self, nbytes=None):
         """Read from the UART"""
@@ -629,6 +659,17 @@ def write(self, buf):
         """Write to the UART from a buffer"""
         return self._uart.write(buf)
 
+    @property
+    def baudrate(self):
+        """The current baudrate."""
+        return self._baudrate
+
+    @baudrate.setter
+    def baudrate(self, value):
+        self._baudrate = value
+        if hasattr(self, "_uart") and hasattr(self._uart, "baudrate"):
+            self._uart.baudrate = value
+
     @property
     def in_waiting(self):
         """The number of bytes in the input buffer, available to be read"""
@@ -639,3 +680,25 @@ def in_waiting(self):
         if hasattr(self._uart, "any"):
             return self._uart.any()
         raise NotImplementedError("in_waiting not supported on this platform")
+
+    @property
+    def timeout(self):
+        """The timeout in seconds (float)."""
+        if hasattr(self._uart, "timeout"):
+            return self._uart.timeout
+        return self._timeout
+
+    @timeout.setter
+    def timeout(self, value):
+        self._timeout = value
+        if hasattr(self._uart, "timeout"):
+            self._uart.timeout = value
+
+    def reset_input_buffer(self):
+        """Discard any unread data in the input buffer."""
+        if hasattr(self._uart, "reset_input_buffer"):
+            self._uart.reset_input_buffer()
+        else:
+            # Fallback: read and discard all available bytes
+            while self.in_waiting:
+                self._uart.read(self.in_waiting)