ok_serial

 53class SerialConnection(contextlib.AbstractContextManager):
 54    """An open connection to a serial port."""
 55
 56    def __init__(
 57        self,
 58        *,
 59        match: str | SerialPortMatcher | None = None,
 60        port: str | SerialPort | None = None,
 61        opts: SerialConnectionOptions = SerialConnectionOptions(),
 62        **kwargs,
 63    ):
 64        """
 65        Opens a serial port to make it available for use.
 66        - `match` can be a [port match expression](https://github.com/egnor/ok-py-serial#serial-port-match-expressions) matching exactly one port...
 67          - OR `port` must name a raw system serial device to open.
 68        - `opts` can define baud rate and other port parameters...
 69          - OR other keywords are forwarded to `SerialConnectionOptions`
 70
 71        Call `close` to release the port; use
 72        `SerialConnection` as the target of a
 73        [`with` statement](https://docs.python.org/3/reference/compound_stmts.html#with)
 74        to automatically close the port on exit from the `with` body.
 75
 76        Example:
 77        ```
 78        with SerialConnection(match="vid_pid=0403:6001", baud=115200, sharing="polite") as p:
 79            ... interact with `p` ...
 80            # automatically closed on exit from block
 81        ```
 82
 83        Raises:
 84        - `SerialOpenException`: I/O error opening the specified port
 85        - `SerialOpenBusy`: The port is already in use
 86        - `SerialScanException`: System error scanning ports to find `match`
 87        - `SerialMatcherInvalid`: Bad format of `match` string
 88        """
 89
 90        assert (match is not None) + (port is not None) == 1
 91        opts = dataclasses.replace(opts, **kwargs)
 92
 93        if match is not None:
 94            if isinstance(match, str):
 95                match = SerialPortMatcher(match)
 96            if not (found := scan_serial_ports()):
 97                raise _exceptions.SerialOpenException("No ports found")
 98            if not (matched := match.filter(found)):
 99                msg = f"No ports match {match!r}"
100                raise _exceptions.SerialOpenException(msg)
101            if len(matched) > 1:
102                matched_text = "".join(f"\n  {p}" for p in matched)
103                msg = f'Multiple ports match "{match}": {matched_text}'
104                raise _exceptions.SerialOpenException(msg)
105            port = matched[0].name
106            log.debug("Scanned %r, found %s", match, port)
107
108        assert port is not None
109        if isinstance(port, SerialPort):
110            port = port.name
111
112        with contextlib.ExitStack() as cleanup:
113            cleanup.enter_context(using_lock_file(port, opts.sharing))
114
115            try:
116                pyserial = cleanup.enter_context(
117                    serial.Serial(
118                        port=port,
119                        baudrate=opts.baud,
120                        write_timeout=0.1,
121                    )
122                )
123                log.debug("Opened %s %s", port, opts)
124            except OSError as ex:
125                if ex.errno == errno.EBUSY:
126                    msg = "Serial port busy (EBUSY)"
127                    raise _exceptions.SerialOpenBusy(msg, port) from ex
128                else:
129                    msg = "Serial port open error"
130                    raise _exceptions.SerialOpenException(msg, port) from ex
131
132            if hasattr(pyserial, "fileno"):
133                fd, share = pyserial.fileno(), opts.sharing
134                cleanup.enter_context(using_fd_lock(port, fd, share))
135
136            self._io = cleanup.enter_context(_IoThreads(pyserial))
137            self._io.start()
138            self._cleanup = cleanup.pop_all()
139
140    def __del__(self) -> None:
141        if hasattr(self, "_cleanup"):
142            self._cleanup.close()
143
144    def __exit__(self, exc_type, exc_value, traceback) -> None:
145        self._cleanup.__exit__(exc_type, exc_value, traceback)
146
147    def __repr__(self) -> str:
148        return f"SerialConnection({self._io.pyserial.port!r})"
149
150    def close(self) -> None:
151        """
152        Releases the serial port connection and any associated locks.
153
154        Any I/O operations in progress or attempted after closure will
155        raise an immediate `SerialIoClosed` exception.
156        """
157
158        self._cleanup.close()
159
160    def read_sync(
161        self,
162        *,
163        timeout: float | int | None = None,
164    ) -> bytes:
165        """
166        Waits up to `timeout` seconds (forever for `None`) for data,
167        then returns all of it (b"" on timeout).
168
169        Raises:
170        - `SerialIoException`: port I/O failed and there is no matching data
171        - `SerialIoClosed`: the port was closed and there is no matching data
172        """
173
174        deadline = to_deadline(timeout)
175        while True:
176            with self._io.monitor:
177                if self._io.incoming:
178                    output = bytes(self._io.incoming)
179                    self._io.incoming.clear()
180                    return output
181                elif self._io.exception:
182                    raise self._io.exception
183                elif (wait := from_deadline(deadline)) <= 0:
184                    return b""
185                else:
186                    self._io.monitor.wait(timeout=wait)
187
188    async def read_async(self) -> bytes:
189        """
190        Similar to `read_sync` but returns a
191        [`Future`](https://docs.python.org/3/library/asyncio-future.html#asyncio.Future)
192        instead of blocking the current thread.
193        """
194
195        while True:
196            future = self._io.create_future_in_loop()  # BEFORE read_sync
197            if out := self.read_sync(timeout=0):
198                return out
199            await future
200
201    def write(self, data: bytes | bytearray) -> None:
202        """
203        Adds data to the outgoing buffer to be sent immediately.
204        Never blocks; the buffer can grow indefinitely.
205        (Use `outgoing_size` and `drain_sync`/`drain_async` to manage
206        buffer size.)
207
208        Raises:
209        - `SerialIoException`: port I/O failed
210        - `SerialIoClosed`: the port was closed
211        """
212
213        with self._io.monitor:
214            if self._io.exception:
215                raise self._io.exception
216            elif data:
217                self._io.outgoing.extend(data)
218                self._io.monitor.notify_all()
219
220    def drain_sync(self, *, timeout: float | int | None = None) -> bool:
221        """
222        Waits up to `timeout` seconds (forever for `None`) until
223        all buffered data is transmitted.
224
225        Returns `True` if the drain completed, `False` on timeout.
226
227        Raises:
228        - `SerialIoException`: port I/O failed
229        - `SerialIoClosed`: the port was closed
230        """
231
232        deadline = to_deadline(timeout)
233        while True:
234            with self._io.monitor:
235                if self._io.exception:
236                    raise self._io.exception
237                elif not self._io.outgoing:
238                    return True
239                elif (wait := from_deadline(deadline)) <= 0:
240                    return False
241                else:
242                    self._io.monitor.wait(timeout=wait)
243
244    async def drain_async(self) -> bool:
245        """
246        Similar to `drain_sync` but returns a
247        [`Future`](https://docs.python.org/3/library/asyncio-future.html#asyncio.Future)
248        instead of blocking the current thread.
249        """
250
251        while True:
252            future = self._io.create_future_in_loop()  # BEFORE drain_sync
253            if self.drain_sync(timeout=0):
254                return True
255            await future
256
257    def incoming_size(self) -> int:
258        """
259        Returns the number of bytes waiting to be read.
260        """
261        with self._io.monitor:
262            return len(self._io.incoming)
263
264    def outgoing_size(self) -> int:
265        """
266        Returns the number of bytes waiting to be sent.
267        """
268        with self._io.monitor:
269            return len(self._io.outgoing)
270
271    def set_signals(
272        self,
273        dtr: bool | None = None,
274        rts: bool | None = None,
275        send_break: bool | None = None,
276    ) -> None:
277        """
278        Sets outgoing
279        [RS-232 modem control line](https://en.wikipedia.org/wiki/RS-232#Data_and_control_signals)
280        state (use `None` for no change):
281        - `dtr`: assert Data Terminal Ready
282        - `rts`: assert Ready To Send
283        - `send_break`: send a continuous BREAK condition
284
285        Raises:
286        - `SerialIoException`: port I/O failed
287        - `SerialIoClosed`: the port was closed
288        """
289
290        with self._io.monitor:
291            if self._io.exception:
292                raise self._io.exception
293            try:
294                if dtr is not None:
295                    self._io.pyserial.dtr = dtr
296                if rts is not None:
297                    self._io.pyserial.rts = rts
298                if send_break is not None:
299                    self._io.pyserial.break_condition = send_break
300            except OSError as ex:
301                msg, dev = "Can't set control signals", self._io.pyserial.port
302                self._io.exception = _exceptions.SerialIoException(msg, dev)
303                self._io.exception.__cause__ = ex
304                raise self._io.exception
305
306    def get_signals(self) -> SerialControlSignals:
307        """
308        Returns the current
309        [RS-232 modem control line](https://en.wikipedia.org/wiki/RS-232#Data_and_control_signals) state.
310
311        Raises:
312        - `SerialIoException`: port I/O failed
313        - `SerialIoClosed`: the port was closed
314        """
315
316        with self._io.monitor:
317            if self._io.exception:
318                raise self._io.exception
319            try:
320                return SerialControlSignals(
321                    dtr=self._io.pyserial.dtr,
322                    dsr=self._io.pyserial.dsr,
323                    cts=self._io.pyserial.cts,
324                    rts=self._io.pyserial.rts,
325                    ri=self._io.pyserial.ri,
326                    cd=self._io.pyserial.cd,
327                    sending_break=self._io.pyserial.break_condition,
328                )
329            except OSError as ex:
330                msg, dev = "Can't get control signals", self._io.pyserial.port
331                self._io.exception = _exceptions.SerialIoException(msg, dev)
332                self._io.exception.__cause__ = ex
333                raise self._io.exception
334
335    @property
336    def port_name(self) -> str:
337        """
338        The port's device name, eg. `/dev/ttyACM0` or `COM3`.
339        """
340        return self._io.pyserial.port
341
342    @property
343    def pyserial(self) -> serial.Serial:
344        """
345        The underlying
346        [`pyserial.Serial`](https://pyserial.readthedocs.io/en/latest/pyserial_api.html#serial.Serial)
347        object (API escape hatch).
348        """
349        return self._io.pyserial
350
351    def fileno(self) -> int:
352        """
353        The [Unix FD](https://en.wikipedia.org/wiki/File_descriptor)
354        for the serial connection, -1 if not available.
355        """
356        try:
357            return self._io.serial.fileno()
358        except AttributeError:
359            return -1

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

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

37def scan_serial_ports() -> list[SerialPort]:
38    """
39    Returns a list of serial ports currently attached to the system.
40
41    For testing and encapsulation, if the environment variable
42    `$OK_SERIAL_SCAN_OVERRIDE` is the pathname of a JSON file in
43    `{"port-name": {"attr": "value", ...}, ...}` format, that port listing
44    is returned instead of actual system scan results.
45
46    Raises:
47    - `SerialScanException`: System error scanning ports
48    """
49
50    if ov_path := os.getenv("OK_SERIAL_SCAN_OVERRIDE"):
51        try:
52            out = _ports_from_json_text(pathlib.Path(ov_path).read_text())
53        except (OSError, ValueError) as ex:
54            msg = f"Can't read $OK_SERIAL_SCAN_OVERRIDE {ov_path}"
55            raise SerialScanException(msg) from ex
56
57        log.debug("Read $OK_SERIAL_SCAN_OVERRIDE %s", ov_path)
58    else:
59        try:
60            pyserial_ports = list_ports.comports()
61        except OSError as ex:
62            raise SerialScanException("Can't scan serial") from ex
63        out = []
64        for pyserial_port in pyserial_ports:
65            if port := _port_from_pyserial(pyserial_port):
66                out.append(port)
67
68    out.sort(key=natsort.natsort_keygen(key=lambda p: p.name, alg=natsort.ns.P))
69    log.debug("Found %d ports", len(out))
70    return out

21@dataclasses.dataclass(frozen=True)
22class SerialPort:
23    """Metadata about a serial port found on the system"""
24
25    name: str
26    """The OS device identifier, eg. `/dev/ttyUSB3`, 'COM4', etc."""
27
28    attr: dict[str, str]
29    """
30    [Metadata](https://github.com/egnor/py-ok-serial#serial-port-attributes)
31    """
32
33    def __str__(self):
34        return self.name

 57class SerialPortMatcher:
 58    """A parsed expression for identifying serial ports of interest."""
 59
 60    def __init__(self, match: str):
 61        """Parses a
 62        [serial port match expression](https://github.com/egnor/ok-py-serial#serial-port-match-expressions)
 63        in preparation to identify matching `SerialPort` objects.
 64        """
 65
 66        self._input = match
 67        self._pos: list[_Rule] = []
 68        self._neg: list[_Rule] = []
 69        self._oldest = False
 70        self._newest = False
 71
 72        next_pos = 0
 73        while next_pos < len(match):
 74            tm = _TERM_RE.match(match, pos=next_pos)
 75            if not (tm and tm[0]):
 76                jmatch = json.dumps(match)  # use JSON for consistent quoting
 77                jpre = json.dumps(match[:next_pos])
 78                msg = f"Bad port expr:\n  {jmatch}\n -{'-' * (len(jpre) - 1)}^"
 79                raise SerialMatcherInvalid(msg)
 80
 81            next_pos, groups = tm.end(), tm.groups(default="")
 82            rx_att, rx_ex, rx_text, att, eq_ex, ex, qstr, num, naked = groups
 83            out_list = self._neg if bool(rx_ex or eq_ex or ex) else self._pos
 84            if rx_text:
 85                try:
 86                    out_list.append(_Rule(rx_att, re.compile(rx_text)))
 87                except re.error as ex:
 88                    msg = f"Bad port regex: /{rx_text}/"
 89                    raise SerialMatcherInvalid(msg) from ex
 90            elif qstr:
 91                rx = _qstr_rx(qstr, glob=False, full=bool(att))
 92                out_list.append(_Rule(att, rx))
 93            elif num:
 94                if num[-1:] in "hH":
 95                    value = int("{num[:-1]", 16)
 96                elif num.isdigit():
 97                    value = int(num, 10)  # int('0123', 0) is an error!
 98                else:
 99                    value = int(num, 0)
100                rx_text = f"({num}|0*{value}|(0x)?0*{value:x}h?)"
101                rx = _wrap_rx(rx_text, full=bool(att), wb=True, we=True)
102                out_list.append(_Rule(att, rx))
103            elif _OLDEST_RE.match(naked) and not att and out_list is self._pos:
104                self._oldest = True
105            elif _NEWEST_RE.match(naked) and not att and out_list is self._pos:
106                self._newest = True
107            elif naked:
108                rx = _qstr_rx(naked, glob=True, full=bool(att))
109                out_list.append(_Rule(att, rx))
110            else:
111                assert False, f"bad term match: {tm[0]!r}"
112
113        if self._oldest and self._newest:
114            msg = f"oldest & newest: {match!r}"
115            raise SerialMatcherInvalid(msg)
116        elif not self:
117            log.debug("Parsed %s (any port)", repr(match))
118        elif log.isEnabledFor(logging.DEBUG):
119            rules = "".join(f"\n  {r}" for r in self.patterns())
120            log.debug("Parsed %s:%s", repr(match), rules)
121
122    def __repr__(self) -> str:
123        return f"SerialPortMatcher({self._input!r})"
124
125    def __str__(self) -> str:
126        return self._input
127
128    def __bool__(self) -> bool:
129        return bool(self._oldest or self._newest or self._pos or self._neg)
130
131    def patterns(self) -> list[str]:
132        return [
133            *(["oldest"] if self._oldest else []),
134            *(["newest"] if self._newest else []),
135            *[str(r).replace("~/", "!~/", 1) for r in self._neg],
136            *[str(r) for r in self._pos],
137        ]
138
139    def filter(self, ports: list[SerialPort]) -> list[SerialPort]:
140        """Filters a list of ports according to the match criteria."""
141
142        matches = []
143        for p in ports:
144            attrs = list(p.attr.items())
145            pos = all(any(r.match(k, v) for k, v in attrs) for r in self._pos)
146            neg = any(r.match(k, v) for k, v in attrs for r in self._neg)
147            if pos and not neg:
148                matches.append(p)
149
150        if self._oldest or self._newest:
151            timed = [m for m in matches if "time" in p.attr]
152            timed.sort(key=lambda m: m.attr["time"])
153            return timed[:1] if self._oldest else timed[-1:]
154
155        return matches
156
157    def attr_hit(self, port: SerialPort, key: str) -> bool:
158        """True if the attribute given by 'key' is a positive match."""
159
160        if v := port.attr.get(key, ""):
161            return any(r.match(key, v) for r in self._pos)
162        return False