ok_serial

 69class SerialConnection(contextlib.AbstractContextManager):
 70    """An open connection to a serial port.
 71
 72    Thread-safe: any method may be called from any thread any time, and any
 73    `*_async` method may be awaited from any event loop on any thread any time.
 74    """
 75
 76    def __init__(
 77        self,
 78        *,
 79        match: str | PortPredicate | None = None,
 80        port: str | PortInfo | None = None,
 81        opts: SerialConnectionOptions = SerialConnectionOptions(),
 82        **kwargs,
 83    ):
 84        """Opens a serial port to make it available for use.
 85
 86        - `match` is a
 87          [match string](https://github.com/egnor/ok-py-serial#port-matching)
 88          or `PortInfo -> bool` callable matching exactly one port...
 89          - OR `port` must name a raw system serial device to open.
 90        - `opts` can define baud rate and other port parameters...
 91          - OR other keywords are forwarded to `SerialConnectionOptions`
 92
 93        Call `close` to release the port, or use `SerialConnection` as the
 94        target of a `with` statement.
 95
 96        Example:
 97        ```
 98        with SerialConnection(match="xyz", baud=115200, sharing="polite") as p:
 99            ... interact with `p` ...
100            # automatically closed on exit from block
101        ```
102
103        Raises:
104        - `SerialOpenException` - I/O error opening the specified port
105        - `SerialOpenBusy` - The port is already in use
106        - `SerialScanException` - System error scanning ports to find `match`
107        """
108
109        assert (match is not None) + (port is not None) == 1
110        self._opts = dataclasses.replace(opts, **kwargs)
111
112        if match is not None:
113            if not (found := scan_serial_ports(match)):
114                msg = f"No ports match {match!r}"
115                raise _exceptions.SerialOpenException(msg)
116            if len(found) > 1:
117                detail = "".join(f"\n  {p}" for p in found)
118                msg = f"Multiple ports match {match!r}: {detail}"
119                raise _exceptions.SerialOpenException(msg)
120            port = found[0].name
121            log.debug("Scanned %r, found %s", match, port)
122
123        assert port is not None
124        if isinstance(port, PortInfo):
125            port = port.name
126
127        with contextlib.ExitStack() as cleanup:
128            port_lock = cleanup.enter_context(
129                PortLock(port, self._opts.sharing)
130            )
131
132            try:
133                # (If "polite", wake the readloop periodically for checks.)
134                timeout = 0.5 if self._opts.sharing == "polite" else None
135                pyserial = cleanup.enter_context(
136                    serial.Serial(
137                        port=port,
138                        baudrate=self._opts.baud,
139                        write_timeout=0.1,
140                        timeout=timeout,
141                    )
142                )
143                log.debug("Opened %s %s", port, self._opts)
144            except OSError as ex:
145                if ex.errno == errno.EBUSY:
146                    msg = "Port busy (EBUSY)"
147                    raise _exceptions.SerialOpenBusy(msg, port) from ex
148                else:
149                    msg = "Port open error"
150                    raise _exceptions.SerialOpenException(msg, port) from ex
151
152            if hasattr(pyserial, "fileno"):
153                # unlock fd before closing port (see note on release_fd)
154                cleanup.callback(port_lock.release_fd)
155                port_lock.attach_fd(pyserial.fileno())
156
157            # (annotated because AbstractContextManager.__enter__ gives Any,
158            # which would silently disable checking of every self._io use)
159            self._io: _IoThreads = cleanup.enter_context(
160                _IoThreads(pyserial, port_lock)
161            )
162            self._io.start()
163            self._cleanup_lock = threading.Lock()  # assigned before _cleanup
164            self._cleanup = cleanup.pop_all()
165
166    def __del__(self) -> None:
167        if cleanup := getattr(self, "_cleanup", None):
168            with self._cleanup_lock:
169                cleanup.close()
170
171    def __exit__(self, exc_type, exc_value, traceback) -> None:
172        with self._cleanup_lock:
173            self._cleanup.__exit__(exc_type, exc_value, traceback)
174
175    def __repr__(self) -> str:
176        return f"SerialConnection({self._io.device!r})"
177
178    def close(self) -> None:
179        """Releases the serial port connection and any associated locks.
180
181        Blocks until connection I/O threads have finished.
182        Thread-safe, OK to call repeatedly, and OK to call with I/O in flight.
183        Any I/O operations in progress or attempted afterwards raise an
184        immediate `SerialIoClosed` exception.
185        """
186
187        with self._cleanup_lock:
188            self._cleanup.close()
189
190    def read_sync(
191        self,
192        *,
193        timeout: float | int | None = None,
194    ) -> TimestampBytes:
195        """Waits up to `timeout` seconds (forever for `None`) for any data,
196        then returns all of it (b"" on timeout).
197
198        Thread-safe, but each call takes all the currently buffered data.
199
200        Raises:
201        - `SerialIoException` - port I/O failed and there is no matching data
202        - `SerialIoClosed` - the port was closed and there is no matching data
203        """
204
205        deadline = to_deadline(timeout)
206        while True:
207            with self._io.monitor:
208                if self._io.incoming:
209                    monotime = self._io.incoming_monotime
210                    out = TimestampBytes(self._io.incoming, monotime)
211                    self._io.incoming.clear()
212                    self._io.incoming_monotime = 0.0
213                    return out
214
215                self._io.check_poison_locked()
216                if (wait := from_deadline(deadline)) <= 0:
217                    return TimestampBytes(b"", 0.0)
218                self._io.monitor.wait(timeout=wait)
219
220    async def read_async(self) -> TimestampBytes:
221        """Similar to `read_sync` but returns a coroutine instead of blocking.
222
223        OK to call from any event loop on any thread, but as with `read_sync`
224        each call takes all the currently buffered data.
225
226        Raises `RuntimeError` if there is no running event loop.
227        """
228
229        while True:
230            future = self._io.create_future_in_loop()  # BEFORE read_sync
231            if out := self.read_sync(timeout=0):
232                return out
233            await future
234
235    def write(self, data: bytes | bytearray) -> None:
236        """Adds data to the outgoing buffer to be sent immediately.
237
238        Never blocks; the buffer can grow indefinitely. (Use `outgoing_size`
239        and `drain_sync`/`drain_async` to manage buffer size.)
240
241        Raises:
242        - `SerialIoException` - port I/O failed
243        - `SerialIoClosed` - the port was closed
244        """
245
246        with self._io.monitor:
247            self._io.check_poison_locked()
248            if data:
249                self._io.outgoing.extend(data)
250                self._io.monitor.notify_all()
251
252    def drain_sync(self, *, timeout: float | int | None = None) -> bool:
253        """Waits up to `timeout` seconds (forever for `None`) until
254        all buffered data is transmitted.
255
256        Returns `True` if the drain completed, `False` on timeout.
257
258        Raises:
259        - `SerialIoException` - port I/O failed
260        - `SerialIoClosed` - the port was closed
261        """
262
263        deadline = to_deadline(timeout)
264        while True:
265            with self._io.monitor:
266                self._io.check_poison_locked()
267                if not self._io.outgoing:
268                    return True
269                if (wait := from_deadline(deadline)) <= 0:
270                    return False
271                self._io.monitor.wait(timeout=wait)
272
273    async def drain_async(self) -> bool:
274        """Similar to `drain_sync` but returns a coroutine instead of blocking.
275
276        Raises `RuntimeError` if there is no running event loop.
277        """
278
279        while True:
280            future = self._io.create_future_in_loop()  # BEFORE drain_sync
281            if self.drain_sync(timeout=0):
282                return True
283            await future
284
285    def incoming_size(self) -> int:
286        """Returns the number of bytes waiting to be read."""
287        with self._io.monitor:
288            return len(self._io.incoming)
289
290    def outgoing_size(self) -> int:
291        """Returns the number of bytes waiting to be sent."""
292        with self._io.monitor:
293            return len(self._io.outgoing)
294
295    def set_signals(
296        self,
297        dtr: bool | None = None,
298        rts: bool | None = None,
299        send_break: bool | None = None,
300    ) -> None:
301        """Sets outgoing
302        [RS-232 modem control line](https://en.wikipedia.org/wiki/RS-232#Data_and_control_signals)
303        state (use `None` for no change).
304
305        - `dtr` - assert Data Terminal Ready
306        - `rts` - assert Ready To Send
307        - `send_break` - send a continuous BREAK condition
308
309        Raises:
310        - `SerialIoException` - port I/O failed
311        - `SerialIoClosed` - the port was closed
312        """
313
314        with self._io.monitor:
315            self._io.check_poison_locked()
316            try:
317                if dtr is not None:
318                    self._io.pyserial.dtr = dtr
319                if rts is not None:
320                    self._io.pyserial.rts = rts
321                if send_break is not None:
322                    self._io.pyserial.break_condition = send_break
323            except OSError as ex:
324                msg, dev = "Can't set control signals", self._io.device
325                if ex.errno == errno.ENOTTY:  # could be pty; don't poison
326                    raise _exceptions.SerialIoUnsupported(msg, dev) from ex
327                self._io.poison_locked(_exceptions.SerialIoException, msg, ex)
328                self._io.check_poison_locked()  # report to the caller
329
330    def get_signals(self) -> SerialControlSignals:
331        """Returns the current
332        [RS-232 modem control line](https://en.wikipedia.org/wiki/RS-232#Data_and_control_signals) state.
333
334        Raises:
335        - `SerialIoException` - port I/O failed
336        - `SerialIoClosed` - the port was closed
337        """
338
339        with self._io.monitor:
340            self._io.check_poison_locked()
341            try:
342                return SerialControlSignals(
343                    dtr=self._io.pyserial.dtr,
344                    dsr=self._io.pyserial.dsr,
345                    cts=self._io.pyserial.cts,
346                    rts=self._io.pyserial.rts,
347                    ri=self._io.pyserial.ri,
348                    cd=self._io.pyserial.cd,
349                    sending_break=self._io.pyserial.break_condition,
350                )
351            except OSError as ex:
352                msg, dev = ("Can't get control signals", self._io.device)
353                if ex.errno == errno.ENOTTY:  # could be pty; don't poison
354                    raise _exceptions.SerialIoUnsupported(msg, dev) from ex
355                self._io.poison_locked(_exceptions.SerialIoException, msg, ex)
356                self._io.check_poison_locked()  # report to the caller
357                assert False, "check_poison_locked() should have raised"
358
359    @property
360    def port_name(self) -> str:
361        """The port's device name, eg. `/dev/ttyACM0` or `COM3`."""
362        return self._io.device
363
364    @property
365    def pyserial(self) -> serial.Serial:
366        """The underlying
367        [`pyserial.Serial`](https://pyserial.readthedocs.io/en/latest/pyserial_api.html#serial.Serial)
368        object (API escape hatch).
369
370        NOT SYNCHRONIZED. Use at your own risk.
371        """
372        return self._io.pyserial
373
374    def fileno(self) -> int:
375        """The [Unix FD](https://en.wikipedia.org/wiki/File_descriptor)
376        for the serial connection, -1 if not available.
377        """
378        pyserial = self._io.pyserial
379        try:
380            return pyserial.fileno()
381        except AttributeError:  # no fileno() at all (eg. on Windows)
382            return -1
383        except OSError:  # port closed, or otherwise has no descriptor
384            return -1

21@dataclasses.dataclass(frozen=True)
22class SerialConnectionOptions:
23    """Optional parameters for `SerialConnection`."""
24
25    baud: int = 115200
26    """The [baud rate](https://en.wikipedia.org/wiki/Baud) to use."""
27
28    sharing: SerialSharingType = "exclusive"
29    """Port access negotiation strategy.
30
31    - `"oblivious"` - Don't perform any locking.
32    - `"polite"` - Defer to any other use of the port; don't lock the port.
33    - `"exclusive"` - Require exclusive access; lock the port or fail.
34    - `"stomp"` - Try to kill other processes using the port, try to lock the
35      port, open the port regardless. Use with care!
36    """

39@dataclasses.dataclass(frozen=True)
40class SerialControlSignals:
41    """[RS-232 modem control lines](https://en.wikipedia.org/wiki/RS-232#Data_and_control_signals).
42
43    Includes outgoing ("DTE to DCE") and incoming ("DCE to DTE") signals.
44    """
45
46    dtr: bool
47    dsr: bool
48    cts: bool
49    rts: bool
50    ri: bool
51    cd: bool
52    sending_break: bool

 26def scan_serial_ports(
 27    match: str | PortPredicate | None = None,
 28) -> list[PortInfo]:
 29    """
 30    Returns a list of serial ports currently attached to the system.
 31
 32    If set, `match` is a
 33    [match string](https://github.com/egnor/ok-py-serial#port-matching)
 34    or `PortInfo -> bool` callable to filter the ports returned.
 35
 36    For testing and encapsulation, if the environment variable
 37    `$OK_SERIAL_SCAN_OVERRIDE` is the pathname of a JSON file in
 38    `{"device-path": {"attr": "value", ...}, ...}` format, that port listing
 39    is returned instead of actual system scan results.
 40
 41    Raises:
 42    - `SerialScanException` - System error scanning ports
 43    """
 44
 45    if ov_path := os.getenv("OK_SERIAL_SCAN_OVERRIDE"):
 46        # Externally overridden port list
 47        try:
 48            with open(ov_path) as file:
 49                found = _ports_from_json_text(file.read())
 50        except (OSError, ValueError) as ex:
 51            msg = f"Can't read $OK_SERIAL_SCAN_OVERRIDE {ov_path}"
 52            raise SerialScanException(msg) from ex
 53
 54        log.debug("Read $OK_SERIAL_SCAN_OVERRIDE %s", ov_path)
 55    else:
 56        # Use pyserial's port scanner
 57        try:
 58            pyserial_ports = list_ports.comports()
 59        except OSError as ex:
 60            raise SerialScanException("Can't scan serial") from ex
 61        found = []
 62        for pyserial_port in pyserial_ports:
 63            if port := _port_from_pyserial(pyserial_port):
 64                log.debug("pyserial port: %s", port)
 65                found.append(port)
 66
 67        # If `match` is a valid serial device (or symlink thereto), include it
 68        by_path = {os.path.realpath(p.name): p for p in found}
 69        if match_path := isinstance(match, str) and os.path.realpath(match):
 70            if not (exact_port := by_path.get(match_path)):
 71                if exact_port := _port_from_path(match_path):
 72                    log.debug("Named device: %s", exact_port)
 73                    found.append(exact_port)
 74                    by_path[match_path] = exact_port
 75            if exact_port:
 76                # make sure it will pass the filter
 77                exact_port.attr["path_found"] = match
 78
 79        # Look for device symlink aliases
 80        link_top = "/dev/serial"
 81        for dir, dirnames, filenames in os.walk(link_top):
 82            assert dir.startswith(link_top)
 83            dirnames.sort(key=_PATH_SORT_KEY)
 84            filenames.sort(key=_PATH_SORT_KEY)
 85            key_base = _NONALNUM_RX.sub("_", dir[len(link_top) :]).strip("_")
 86            for filename in filenames:
 87                if (link := os.path.join(dir, filename)) not in by_path:
 88                    if port := by_path.get(os.path.realpath(link)):
 89                        key, n = f"link_{key_base}", 1
 90                        while key in port.attr:
 91                            key = f"link_{key_base}_" + str(n := n + 1)
 92                        port.attr[key] = link
 93
 94    if match:
 95        culled = list(filter(compile_match(match), found))
 96        log.debug("Found %d ports, %d match %r", len(found), len(culled), match)
 97    else:
 98        culled = found
 99        log.debug("Found %d ports", len(found))
100
101    culled.sort(key=_INFO_SORT_KEY)
102    return culled

21def scan_uf2_devices(
22    match: str | PortPredicate | None = None,
23) -> list[PortInfo]:
24    """
25    Returns a list of mounted [UF2](https://microsoft.github.io/uf2/)
26    filesystems. (These are not serial ports, but may be of common interest.)
27
28    If set, `match` is a
29    [match string](https://github.com/egnor/ok-py-serial#port-matching)
30    or `PortInfo -> bool` callable to filter the devices returned.
31
32    For testing and encapsulation, if the environment variable
33    `$OK_SERIAL_SCAN_UF2_OVERRIDE` is the pathname of a JSON file in
34    `{"path-name": {"attr": "value", ...}, ...}` format, that port listing
35    is returned instead of actual system scan results.
36
37    Raises:
38    - `SerialScanException` - System error scanning filesystems.
39    """
40
41    if ov_path := os.getenv("OK_SERIAL_SCAN_UF2_OVERRIDE"):
42        # Externally overridden device list
43        try:
44            with open(ov_path) as file:
45                found = _devices_from_json_text(file.read())
46        except (OSError, ValueError) as ex:
47            msg = f"Can't read $OK_SERIAL_SCAN_UF2_OVERRIDE {ov_path}"
48            raise SerialScanException(msg) from ex
49
50        log.debug("Read $OK_SERIAL_SCAN_OVERRIDE %s", ov_path)
51    else:
52        try:
53            partitions = psutil.disk_partitions()
54        except OSError as ex:
55            raise SerialScanException("Can't scan mount points") from ex
56
57        # Include `match` in the list of dirs in case it's a direct pathname
58        mpoints = set(os.path.realpath(p.mountpoint) for p in partitions)
59        if match_path := isinstance(match, str) and os.path.realpath(match):
60            mpoints.add(match_path)
61
62        found = []
63        for mpoint in mpoints:
64            if dev := _device_from_dir(mpoint):
65                found.append(dev)
66                if mpoint == match_path:
67                    assert isinstance(match, str)
68                    dev.attr["path_found"] = match
69
70    if match:
71        culled = list(filter(compile_match(match), found))
72        n_found, n_match = len(found), len(culled)
73        log.debug("Found %d UF2 devices, %d match %r", n_found, n_match, match)
74    else:
75        culled = found
76        log.debug("Found %d UF2 devices", len(found))
77
78    culled.sort(key=_INFO_SORT_KEY)
79    return culled

 6@dataclasses.dataclass(frozen=True)
 7class PortInfo:
 8    """Metadata about a serial port found on the system"""
 9
10    name: str
11    """The OS device identifier, eg. `/dev/ttyUSB3`, 'COM4', etc."""
12
13    attr: dict[str, str]
14    """
15    [Attributes](https://github.com/egnor/py-ok-serial#serial-port-attributes)
16    """
17
18    def __str__(self):
19        return self.name

 36class SerialConnectionMonitor(contextlib.AbstractContextManager):
 37    """
 38    Utility class to maintain a connection to a serial port of interest,
 39    re-scanning and re-connecting as needed after errors, with periodic retry.
 40    This is used for robust communication with a serial device which might be
 41    plugged and unplugged during operation.
 42
 43    Thread-safe: any method may be called from any thread any time, and any
 44    `*_async` method may be awaited from any event loop in any thread any time.
 45    """
 46
 47    def __init__(
 48        self,
 49        match: str | PortPredicate | None = None,
 50        *,
 51        baud: int = 0,
 52        copts: SerialConnectionOptions = SerialConnectionOptions(),
 53        mopts: SerialMonitorOptions = SerialMonitorOptions(),
 54    ):
 55        """
 56        Prepare to manage a serial port connection.
 57        - `match` selects the port of interest: a
 58          [match string](https://github.com/egnor/ok-py-serial#port-matching),
 59          a `PortInfo -> bool` callable, or `None` for any port
 60        - `copts` can define parameters for connecting (eg. baud rate)
 61          - OR `baud` can set the baud rate (as a shortcut)
 62        - `mopts` can define parameters for tracking (eg. re-scan interval)
 63
 64        Actual port scans and connections only happen when `connect_*`
 65        is called. Use `SerialConnectionMonitor` as the target of a `with`
 66        statement to release any open connection when you're done with it.
 67        (After that, don't use it again, create a fresh one to resume.)
 68        """
 69
 70        if baud:
 71            copts = dataclasses.replace(copts, baud=baud)
 72
 73        self._match = match
 74        self._copts = copts
 75        self._mopts = mopts
 76
 77        self._lock = threading.Lock()
 78        self._scan_matched: PortInfo | None = None
 79        self._scan_deadline: float | None = None
 80        self._next_scan = 0.0
 81        self._conn: SerialConnection | None = None
 82        self._conn_error: SerialException | None = None
 83
 84        log.debug("Tracking %r", match or "(any port)")
 85
 86    def __exit__(self, exc_type, exc_value, traceback):
 87        """Closes any open connection with `SerialConnection.close`."""
 88
 89        with self._lock:
 90            if self._conn:
 91                log.debug("Closing %s", self._conn.port_name)
 92                self._conn.close()
 93                self._conn = None
 94
 95    def __repr__(self) -> str:
 96        return (
 97            f"SerialConnectionMonitor({self._match!r}, "
 98            f"copts={self._copts!r}), "
 99            f"mopts={self._mopts!r}"
100        )
101
102    def connect_sync(
103        self, timeout: float | int | None = None
104    ) -> SerialConnection | None:
105        """
106        If a connection is established and healthy, returns it immediately.
107
108        Otherwise, waits up to `timeout` seconds (forever for `None`) for
109        serial port(s) to appear matching this monitor's requirements,
110        returning the first successful connection from among them.
111
112        Returns `None` on reaching the timeout argument.
113
114        Raises:
115        - `SerialScanException` - System error scanning ports
116        - `SerialMonitorExhausted` - Gave up scanning (see `scan_timeout`)
117        """
118
119        call_deadline = to_deadline(timeout)
120        while True:
121            with self._lock:
122                # Return an existing live connection if possible
123                if self._conn:
124                    try:
125                        self._conn.write(b"")  # check for liveness
126                        return self._conn
127                    except SerialIoException as ex:
128                        log_level = 20 if isinstance(ex, SerialIoClosed) else 30
129                        log.log(log_level, "⛓️‍💥 %s (reconnecting)", ex)
130
131                    self._conn.close()
132                    self._conn = None
133
134                if self._scan_deadline is None:
135                    scan_timeout = self._mopts.scan_timeout
136                    self._scan_deadline = to_deadline(scan_timeout)
137                    if scan_timeout is None:
138                        log.info("🔎 Scanning for %r (ongoing)", self._match)
139                    elif scan_timeout > 0:
140                        msg = "🔎 Scanning for %r (%.2fs timeout)"
141                        log.info(msg, self._match, scan_timeout)
142                    else:
143                        log.info("🔎 Looking for %r", self._match)
144
145                # Re-scan for ports at the specified interval
146                if (wait := from_deadline(self._next_scan)) <= 0:
147                    matched = scan_serial_ports(self._match)
148                    if len(matched) == 1:
149                        self._scan_matched = matched[0]
150                    elif matched:
151                        detail = "".join(f"\n  {p}" for p in matched)
152                        msg = f"Multiple ports match {self._match!r}:{detail}"
153                        self._conn_error = SerialScanException(msg)
154                        self._scan_matched = None
155                        log.warning("%s", self._conn_error)
156                    else:
157                        msg = f"No ports match {self._match!r}"
158                        self._conn_error = SerialScanException(msg)
159                        self._scan_matched = None
160                        log.debug("%s", self._conn_error)
161
162                    wait = self._mopts.scan_interval
163                    self._next_scan = to_deadline(wait)
164
165                if port := self._scan_matched:
166                    try:
167                        msg = "🔌 Connecting to %s (%dbps)"
168                        log.info(msg, port.name, self._copts.baud)
169                        conn = SerialConnection(port=port, opts=self._copts)
170                        self._conn, self._conn_error = conn, None
171                        self._scan_deadline = None  # reset for next scan
172                        return self._conn
173                    except SerialOpenException as ex:
174                        self._conn_error = ex
175                        self._scan_matched = None  # cool down until re-scan
176                        log.warning("%s", self._conn_error)
177
178                assert self._scan_deadline is not None  # still under the lock
179                if from_deadline(self._scan_deadline) < wait:
180                    scan_timeout = self._mopts.scan_timeout
181                    msg = f"Can't open {self._match!r}"
182                    if scan_timeout and scan_timeout > 0:
183                        msg += f" ({scan_timeout:.2f}s timeout)"
184                    raise SerialMonitorExhausted(msg) from self._conn_error
185
186            if from_deadline(call_deadline) < wait:
187                return None
188
189            log.debug("Next scan in %.2fs", wait)
190            time.sleep(wait)
191
192    async def connect_async(self) -> SerialConnection:
193        """
194        Similar to `connect_sync` but returns a coroutine instead of
195        blocking the current thread while waiting to retry after failure.
196
197        Note, actual connection attempts do run synchronously, blocking this
198        loop thread. This is not typically unbounded but can take nontrivial
199        time depending on the platform and its approach to USB and such.
200        """
201
202        while True:
203            if conn := self.connect_sync(timeout=0):
204                return conn
205            with self._lock:
206                wait = from_deadline(self._next_scan)
207            log.debug("Next scan in %.2fs", wait)
208            await asyncio.sleep(wait)