Split up on_close and on_open kwargs and fix Windows

docs/how-to/pyserial-migration.md

@@ -68,6 +68,8 @@ are listed in the [main API documentation](../api.md). The most common ones are
 | `Serial(writeTimeout=...)`  | `write_timeout=...`        | Constructor kwarg                            |
 | `Serial(bytesize=...)`      | `byte_size=...`            | Constructor kwarg                            |
 | `Serial(do_not_open=False)` | —                          | Not supported; open explicitly via `open()`  |
+| `Serial(rtsdtr_on_open=X)`  | `dtr_on_open=X, rts_on_open=X`   | Now controlled per-pin                  |
+| `Serial(rtsdtr_on_close=X)` | `dtr_on_close=X, rts_on_close=X` | Now controlled per-pin                  |
 | `SerialPortInfo[i]`         | attribute access           | Slicing `SerialPortInfo` is deprecated       |
 | `SerialPortInfo.description`| `SerialPortInfo.product`   |                                              |
 

serialx/common.py

@@ -247,9 +247,15 @@ class _CommonConnectKwargs(TypedDict, total=False):
     byte_size: int
     read_timeout: float | None
     write_timeout: float | None
+    dtr_on_open: PinState
+    rts_on_open: PinState
+    dtr_on_close: PinState
+    rts_on_close: PinState
+    exclusive: bool
+
+    # backwards compatibility kwargs
     rtsdtr_on_open: PinState
     rtsdtr_on_close: PinState
-    exclusive: bool
 
     # pyserial compatibility kwargs
     port: str | None
@@ -410,7 +416,13 @@ def replacement(self: BaseSerial, /, *args: _P.args, **kwargs: _P.kwargs) -> _R:
 
 
 class BaseSerial(io.RawIOBase):
-    """Base class for serial port communication."""
+    """Base class for serial port communication.
+
+    .. deprecated:: 1.9.0
+        The ``rtsdtr_on_open`` and ``rtsdtr_on_close`` constructor kwargs are
+        deprecated; use the per-pin ``dtr_on_open``, ``rts_on_open``,
+        ``dtr_on_close``, and ``rts_on_close`` instead.
+    """
 
     def __init__(
         self,
@@ -425,8 +437,10 @@ def __init__(
         byte_size: int = 8,
         read_timeout: float | None = None,
         write_timeout: float | None = None,
-        rtsdtr_on_open: PinState = PinState.HIGH,
-        rtsdtr_on_close: PinState = PinState.LOW,
+        dtr_on_open: PinState = PinState.HIGH,
+        rts_on_open: PinState = PinState.HIGH,
+        dtr_on_close: PinState = PinState.LOW,
+        rts_on_close: PinState = PinState.LOW,
         exclusive: bool = True,
         # pyserial compatibility kwargs
         port: str | None = None,
@@ -435,6 +449,9 @@ def __init__(
         do_not_open: bool | None = None,
         writeTimeout: float | None = None,
         inter_byte_timeout: int | None = None,
+        # Legacy kwargs
+        rtsdtr_on_open: PinState = PinState.UNDEFINED,
+        rtsdtr_on_close: PinState = PinState.UNDEFINED,
         # Internal pyserial compatibility signal
         _wrap_exceptions: bool = False,
     ) -> None:
@@ -464,8 +481,34 @@ def __init__(
         self._read_timeout = read_timeout
         self._write_timeout = write_timeout
 
-        self._rtsdtr_on_open = rtsdtr_on_open
-        self._rtsdtr_on_close = rtsdtr_on_close
+        if (
+            rtsdtr_on_open is not PinState.UNDEFINED
+            or rtsdtr_on_close is not PinState.UNDEFINED
+        ):
+            warnings.warn(
+                "`rtsdtr_on_open`/`rtsdtr_on_close` are deprecated; use the per-pin"
+                " `dtr_on_open`/`rts_on_open`/`dtr_on_close`/`rts_on_close` instead",
+                DeprecationWarning,
+                stacklevel=2,
+            )
+
+        self._rts_on_open: PinState = (
+            rts_on_open if rtsdtr_on_open is PinState.UNDEFINED else rtsdtr_on_open
+        )
+        self._rts_on_close: PinState = (
+            rts_on_close if rtsdtr_on_close is PinState.UNDEFINED else rtsdtr_on_close
+        )
+        self._dtr_on_open: PinState = (
+            dtr_on_open if rtsdtr_on_open is PinState.UNDEFINED else rtsdtr_on_open
+        )
+        self._dtr_on_close: PinState = (
+            dtr_on_close if rtsdtr_on_close is PinState.UNDEFINED else rtsdtr_on_close
+        )
+
+        # Last-written DTR/RTS output state, for readback on backends that can't
+        # report output lines from hardware
+        self._dtr_state = PinState.UNDEFINED
+        self._rts_state = PinState.UNDEFINED
 
         self._auto_close = False
 
@@ -562,7 +605,26 @@ def write_timeout(self) -> float | None:
     def get_modem_pins(self) -> ModemPins:
         """Get modem control bits."""
         self._check_broken()
-        return self._get_modem_pins()
+        pins = self._get_modem_pins()
+        return dataclasses.replace(
+            pins,
+            dtr=pins.dtr if pins.dtr is not PinState.UNDEFINED else self._dtr_state,
+            rts=pins.rts if pins.rts is not PinState.UNDEFINED else self._rts_state,
+        )
+
+    def _modem_pins_on_open(self) -> ModemPins:
+        """DTR/RTS to apply on open."""
+        return ModemPins(
+            dtr=self._dtr_on_open if not self._dsrdtr else PinState.UNDEFINED,
+            rts=self._rts_on_open if not self._rtscts else PinState.UNDEFINED,
+        )
+
+    def _modem_pins_on_close(self) -> ModemPins:
+        """DTR/RTS to apply on close."""
+        return ModemPins(
+            dtr=self._dtr_on_close if not self._dsrdtr else PinState.UNDEFINED,
+            rts=self._rts_on_close if not self._rtscts else PinState.UNDEFINED,
+        )
 
     @maybe_wrap_exceptions
     def set_modem_pins(
@@ -596,7 +658,13 @@ def set_modem_pins(
                 dsr=PinState.convert(dsr),
             )
 
-        return self._set_modem_pins(pins)
+        self._set_modem_pins(pins)
+
+        if pins.dtr is not PinState.UNDEFINED:
+            self._dtr_state = pins.dtr
+
+        if pins.rts is not PinState.UNDEFINED:
+            self._rts_state = pins.rts
 
     @abstractmethod
     def _get_modem_pins(self) -> ModemPins:
@@ -674,14 +742,24 @@ def stopbits(self) -> StopBits:
         return self._stopbits
 
     @property
-    def rtsdtr_on_open(self) -> PinState:
-        """Get the RTS/DTR pin state (on open) setting."""
-        return self._rtsdtr_on_open
+    def dtr_on_open(self) -> PinState:
+        """Get the DTR pin state (on open) setting."""
+        return self._dtr_on_open
 
     @property
-    def rtsdtr_on_close(self) -> PinState:
-        """Get the RTS/DTR pin state (on close) setting."""
-        return self._rtsdtr_on_close
+    def rts_on_open(self) -> PinState:
+        """Get the RTS pin state (on open) setting."""
+        return self._rts_on_open
+
+    @property
+    def dtr_on_close(self) -> PinState:
+        """Get the DTR pin state (on close) setting."""
+        return self._dtr_on_close
+
+    @property
+    def rts_on_close(self) -> PinState:
+        """Get the RTS pin state (on close) setting."""
+        return self._rts_on_close
 
     @property
     def exclusive(self) -> bool:
@@ -963,21 +1041,31 @@ def isOpen(self) -> bool:
     @property
     def dtr(self) -> bool | None:
         """Get DTR modem bit."""
+        if not self.is_open:
+            return self._dtr_on_open.to_bool()
         return self.get_modem_pins().dtr.to_bool()
 
     @dtr.setter
     def dtr(self, value: bool) -> None:
         """Set DTR modem bit."""
+        if not self.is_open:
+            self._dtr_on_open = PinState.convert(value)
+            return
         self.set_modem_pins(dtr=bool(value))
 
     @property
     def rts(self) -> bool | None:
         """Get RTS modem bit."""
+        if not self.is_open:
+            return self._rts_on_open.to_bool()
         return self.get_modem_pins().rts.to_bool()
 
     @rts.setter
     def rts(self, value: bool) -> None:
         """Set RTS modem bit."""
+        if not self.is_open:
+            self._rts_on_open = PinState.convert(value)
+            return
         self.set_modem_pins(rts=bool(value))
 
     @property

serialx/platforms/serial_posix.py

@@ -220,15 +220,21 @@ def _build_tcsetattr_flags(self) -> TcsetattrFlags:
         # Ignore modem control lines
         cflag |= termios.CLOCAL
 
-        # Lower modem control lines after last process closes the device (hang up)
-        if self._rtsdtr_on_close is PinState.UNDEFINED:
+        # Lower modem control lines after last process closes the device (hang up).
+        # HUPCL is all-or-nothing: it lowers both DTR and RTS together, so POSIX can
+        # only honor uniform close states.
+        if (
+            self._dtr_on_close is PinState.UNDEFINED
+            and self._rts_on_close is PinState.UNDEFINED
+        ):
             pass
-        elif self._rtsdtr_on_close is PinState.HIGH:
+        elif self._dtr_on_close is PinState.LOW and self._rts_on_close is PinState.LOW:
+            cflag |= termios.HUPCL
+        else:
             raise UnsupportedSetting(
-                "POSIX only supports setting RTS/DTR to LOW on close"
+                "POSIX only supports lowering both DTR and RTS together on close"
+                " (dtr_on_close=rts_on_close=LOW) or leaving both untouched"
             )
-        else:
-            cflag |= termios.HUPCL
 
         cflag |= self._build_character_size_flags()
         cflag |= self._build_parity_flags()
@@ -312,7 +318,7 @@ def _configure_port(self) -> None:
 
         self._after_configure_port()
 
-        self.set_modem_pins(dtr=self._rtsdtr_on_open, rts=self._rtsdtr_on_open)
+        self.set_modem_pins(self._modem_pins_on_open())
 
         # Flush input and output buffers to discard stale data
         termios.tcflush(self._fileno, termios.TCIOFLUSH)
@@ -547,7 +553,7 @@ async def _get_modem_pins(self) -> ModemPins:
     async def _set_modem_pins(self, modem_pins: ModemPins) -> None:
         """Set modem control bits, internal."""
         assert self._serial is not None
-        await self._loop.run_in_executor(None, self._serial._set_modem_pins, modem_pins)
+        await self._loop.run_in_executor(None, self._serial.set_modem_pins, modem_pins)
 
 
 register_uri_handler(

serialx/platforms/serial_pyodide/__init__.py

@@ -164,6 +164,11 @@ def __init__(
         self._reader_task: asyncio.Task[None] | None = None
         self._writer_task: asyncio.Task[None] | None = None
 
+        # Last-written DTR/RTS output state; Web Serial `getSignals` only reports
+        # input lines, so output readback comes from this cache.
+        self._dtr_state = PinState.UNDEFINED
+        self._rts_state = PinState.UNDEFINED
+
     async def _connect(  # type: ignore[override]
         self,
         *,
@@ -233,11 +238,10 @@ async def _connect(  # type: ignore[override]
         self._js_port = port
         assert self._js_port is not None
 
-        if self._serial.rtsdtr_on_open is not PinState.UNDEFINED:
-            await self.set_modem_pins(
-                rts=self._serial.rtsdtr_on_open,
-                dtr=self._serial.rtsdtr_on_open,
-            )
+        await self.set_modem_pins(
+            rts=self._serial.rts_on_open,
+            dtr=self._serial.dtr_on_open,
+        )
 
         readable = self._js_port.readable
         assert readable is not None
@@ -286,7 +290,10 @@ async def _get_modem_pins(self) -> ModemPins:
         assert self._js_port is not None
         result = await self._js_port.getSignals()
 
+        # `getSignals` only reports input lines; DTR/RTS come from the cache
         return ModemPins(
+            dtr=self._dtr_state,
+            rts=self._rts_state,
             cts=PinState.convert(result.clearToSend),
             car=PinState.convert(result.dataCarrierDetect),
             rng=PinState.convert(result.ringIndicator),
@@ -305,6 +312,11 @@ async def _set_modem_pins(self, modem_pins: ModemPins) -> None:
             assert self._js_port is not None
             await self._js_port.setSignals(**signals)
 
+        if modem_pins.dtr is not PinState.UNDEFINED:
+            self._dtr_state = modem_pins.dtr
+        if modem_pins.rts is not PinState.UNDEFINED:
+            self._rts_state = modem_pins.rts
+
     def write(self, data: bytes | bytearray | memoryview) -> None:
         """Write data to the transport."""
         if self._closing:
@@ -356,12 +368,11 @@ async def _close_port(self, exception: Exception | None) -> None:
             self._js_writer = None
 
         if self._js_port is not None:
-            if self._serial.rtsdtr_on_close is not PinState.UNDEFINED:
-                with contextlib.suppress(Exception):
-                    await self.set_modem_pins(
-                        rts=self._serial.rtsdtr_on_close,
-                        dtr=self._serial.rtsdtr_on_close,
-                    )
+            with contextlib.suppress(Exception):
+                await self.set_modem_pins(
+                    rts=self._serial.rts_on_close,
+                    dtr=self._serial.dtr_on_close,
+                )
             await self._js_port.close()
             self._js_port = None
 

serialx/platforms/serial_win32.py

@@ -11,6 +11,7 @@
 import pywintypes
 from typing_extensions import Buffer, Unpack
 from win32con import (
+    DTR_CONTROL_DISABLE,
     DTR_CONTROL_ENABLE,
     DTR_CONTROL_HANDSHAKE,
     EVENPARITY,
@@ -26,6 +27,7 @@
     ONE5STOPBITS,
     ONESTOPBIT,
     OPEN_EXISTING,
+    RTS_CONTROL_DISABLE,
     RTS_CONTROL_ENABLE,
     RTS_CONTROL_HANDSHAKE,
     SPACEPARITY,
@@ -252,6 +254,9 @@ def _configure_port(self) -> None:
             if self._rtscts:
                 dcb.fRtsControl = RTS_CONTROL_HANDSHAKE
                 dcb.fOutxCtsFlow = 1
+            elif self._rts_on_open is PinState.LOW:
+                dcb.fRtsControl = RTS_CONTROL_DISABLE
+                dcb.fOutxCtsFlow = 0
             else:
                 dcb.fRtsControl = RTS_CONTROL_ENABLE
                 dcb.fOutxCtsFlow = 0
@@ -266,6 +271,9 @@ def _configure_port(self) -> None:
             if self._dsrdtr:
                 dcb.fDtrControl = DTR_CONTROL_HANDSHAKE
                 dcb.fOutxDsrFlow = 1
+            elif self._dtr_on_open is PinState.LOW:
+                dcb.fDtrControl = DTR_CONTROL_DISABLE
+                dcb.fOutxDsrFlow = 0
             else:
                 dcb.fDtrControl = DTR_CONTROL_ENABLE
                 dcb.fOutxDsrFlow = 0
@@ -277,12 +285,8 @@ def _configure_port(self) -> None:
 
             SetCommState(self._handle, dcb)
 
-            # RTS cannot be manually set when hardware flow control is enabled
-            if not self._rtscts:
-                self.set_modem_pins(
-                    dtr=self._rtsdtr_on_open,
-                    rts=self._rtsdtr_on_open,
-                )
+            # Driver-owned lines (RTS under rtscts, DTR under dsrdtr) are skipped
+            self.set_modem_pins(self._modem_pins_on_open())
 
             # Clear any errors
             ClearCommError(self._handle)
@@ -298,16 +302,13 @@ def fileno(self) -> int:
     def _close(self) -> None:
         """Close the serial port and release all handles."""
         if self._handle is not None:
-            # Windows has no way to automatically do this on close, we do it manually
-            if not self._rtscts:
-                # RTS cannot be manually set when hardware flow control is enabled
-                try:
-                    self.set_modem_pins(
-                        dtr=self._rtsdtr_on_close,
-                        rts=self._rtsdtr_on_close,
-                    )
-                except OSError:
-                    LOGGER.debug("Failed to set modem pins on close", exc_info=True)
+            # Windows has no way to automatically do this on close, we do it manually. A
+            # driver-owned line (RTS under rtscts, DTR under dsrdtr) cannot be adjusted
+            # via EscapeCommFunction, so those are skipped.
+            try:
+                self.set_modem_pins(self._modem_pins_on_close())
+            except OSError:
+                LOGGER.debug("Failed to set modem pins on close", exc_info=True)
 
             try:
                 CancelIo(self._handle)