diff --git a/nvcurve/wireview.py b/nvcurve/wireview.py index b93cd61..2be0c22 100644 --- a/nvcurve/wireview.py +++ b/nvcurve/wireview.py @@ -19,11 +19,19 @@ import logging import os import struct import time +from typing import Any -import serial +try: + import serial +except ImportError: # pyserial missing — the serial transport is disabled, + # but the rest of the server keeps running (a stale venv after a code + # update must not take the whole web server down). + serial = None log = logging.getLogger("nvcurve.wireview") +_serial_missing_warned = False + # ── Device identification ───────────────────────────────────────────────────── USB_VENDOR_ID = "0483" # STMicroelectronics (CDC/ACM) @@ -407,12 +415,28 @@ class WireViewSerialDevice: # ── Transport ── + def _open_port(self): + """Open the serial port, or None when pyserial is missing or the + port is unavailable.""" + global _serial_missing_warned + if serial is None: + if not _serial_missing_warned: + _serial_missing_warned = True + log.warning( + "pyserial is not installed — WireView serial transport " + "disabled (install pyserial, e.g. `uv sync`)" + ) + return None + try: + return serial.Serial(self._port, self._baud, timeout=READ_TIMEOUT_S) + except OSError: # SerialException is an OSError + return None + def _read_welcome(self) -> str | None: """Assert RTS and read the NUL-terminated welcome string the device answers with. Null when nothing (or no terminator) arrives in time.""" - try: - ser = serial.Serial(self._port, self._baud, timeout=READ_TIMEOUT_S) - except (serial.SerialException, OSError): + ser = self._open_port() + if ser is None: return None try: ser.reset_input_buffer() @@ -439,7 +463,7 @@ class WireViewSerialDevice: if nul >= 0: return bytes(buf[:nul]).decode("ascii", errors="replace") return None - except (serial.SerialException, OSError): + except OSError: # SerialException is an OSError return None finally: ser.close() @@ -448,9 +472,8 @@ class WireViewSerialDevice: """Open the port, send cmd, read exactly response_size bytes (one second budget), close the port. None when the port is unavailable or the reply is incomplete.""" - try: - ser = serial.Serial(self._port, self._baud, timeout=READ_TIMEOUT_S) - except (serial.SerialException, OSError): + ser = self._open_port() + if ser is None: return None try: ser.reset_input_buffer() @@ -459,13 +482,13 @@ class WireViewSerialDevice: if response_size == 0: return b"" return self._read_exact(ser, response_size) - except (serial.SerialException, OSError): + except OSError: # SerialException is an OSError return None finally: ser.close() @staticmethod - def _read_exact(ser: serial.Serial, size: int) -> bytes | None: + def _read_exact(ser: Any, size: int) -> bytes | None: """Read exactly size bytes within one second, or None.""" buf = bytearray() deadline = time.monotonic() + READ_TIMEOUT_S diff --git a/tests/test_wireview.py b/tests/test_wireview.py index fa0fe72..2f2a8fb 100644 --- a/tests/test_wireview.py +++ b/tests/test_wireview.py @@ -469,6 +469,40 @@ def test_serial_device_no_response(): t.join(timeout=2) +def test_serial_transport_missing_pyserial(): + """A missing pyserial degrades gracefully: connect fails, reads are + None, and the warning is logged once — not on every attempt.""" + import logging + + records: list[str] = [] + + class Capture(logging.Handler): + def emit(self, record: logging.LogRecord) -> None: + records.append(record.getMessage()) + + logger = logging.getLogger("nvcurve.wireview") + handler = Capture() + old_level = logger.level + logger.addHandler(handler) + logger.setLevel(logging.WARNING) + try: + with mock.patch.object(wv, "serial", None): + wv._serial_missing_warned = False + dev = wv.WireViewSerialDevice("/dev/ttyACM99") + check("no-pyserial connect", not dev.connect()) + check("no-pyserial not rejected", not dev.rejected) + check("no-pyserial sample", dev.read_sample() is None) + # A second attempt must not re-warn. + dev2 = wv.WireViewSerialDevice("/dev/ttyACM99") + check("no-pyserial second attempt", not dev2.connect()) + warnings = [r for r in records if "pyserial" in r] + check("no-pyserial warns once", len(warnings) == 1) + finally: + logger.removeHandler(handler) + logger.setLevel(old_level) + wv._serial_missing_warned = False + + def main() -> int: print("wireview tests:") test_parse_sensor_struct() @@ -485,6 +519,7 @@ def main() -> int: test_serial_device_protocol() test_serial_device_rejected_product() test_serial_device_no_response() + test_serial_transport_missing_pyserial() print(f"\n{PASS} passed, {FAIL} failed") return 1 if FAIL else 0