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

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