fix: degrade gracefully when pyserial is missing

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.
This commit is contained in:
ARIA committed 2026-10-03 13:32:43 +02:00
1 parent 526b71b45f
commit b8e91e3c91
2 files changed
+67 -9

No files matched your search

+32 -9
View File
@@ -19,11 +19,19 @@ import logging
import os import os
import struct import struct
import time import time
from typing import Any
try:
import serial 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") log = logging.getLogger("nvcurve.wireview")
_serial_missing_warned = False
# ── Device identification ───────────────────────────────────────────────────── # ── Device identification ─────────────────────────────────────────────────────
USB_VENDOR_ID = "0483" # STMicroelectronics (CDC/ACM) USB_VENDOR_ID = "0483" # STMicroelectronics (CDC/ACM)
@@ -407,12 +415,28 @@ class WireViewSerialDevice:
# ── Transport ── # ── 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: def _read_welcome(self) -> str | None:
"""Assert RTS and read the NUL-terminated welcome string the device """Assert RTS and read the NUL-terminated welcome string the device
answers with. Null when nothing (or no terminator) arrives in time.""" answers with. Null when nothing (or no terminator) arrives in time."""
try: ser = self._open_port()
ser = serial.Serial(self._port, self._baud, timeout=READ_TIMEOUT_S) if ser is None:
except (serial.SerialException, OSError):
return None return None
try: try:
ser.reset_input_buffer() ser.reset_input_buffer()
@@ -439,7 +463,7 @@ class WireViewSerialDevice:
if nul >= 0: if nul >= 0:
return bytes(buf[:nul]).decode("ascii", errors="replace") return bytes(buf[:nul]).decode("ascii", errors="replace")
return None return None
except (serial.SerialException, OSError): except OSError: # SerialException is an OSError
return None return None
finally: finally:
ser.close() ser.close()
@@ -448,9 +472,8 @@ class WireViewSerialDevice:
"""Open the port, send cmd, read exactly response_size bytes (one """Open the port, send cmd, read exactly response_size bytes (one
second budget), close the port. None when the port is unavailable second budget), close the port. None when the port is unavailable
or the reply is incomplete.""" or the reply is incomplete."""
try: ser = self._open_port()
ser = serial.Serial(self._port, self._baud, timeout=READ_TIMEOUT_S) if ser is None:
except (serial.SerialException, OSError):
return None return None
try: try:
ser.reset_input_buffer() ser.reset_input_buffer()
@@ -459,13 +482,13 @@ class WireViewSerialDevice:
if response_size == 0: if response_size == 0:
return b"" return b""
return self._read_exact(ser, response_size) return self._read_exact(ser, response_size)
except (serial.SerialException, OSError): except OSError: # SerialException is an OSError
return None return None
finally: finally:
ser.close() ser.close()
@staticmethod @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.""" """Read exactly size bytes within one second, or None."""
buf = bytearray() buf = bytearray()
deadline = time.monotonic() + READ_TIMEOUT_S deadline = time.monotonic() + READ_TIMEOUT_S
+35
View File
@@ -469,6 +469,40 @@ def test_serial_device_no_response():
t.join(timeout=2) 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: def main() -> int:
print("wireview tests:") print("wireview tests:")
test_parse_sensor_struct() test_parse_sensor_struct()
@@ -485,6 +519,7 @@ def main() -> int:
test_serial_device_protocol() test_serial_device_protocol()
test_serial_device_rejected_product() test_serial_device_rejected_product()
test_serial_device_no_response() test_serial_device_no_response()
test_serial_transport_missing_pyserial()
print(f"\n{PASS} passed, {FAIL} failed") print(f"\n{PASS} passed, {FAIL} failed")
return 1 if FAIL else 0 return 1 if FAIL else 0