From b8e91e3c91dc3303dd96d878d43f25d4cd558f4c Mon Sep 17 00:00:00 2001 From: ARIA Date: Sat, 3 Oct 2026 13:32:43 +0200 Subject: [PATCH] fix: degrade gracefully when pyserial is missing MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A stale venv after a code update (editable install + git pull) left pyserial out of the environment, and the hard module-level import in wireview.py took the entire web server down — even on machines without a WireView device. Guard the import: without pyserial the serial transport is disabled (one-time warning, connect fails, reads return None) while the rest of the server keeps running. The hwmon transport is unaffected. Also simplify the except clauses to OSError (SerialException is an OSError subclass) so they no longer reference the possibly-None module. --- nvcurve/wireview.py | 43 ++++++++++++++++++++++++++++++++---------- tests/test_wireview.py | 35 ++++++++++++++++++++++++++++++++++ 2 files changed, 68 insertions(+), 10 deletions(-) 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