ok_serial

 72    def __init__(
 73        self,
 74        *,
 75        match: str | PortPredicate | None = None,
 76        port: str | SerialPort | None = None,
 77        opts: SerialConnectionOptions = SerialConnectionOptions(),
 78        **kwargs,
 79    ):
 80        """
 81        Opens a serial port to make it available for use.
 82        - `match` is a
 83          [match string](https://github.com/egnor/ok-py-serial#port-matching)
 84          or `SerialPort -> bool` callable matching exactly one port...
 85          - OR `port` must name a raw system serial device to open.
 86        - `opts` can define baud rate and other port parameters...
 87          - OR other keywords are forwarded to `SerialConnectionOptions`
 88
 89        Call `close` to release the port, or use `SerialConnection` as the
 90        target of a `with` statement.
 91
 92        Example:
 93        ```
 94        with SerialConnection(match="0403:6001", baud=115200, sharing="polite") as p:
 95            ... interact with `p` ...
 96            # automatically closed on exit from block
 97        ```
 98
 99        Raises:
100        - `SerialOpenException` - I/O error opening the specified port
101        - `SerialOpenBusy` - The port is already in use
102        - `SerialScanException` - System error scanning ports to find `match`
103        """
104
105        assert (match is not None) + (port is not None) == 1
106        opts = dataclasses.replace(opts, **kwargs)
107
108        if match is not None:
109            if not (found := scan_serial_ports(match)):
110                msg = f"No ports match {match!r}"
111                raise _exceptions.SerialOpenException(msg)
112            if len(found) > 1:
113                detail = "".join(f"\n  {p}" for p in found)
114                msg = f"Multiple ports match {match!r}: {detail}"
115                raise _exceptions.SerialOpenException(msg)
116            port = found[0].name
117            log.debug("Scanned %r, found %s", match, port)
118
119        assert port is not None
120        if isinstance(port, SerialPort):
121            port = port.name
122
123        with contextlib.ExitStack() as cleanup:
124            port_lock = cleanup.enter_context(PortLock(port, opts.sharing))
125
126            try:
127                # (If "polite", wake the readloop periodically for checks.)
128                pyserial = cleanup.enter_context(
129                    serial.Serial(
130                        port=port,
131                        baudrate=opts.baud,
132                        write_timeout=0.1,
133                        timeout=(0.5 if opts.sharing == "polite" else None),
134                    )
135                )
136                log.debug("Opened %s %s", port, opts)
137            except OSError as ex:
138                if ex.errno == errno.EBUSY:
139                    msg = "Serial port busy (EBUSY)"
140                    raise _exceptions.SerialOpenBusy(msg, port) from ex
141                else:
142                    msg = "Serial port open error"
143                    raise _exceptions.SerialOpenException(msg, port) from ex
144
145            if hasattr(pyserial, "fileno"):
146                # unlock fd before closing port (see note on release_fd)
147                cleanup.callback(port_lock.release_fd)
148                port_lock.attach_fd(pyserial.fileno())
149
150            self._io = cleanup.enter_context(_IoThreads(pyserial, port_lock))
151            self._io.start()
152            self._cleanup = cleanup.pop_all()

164    def close(self) -> None:
165        """
166        Releases the serial port connection and any associated locks.
167
168        Any I/O operations in progress or attempted after closure will
169        raise an immediate `SerialIoClosed` exception.
170        """
171
172        self._cleanup.close()

174    def read_sync(
175        self,
176        *,
177        timeout: float | int | None = None,
178    ) -> TimestampBytes:
179        """
180        Waits up to `timeout` seconds (forever for `None`) for data,
181        then returns all of it (b"" on timeout).
182
183        Raises:
184        - `SerialIoException` - port I/O failed and there is no matching data
185        - `SerialIoClosed` - the port was closed and there is no matching data
186        """
187
188        deadline = to_deadline(timeout)
189        while True:
190            with self._io.monitor:
191                if self._io.incoming:
192                    monotime = self._io.incoming_monotime
193                    out = TimestampBytes(self._io.incoming, monotime)
194                    self._io.incoming.clear()
195                    self._io.incoming_monotime = 0.0
196                    return out
197                elif self._io.exception:
198                    raise self._io.exception
199                elif (wait := from_deadline(deadline)) <= 0:
200                    return TimestampBytes(b"", 0.0)
201                else:
202                    self._io.monitor.wait(timeout=wait)

204    async def read_async(self) -> TimestampBytes:
205        """
206        Similar to `read_sync` but returns a coroutine instead of
207        blocking the current thread.
208        """
209
210        while True:
211            future = self._io.create_future_in_loop()  # BEFORE read_sync
212            if out := self.read_sync(timeout=0):
213                return out
214            await future

216    def write(self, data: bytes | bytearray) -> None:
217        """
218        Adds data to the outgoing buffer to be sent immediately.
219        Never blocks; the buffer can grow indefinitely.
220        (Use `outgoing_size` and `drain_sync`/`drain_async` to manage
221        buffer size.)
222
223        Raises:
224        - `SerialIoException` - port I/O failed
225        - `SerialIoClosed` - the port was closed
226        """
227
228        with self._io.monitor:
229            if self._io.exception:
230                raise self._io.exception
231            elif data:
232                self._io.outgoing.extend(data)
233                self._io.monitor.notify_all()

235    def drain_sync(self, *, timeout: float | int | None = None) -> bool:
236        """
237        Waits up to `timeout` seconds (forever for `None`) until
238        all buffered data is transmitted.
239
240        Returns `True` if the drain completed, `False` on timeout.
241
242        Raises:
243        - `SerialIoException` - port I/O failed
244        - `SerialIoClosed` - the port was closed
245        """
246
247        deadline = to_deadline(timeout)
248        while True:
249            with self._io.monitor:
250                if self._io.exception:
251                    raise self._io.exception
252                elif not self._io.outgoing:
253                    return True
254                elif (wait := from_deadline(deadline)) <= 0:
255                    return False
256                else:
257                    self._io.monitor.wait(timeout=wait)

259    async def drain_async(self) -> bool:
260        """
261        Similar to `drain_sync` but returns a coroutine instead of
262        blocking the current thread.
263        """
264
265        while True:
266            future = self._io.create_future_in_loop()  # BEFORE drain_sync
267            if self.drain_sync(timeout=0):
268                return True
269            await future

271    def incoming_size(self) -> int:
272        """
273        Returns the number of bytes waiting to be read.
274        """
275        with self._io.monitor:
276            return len(self._io.incoming)

278    def outgoing_size(self) -> int:
279        """
280        Returns the number of bytes waiting to be sent.
281        """
282        with self._io.monitor:
283            return len(self._io.outgoing)

285    def set_signals(
286        self,
287        dtr: bool | None = None,
288        rts: bool | None = None,
289        send_break: bool | None = None,
290    ) -> None:
291        """
292        Sets outgoing
293        [RS-232 modem control line](https://en.wikipedia.org/wiki/RS-232#Data_and_control_signals)
294        state (use `None` for no change):
295        - `dtr` - assert Data Terminal Ready
296        - `rts` - assert Ready To Send
297        - `send_break` - send a continuous BREAK condition
298
299        Raises:
300        - `SerialIoException` - port I/O failed
301        - `SerialIoClosed` - the port was closed
302        """
303
304        with self._io.monitor:
305            if self._io.exception:
306                raise self._io.exception
307            try:
308                if dtr is not None:
309                    self._io.pyserial.dtr = dtr
310                if rts is not None:
311                    self._io.pyserial.rts = rts
312                if send_break is not None:
313                    self._io.pyserial.break_condition = send_break
314            except OSError as ex:
315                msg, dev = "Can't set control signals", self._io.pyserial.port
316                self._io.exception = _exceptions.SerialIoException(msg, dev)
317                self._io.exception.__cause__ = ex
318                raise self._io.exception

320    def get_signals(self) -> SerialControlSignals:
321        """
322        Returns the current
323        [RS-232 modem control line](https://en.wikipedia.org/wiki/RS-232#Data_and_control_signals) state.
324
325        Raises:
326        - `SerialIoException` - port I/O failed
327        - `SerialIoClosed` - the port was closed
328        """
329
330        with self._io.monitor:
331            if self._io.exception:
332                raise self._io.exception
333            try:
334                return SerialControlSignals(
335                    dtr=self._io.pyserial.dtr,
336                    dsr=self._io.pyserial.dsr,
337                    cts=self._io.pyserial.cts,
338                    rts=self._io.pyserial.rts,
339                    ri=self._io.pyserial.ri,
340                    cd=self._io.pyserial.cd,
341                    sending_break=self._io.pyserial.break_condition,
342                )
343            except OSError as ex:
344                msg, dev = "Can't get control signals", self._io.pyserial.port
345                self._io.exception = _exceptions.SerialIoException(msg, dev)
346                self._io.exception.__cause__ = ex
347                raise self._io.exception

349    @property
350    def port_name(self) -> str:
351        """
352        The port's device name, eg. `/dev/ttyACM0` or `COM3`.
353        """
354        return self._io.pyserial.port

356    @property
357    def pyserial(self) -> serial.Serial:
358        """
359        The underlying
360        [`pyserial.Serial`](https://pyserial.readthedocs.io/en/latest/pyserial_api.html#serial.Serial)
361        object (API escape hatch).
362        """
363        return self._io.pyserial

365    def fileno(self) -> int:
366        """
367        The [Unix FD](https://en.wikipedia.org/wiki/File_descriptor)
368        for the serial connection, -1 if not available.
369        """
370        try:
371            return self._io.serial.fileno()
372        except AttributeError:
373            return -1

47    def __init__(
48        self,
49        match: str | PortPredicate | None = None,
50        *,
51        baud: int = 0,
52        topts: SerialTrackerOptions = SerialTrackerOptions(),
53        copts: SerialConnectionOptions = SerialConnectionOptions(),
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 `SerialPort -> bool` callable, or `None` for any port
60        - `topts` can define parameters for tracking (eg. re-scan interval)
61        - `copts` can define parameters for connecting (eg. baud rate)
62          - OR `baud` can set the baud rate (as a shortcut)
63
64        Actual port scans and connections only happen when `connect_*`
65        is called. Call `close` to end any open connection, and/or use
66        `SerialPortTracker` as the target of a `with` statement.
67        """
68
69        if baud:
70            copts = dataclasses.replace(copts, baud=baud)
71
72        self.match = match
73        self._tracker_opts = topts
74        self._conn_opts = copts
75
76        self._lock = threading.Lock()
77        self._baseline_keys: set[str] | None = None
78        self._scan_matched: SerialPort | None = None
79        self._scan_deadline: float | None = None
80        self._next_scan = 0.0
81        self._reconnect_count = 0
82        self._conn: SerialConnection | None = None
83        self._conn_error: SerialException | None = None
84
85        log.debug("Tracking %r", match or "(any port)")

 97    def close(self) -> None:
 98        """
 99        Closes any open connection with `SerialConnection.close`. A subsequent
100        call to `connect_sync`/`connect_async` will establish a new connection.
101        """
102
103        with self._lock:
104            if self._conn:
105                log.debug("Closing %s", self._conn.port_name)
106                self._conn.close()

108    def connect_sync(
109        self, timeout: float | int | None = None
110    ) -> SerialConnection | None:
111        """
112        If a connection is established and healthy, returns it immediately.
113
114        Otherwise, waits up to `timeout` seconds (forever for `None`) for
115        serial port(s) to appear matching this tracker's requirements,
116        returning the first successful connection from among them.
117
118        Returns `None` on reaching the timeout argument.
119
120        Raises:
121        - `SerialScanException` - System error scanning ports
122        - `SerialTrackerExhausted` - Permanent timeout or reconnect limit hit
123        """
124
125        call_deadline = to_deadline(timeout)
126        while True:
127            with self._lock:
128                # Return an existing live connection if possible
129                if self._conn:
130                    try:
131                        self._conn.write(b"")  # check for liveness
132                        return self._conn
133                    except SerialIoException as ex:
134                        if self._tracker_opts.reconnect_limit == 0:
135                            msg = f"{ex} (reconnect disabled)"
136                            raise SerialTrackerExhausted(msg) from ex
137                        log_level = 20 if isinstance(ex, SerialIoClosed) else 30
138                        log.log(log_level, "⛓️‍💥 %s", ex)
139
140                    self._conn.close()
141                    self._conn = None
142                    self._reconnect_count += 1
143                    limit = self._tracker_opts.reconnect_limit
144                    if limit is not None and self._reconnect_count > limit:
145                        msg = f"{self.match!r} reconnect limit met ({limit})"
146                        raise SerialTrackerExhausted(msg)
147
148                if self._scan_deadline is None:
149                    scan_timeout = self._tracker_opts.scan_timeout
150                    self._scan_deadline = to_deadline(scan_timeout)
151                    if scan_timeout is None:
152                        log.info("🔎 Scanning for %r (ongoing)", self.match)
153                    elif scan_timeout > 0:
154                        msg = "🔎 Scanning for %r (%.2fs timeout)"
155                        log.info(msg, self.match, scan_timeout)
156                    else:
157                        log.info("🔎 Looking for %r", self.match)
158
159                # Re-scan for ports at the specified interval
160                if (wait := from_deadline(self._next_scan)) <= 0:
161                    matched = scan_serial_ports(self.match)
162                    if len(matched) == 1:
163                        self._scan_matched = matched[0]
164                    elif matched:
165                        detail = "".join(f"\n  {p}" for p in matched)
166                        msg = f"Multiple ports match {self.match!r}:{detail}"
167                        self._conn_error = SerialScanException(msg)
168                        self._scan_matched = None
169                        log.warning("%s", self._conn_error)
170                    else:
171                        msg = f"No ports match {self.match!r}"
172                        self._conn_error = SerialScanException(msg)
173                        self._scan_matched = None
174                        log.debug("%s", self._conn_error)
175
176                    wait = self._tracker_opts.scan_interval
177                    self._next_scan = to_deadline(wait)
178
179                if port := self._scan_matched:
180                    try:
181                        log.info("🔗 Connecting to %s", port.name)
182                        opts = self._conn_opts
183                        self._conn = SerialConnection(port=port, opts=opts)
184                        self._conn_error = None
185                        self._scan_deadline = None  # reset for next scan
186                        return self._conn
187                    except SerialOpenException as ex:
188                        self._conn_error = ex
189                        self._scan_matched = None  # cool down until re-scan
190                        log.warning("%s", self._conn_error)
191
192            assert self._scan_deadline is not None
193            if from_deadline(self._scan_deadline) < wait:
194                scan_timeout = self._tracker_opts.scan_timeout
195                msg = f"Can't open {self.match!r}"
196                if scan_timeout and scan_timeout > 0:
197                    msg += f" ({scan_timeout:.2f}s timeout)"
198                raise SerialTrackerExhausted(msg) from self._conn_error
199
200            if from_deadline(call_deadline) < wait:
201                return None
202
203            log.debug("Next scan in %.2fs", wait)
204            time.sleep(wait)

206    async def connect_async(self) -> SerialConnection:
207        """
208        Similar to `connect_sync` but returns a coroutine instead of
209        blocking the current thread.
210        """
211
212        while True:
213            if conn := self.connect_sync(timeout=0):
214                return conn
215            with self._lock:
216                wait = from_deadline(self._next_scan)
217            log.debug("Next scan in %.2fs", wait)
218            await asyncio.sleep(wait)

11    def __init__(self, message: str, port: str | None = None):
12        super().__init__(f"{port}: {message}" if port else message)
13        self.port = port