diff --git a/Makefile b/Makefile index eca3bef..b66c615 100644 --- a/Makefile +++ b/Makefile @@ -25,6 +25,7 @@ dev: ## Create/refresh the dev environment (uv sync) test: ## Run the test suite $(UV) run python tests/test_security.py $(UV) run python tests/test_rm_power.py + $(UV) run python tests/test_wireview.py clean: ## Remove build artifacts rm -rf frontend/dist frontend/node_modules \ No newline at end of file diff --git a/README.md b/README.md index 78e31ef..db88bcd 100644 --- a/README.md +++ b/README.md @@ -15,6 +15,7 @@ NVCurve brings MSI Afterburner-style per-point voltage-frequency curve control t > **Blackwell GPU memory** — This is a specialized fork with extended memory offset support (up to +3000 MHz) for Blackwell GPUs (RTX 50-series). \ > **Fan Controls** — There is an additional "Fans" tab to setup a customized fan curve, controlling all fans or individual fans. \ > **Dashboard** — The default tab is an Dashboard with additional information (PCIe link speed, VBIOS information, Max Core Clock, Throttle Reason and much much more.) \ +> **WireView Pro II** — Native monitoring of the Thermal Grizzly WireView Pro II 12VHPWR connector: the device is auto-detected over USB and a dedicated tab shows live temperature, power and current readings — no external exporter or kernel module required. The tab only appears while the device is connected.\ > **Authentification** — For production deplyoment, I added authentification with bcrypt hashing to allow only one or multiple people to have access. \ > Installing the pre-built PyPI package will NOT include these features. You must build from source. @@ -27,6 +28,9 @@ NVCurve brings MSI Afterburner-style per-point voltage-frequency curve control t Performance Fans + + WireView + ## Prerequisites diff --git a/docs/wireview.png b/docs/wireview.png new file mode 100644 index 0000000..6024605 Binary files /dev/null and b/docs/wireview.png differ diff --git a/frontend/src/App.tsx b/frontend/src/App.tsx index d763434..bed0e49 100644 --- a/frontend/src/App.tsx +++ b/frontend/src/App.tsx @@ -11,10 +11,12 @@ import { PerformancePanel } from "./components/Limits/PerformancePanel.js"; import { PerformanceMonitor } from "./components/Monitor/PerformanceMonitor.js"; import { FanMonitor } from "./components/Monitor/FanMonitor.js"; import { FanCurveEditor } from "./components/Fans/FanCurveEditor.js"; +import { WireViewPanel } from "./components/WireView/WireViewPanel.js"; import { ProfilePanel } from "./components/Profiles/ProfilePanel.js"; import { api, onUnauthorized } from "./api/client.js"; import { LoginScreen } from "./components/Auth/LoginScreen.js"; import { useCurveStore } from "./store/curveStore.js"; +import { useWireview } from "./hooks/useWireview.js"; import { Toaster } from "sonner"; import { Loader, ChevronDown } from "lucide-react"; import { useState, useRef, useEffect } from "react"; @@ -94,9 +96,10 @@ function MainApp({ const { dashboard, loading: dashboardLoading } = useDashboard(); const { setCurve, activeProfile, setActiveProfile, selectedGpuIndex } = useCurveStore(); + const wireview = useWireview(); const [activeTab, setActiveTab] = useState< - "dashboard" | "curve" | "performance" | "fans" + "dashboard" | "curve" | "performance" | "fans" | "wireview" >("dashboard"); const [fanState, setFanState] = useState(null); const [activeDomain, setActiveDomain] = useState<"gpu" | "memory">("gpu"); @@ -195,6 +198,15 @@ function MainApp({ > Fans + {wireview.available && ( + + )}
+ ) : activeTab === "wireview" ? ( + ) : (
diff --git a/frontend/src/components/WireView/WireViewPanel.tsx b/frontend/src/components/WireView/WireViewPanel.tsx new file mode 100644 index 0000000..b5f76bd --- /dev/null +++ b/frontend/src/components/WireView/WireViewPanel.tsx @@ -0,0 +1,309 @@ +import { GaugeCard } from "../Monitor/GaugeCard.js"; +import { fmt } from "../../utils/units.js"; +import type { WireViewInfo, WireViewSample } from "../../types.js"; + +interface Props { + info: WireViewInfo | null; + sample: WireViewSample | null; + history: WireViewSample[]; +} + +// Per-pin over-current protection of the 12 VHPWR connector (A). +const PIN_OCP_A = 55; + +const FAULT_NAMES: Record = { + 0: "Chip over-temperature", + 1: "Sensor over-temperature", + 2: "Over-current (OCP)", + 3: "Wire over-current", + 4: "Over-power (OPP)", + 5: "Current imbalance", +}; + +function decodeFaults(mask: number): string[] { + return Object.entries(FAULT_NAMES) + .filter(([bit]) => mask & (1 << Number(bit))) + .map(([, name]) => name); +} + +function pluck(history: WireViewSample[], key: keyof WireViewSample): number[] { + return history.map((s) => (s[key] as number | null) ?? 0); +} + +function FaultBadge({ + label, + mask, +}: { + label: string; + mask: number; +}) { + const faults = decodeFaults(mask); + return ( +
0 + ? "bg-red-500/10 border-red-500/40" + : "bg-zinc-900 border-zinc-800" + }`} + > +
+ + {label} + + 0 ? "text-red-400" : "text-zinc-600" + }`} + > + 0x{mask.toString(16).toUpperCase().padStart(4, "0")} + +
+ {faults.length > 0 ? ( +
+ {faults.map((f) => ( + + {f} + + ))} +
+ ) : ( +
No faults
+ )} +
+ ); +} + +function PinCard({ + index, + voltage, + current, + power, +}: { + index: number; + voltage: number; + current: number; + power: number; +}) { + const loadPct = Math.min(100, (current / PIN_OCP_A) * 100); + const barColor = + loadPct >= 90 ? "#f87171" : loadPct >= 75 ? "#fbbf24" : "#34d399"; + return ( +
+
+ + Pin {index + 1} + + + {current.toFixed(1)} / {PIN_OCP_A} A + +
+
+
+
V
+
+ {voltage.toFixed(2)} +
+
+
+
A
+
+ {current.toFixed(2)} +
+
+
+
W
+
+ {power.toFixed(1)} +
+
+
+
+
+
+
+ ); +} + +function InfoItem({ + label, + value, +}: { + label: string; + value: string | null | undefined; +}) { + if (!value) return null; + return ( +
+ {label} + + {value} + +
+ ); +} + +export function WireViewPanel({ info, sample, history }: Props) { + if (!sample) { + return ( +
+
+ WireView Pro II not connected +
+
+ Plug in the device — the tab appears automatically once it is + detected. +
+
+ ); + } + + const psuCap = sample.psu_capability_w > 0 ? sample.psu_capability_w : 600; + + return ( +
+ {/* ── Header ─────────────────────────────────────────────────── */} +
+
+ + + {info?.device_name ?? "WireView Pro II"} + +
+ + {info?.hw_rev ? `HW ${info.hw_rev}` : ""} + {info?.firmware_version ? ` · FW ${info.firmware_version}` : ""} + + + PSU {psuCap} W + + + {info?.transport === "hwmon" + ? `hwmon: ${info.port}` + : `serial: ${info?.port ?? ""}`} + +
+ + {/* ── Main gauges ────────────────────────────────────────────── */} +
+
+ 12 VHPWR Connector +
+
+ + + + +
+
+ + {/* ── Per-pin ────────────────────────────────────────────────── */} +
+
+ Per-Pin Readings +
+
+ {sample.pins.map((pin, i) => ( + + ))} +
+
+ + {/* ── Temperatures ───────────────────────────────────────────── */} +
+
+ Temperatures +
+
+ + + + +
+
+ + {/* ── Faults + device info ───────────────────────────────────── */} +
+
+ + +
+
+
+ Device +
+
+ + + + + + +
+
+
+
+ ); +} diff --git a/frontend/src/hooks/useWireview.ts b/frontend/src/hooks/useWireview.ts new file mode 100644 index 0000000..9221ac9 --- /dev/null +++ b/frontend/src/hooks/useWireview.ts @@ -0,0 +1,40 @@ +import { useEffect, useRef, useState } from "react"; +import { createWsConnection } from "../api/websocket.js"; +import type { + WireViewInfo, + WireViewSample, + WireViewWsMessage, +} from "../types.js"; + +const MAX_HISTORY = 120; // ~2 min at 1 s polling + +export function useWireview() { + const [available, setAvailable] = useState(false); + const [info, setInfo] = useState(null); + const [sample, setSample] = useState(null); + const [history, setHistory] = useState([]); + const wsRef = useRef | null>(null); + + useEffect(() => { + wsRef.current = createWsConnection( + "/ws/wireview", + (msg) => { + if (msg.type === "sample") { + setAvailable(true); + setInfo(msg.info); + setSample(msg.sample); + setHistory((h) => [...h.slice(-(MAX_HISTORY - 1)), msg.sample]); + } else { + setAvailable(false); + setInfo(null); + setSample(null); + setHistory([]); + } + }, + () => {}, + ); + return () => wsRef.current?.close(); + }, []); + + return { available, info, sample, history }; +} diff --git a/frontend/src/types.ts b/frontend/src/types.ts index 2e8b8bd..ae7226a 100644 --- a/frontend/src/types.ts +++ b/frontend/src/types.ts @@ -145,3 +145,50 @@ export interface ProfileData { fan_curve: FanPoint[] | null; fan_targets: number[] | null; } + +// ── WireView Pro II (Thermal Grizzly 12 VHPWR connector monitor) ── + +export interface WireViewPin { + voltage_v: number; + current_a: number; + power_w: number; +} + +export interface WireViewSample { + timestamp: number; + power_total_w: number; + current_total_a: number; + voltage_avg_v: number; + pins: WireViewPin[]; + // Temperatures are null when the source has no channel for them + // (e.g. an hwmon module without external-sensor channels). + temp_in_c: number | null; + temp_out_c: number | null; + temp_ext1_c: number | null; + temp_ext2_c: number | null; + fan_duty_pct: number; + fault_status: number; + fault_log: number; + psu_capability_w: number; +} + +export interface WireViewInfo { + device_name: string; + hw_rev: string; + firmware_version: string; + uid: string; + build: string; + transport: "serial" | "hwmon"; + port: string; +} + +export interface WireViewState { + available: boolean; + connected: boolean; + info: WireViewInfo | null; + sample: WireViewSample | null; +} + +export type WireViewWsMessage = + | { type: "unavailable" } + | { type: "sample"; info: WireViewInfo; sample: WireViewSample }; diff --git a/nvcurve/server.py b/nvcurve/server.py index 25c6028..d433abe 100644 --- a/nvcurve/server.py +++ b/nvcurve/server.py @@ -72,6 +72,7 @@ from .profiles.native import ( rename_profile, save_profile, ) +from .wireview import create_device, find_wireview_ports from .safety import check_negative_freq_warnings, validate_write log = logging.getLogger("nvcurve.server") @@ -103,6 +104,15 @@ def _open_browser_as_user(url: str) -> None: _state: dict[str, Any] = { "gpus": {}, # dict[int, dict] mapping gpu_index -> gpu state "config": default_config, + "wireview": { + "clients": set(), # connected /ws/wireview clients + "device": None, # WireViewSerialDevice | WireViewHwmonDevice | None + "info": None, # device identity (static per connection) + "last_sample": None, # most recent sensor sample + "connected": False, # last read succeeded + "failures": 0, # consecutive failed reads + "rejected_ports": set(), # ports reporting an unsupported product + }, } @@ -258,6 +268,126 @@ async def _fan_poller(gpu_index: int) -> None: await asyncio.sleep(2.0) +# ── WireView Pro II (Thermal Grizzly) ───────────────────────────────────────── +# Consecutive failed reads after which a connected device is given up. Same +# rule as the exporter: one corrupt frame or a slow read never trips it, and +# an unplug is caught at once by the port-node check. +WIREVIEW_MAX_FAILED_READS = 6 +WIREVIEW_WATCHDOG_INTERVAL_S = 5.0 + + +async def _wireview_disconnect() -> None: + """Drop the connected WireView device and tell clients it is gone.""" + wv = _state["wireview"] + device = wv["device"] + if device is None: + return + wv["device"] = None + wv["connected"] = False + wv["info"] = None + wv["last_sample"] = None + wv["failures"] = 0 + await _run(device.close) + log.info("WireView disconnected") + await _broadcast(wv["clients"], {"type": "unavailable"}) + + +async def _wireview_connect() -> None: + """Detect and connect a WireView Pro II. No-op when none is attached or + one is already connected.""" + wv = _state["wireview"] + if wv["device"] is not None: + return + # Forget rejected ports that disappeared, so a re-plug gets a fresh probe. + if wv["rejected_ports"]: + present = set(await _run(find_wireview_ports)) + wv["rejected_ports"] &= present + device = await _run(create_device) + if device is None: + return + # A port that reported an unsupported product is not re-probed (and + # re-logged) on every watchdog tick. + if device.port in wv["rejected_ports"]: + return + ok = await _run(device.connect) + if not ok: + await _run(device.close) + if device.rejected: + wv["rejected_ports"].add(device.port) + return + wv["device"] = device + wv["failures"] = 0 + wv["connected"] = True + wv["info"] = await _run(device.info) + # Push the first sample right away so clients don't wait for the next tick. + sample = await _run(device.read_sample) + if sample is not None: + wv["last_sample"] = sample + await _broadcast( + wv["clients"], {"type": "sample", "info": wv["info"], "sample": sample} + ) + else: + wv["connected"] = False + + +async def _wireview_poller() -> None: + """Read the WireView at poll_interval_s and push to connected WS clients. + + Skips reads while no client is subscribed (like the monitor poller), so + idle polling never contends for the port with other tools (official GUI, + wireviewd).""" + cfg: Config = _state["config"] + while True: + try: + wv = _state["wireview"] + device = wv["device"] + if device is not None and wv["clients"]: + sample = await _run(device.read_sample) + if wv["device"] is not device: + # The device was disconnected (unplug) while the read was + # in flight — drop the stale result instead of + # resurrecting state. + pass + elif sample is not None: + wv["failures"] = 0 + wv["connected"] = True + wv["last_sample"] = sample + await _broadcast( + wv["clients"], + {"type": "sample", "info": wv["info"], "sample": sample}, + ) + else: + wv["failures"] += 1 + if ( + not await _run(device.node_exists) + or wv["failures"] >= WIREVIEW_MAX_FAILED_READS + ): + await _wireview_disconnect() + except asyncio.CancelledError: + return + except Exception as exc: + log.warning("WireView poller error: %s", exc) + await asyncio.sleep(cfg.poll_interval_s) + + +async def _wireview_watchdog() -> None: + """Hot-plug detection: connect when a WireView appears, disconnect when + its device node disappears.""" + while True: + await asyncio.sleep(WIREVIEW_WATCHDOG_INTERVAL_S) + try: + wv = _state["wireview"] + device = wv["device"] + if device is None: + await _wireview_connect() + elif not await _run(device.node_exists): + await _wireview_disconnect() + except asyncio.CancelledError: + return + except Exception as exc: + log.warning("WireView watchdog error: %s", exc) + + async def _activate_fan_curve( gpu_index: int, curve: list, fans: list[int] | None = None ) -> None: @@ -380,6 +510,16 @@ async def lifespan(app: FastAPI): except Exception as exc: log.error("Failed to initialize GPU %d: %s", idx, exc) + # ── WireView Pro II detection ───────────────────────────────────────────── + # Independent of GPU discovery: the tab appears whenever the connector + # monitor is attached, and the watchdog handles hot-plug afterwards. + try: + await _wireview_connect() + except Exception as exc: + log.warning("WireView initial detection failed: %s", exc) + poller_tasks.append(asyncio.create_task(_wireview_poller())) + poller_tasks.append(asyncio.create_task(_wireview_watchdog())) + # ── Backward Compatibility Bridge ────────────────────────────────────────── # NOTE: This auto-load path is for users running the server directly (e.g. # via an old systemd unit file that lacks the new daemon mode). @@ -465,6 +605,14 @@ async def lifespan(app: FastAPI): with suppress(asyncio.CancelledError): await task + # Release the WireView port (if connected) so other tools can use it. + wv = _state["wireview"] + if wv["device"] is not None: + with suppress(Exception): + await _run(wv["device"].close) + wv["device"] = None + wv["connected"] = False + for gpu_index, g_state in _state["gpus"].items(): if g_state.get("fan_poller_task"): g_state["fan_poller_task"].cancel() @@ -1442,6 +1590,25 @@ async def api_fans_speed(req: FanSpeedRequest, gpu_index: int = 0): return {"ok": True} +# ── WireView Pro II (Thermal Grizzly) ───────────────────────────────────────── + + +@app.get("/api/wireview") +async def api_wireview(): + """WireView Pro II availability and the most recent sensor sample. + + available: a device is connected (the UI shows the WireView tab). + sample: None until the first successful read. + """ + wv = _state["wireview"] + return { + "available": wv["device"] is not None, + "connected": wv["connected"], + "info": wv["info"], + "sample": wv["last_sample"], + } + + # ── Write endpoints ──────────────────────────────────────────────────────────── @@ -1832,6 +1999,51 @@ async def ws_curve(ws: WebSocket): g_state["curve_clients"].discard(ws) +@app.websocket("/ws/wireview") +async def ws_wireview(ws: WebSocket): + """Stream WireView Pro II sensor samples at poll_interval_s. + + Messages: + {"type": "unavailable"} — no device connected + {"type": "sample", "info": {...}, "sample": {...}} — new reading + """ + if not _ws_authenticated(ws): + await ws.close(code=1008) + return + await ws.accept() + try: + data = await ws.receive_json() + if data.get("action") != "subscribe": + await ws.close() + return + except WebSocketDisconnect: + return + except Exception: + await ws.close() + return + + wv = _state["wireview"] + wv["clients"].add(ws) + try: + # Send the current state immediately so the client does not have to + # wait for the next poll tick. + if wv["device"] is not None and wv["last_sample"] is not None: + await ws.send_json( + {"type": "sample", "info": wv["info"], "sample": wv["last_sample"]} + ) + else: + await ws.send_json({"type": "unavailable"}) + + while True: + await ws.receive_text() + except WebSocketDisconnect: + pass + except Exception: + log.debug("wireview ws client error", exc_info=True) + finally: + wv["clients"].discard(ws) + + # ── Frontend SPA ────────────────────────────────────────────────────────────── diff --git a/nvcurve/wireview.py b/nvcurve/wireview.py new file mode 100644 index 0000000..2be0c22 --- /dev/null +++ b/nvcurve/wireview.py @@ -0,0 +1,682 @@ +"""Native reader for the Thermal Grizzly WireView Pro II. + +Talks to the 12 VHPWR connector monitor directly over its USB CDC/ACM +serial port (STM32, VID 0483 / PID 5740) — no exporter, no GUI, no kernel +module required. When the wireview-hwmon kernel module is loaded, the +sysfs node is used instead (the wireviewd daemon owns the port in that +case, so direct serial would corrupt frames). + +Protocol notes (matches the firmware's DEVICE_STR_LEN=32 layout): + * No framing/CRC — a desynced read corrupts arbitrary fields for one + poll. Real frames always carry zero padding bytes and a fan duty + <= 100; anything else is discarded and the next poll realigns. + * The firmware occasionally stops answering the RTS welcome handshake + (observed after USB state changes) while still answering every data + command, so identification falls back to the vendor-data reply. +""" + +import logging +import os +import struct +import time +from typing import Any + +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) +USB_PRODUCT_ID = "5740" # WireView Pro II normal mode +WELCOME_MESSAGE = "Thermal Grizzly WireView Pro II" +MAX_WELCOME_LENGTH = 64 + +# Vendor/product ids reported by CMD_READ_VENDOR_DATA (not the USB ids). +VENDOR_ID_THERMAL_GRIZZLY = 0xEF +PRODUCT_ID_PRO2 = 0x05 +PRODUCT_ID_PRO2_NOCTUA = 0x06 + +DEVICE_NAMES = { + PRODUCT_ID_PRO2: "WireView Pro II", + PRODUCT_ID_PRO2_NOCTUA: "WireView Pro II Noctua Edition", +} + +BAUD_RATE = 115200 +READ_TIMEOUT_S = 1.0 + +# ── Serial protocol commands ────────────────────────────────────────────────── + +CMD_READ_VENDOR_DATA = 0x01 +CMD_READ_UID = 0x02 +CMD_READ_SENSOR_VALUES = 0x04 +CMD_READ_CONFIG = 0x05 +CMD_SCREEN_CHANGE = 0x0C +CMD_READ_BUILD_INFO = 0x0D + +SCREEN_RESUME_UPDATES = 0xF1 + +# ── Wire layout ─────────────────────────────────────────────────────────────── +# SensorStruct (100 bytes, little-endian, pack=4): +# 4x int16 temperatures (0.1 °C): in, out, ext1, ext2 +# uint16 Vdd (mV) +# uint8 fan duty (%) +# pad +# 6x { int16 voltage (mV), pad, uint32 current (mA), uint32 power (mW) } +# uint32 total power (mW) +# uint32 total current (mA) +# uint16 avg voltage (mV) +# uint8 PSU capability (0=600W, 1=450W, 2=300W, 3=150W) +# pad +# uint16 fault status mask +# uint16 fault log mask +SENSOR_STRUCT = struct.Struct( + "<4hHBx" + "".join("hxxII" for _ in range(6)) + "IIHBxHH" +) +SENSOR_STRUCT_SIZE = SENSOR_STRUCT.size # 100 + +# BuildStruct: VendorData(3) + ProductName(32) + BuildInfo(32) + NameLength(1) +BUILD_STRUCT_SIZE = 3 + 32 + 32 + 1 +BUILD_INFO_OFFSET = 3 + 32 # 35 + +PSU_CAPABILITY_W = {0: 600, 1: 450, 2: 300, 3: 150} + +# Fault bitmask (both the active status and the latched log use these bits). +FAULT_BITS = { + 0: "Chip over-temperature", + 1: "Sensor over-temperature", + 2: "Over-current (OCP)", + 3: "Wire over-current", + 4: "Over-power (OPP)", + 5: "Current imbalance", +} + + +def decode_faults(mask: int) -> list[str]: + """Human-readable names of the active fault bits in a status/log mask.""" + return [ + FAULT_BITS[bit] + for bit in sorted(FAULT_BITS) + if mask & (1 << bit) + ] + + +def is_supported_product(vendor_id: int, product_id: int) -> bool: + """True for the products the Pro II protocol serves (5 and 6).""" + return ( + vendor_id == VENDOR_ID_THERMAL_GRIZZLY + and product_id in (PRODUCT_ID_PRO2, PRODUCT_ID_PRO2_NOCTUA) + ) + + +def parse_sensor_struct(buf: bytes) -> dict: + """Decode a 100-byte sensor frame into a JSON-serializable sample. + + Totals are computed from the per-pin readings (voltage * current), + matching the exporter's output; the device's own total fields are + not used. + """ + fields = SENSOR_STRUCT.unpack(buf) + ts_in, ts_out, ts_ext1, ts_ext2, _vdd, fan_duty = fields[:6] + pin_fields = fields[6:24] + ( + _total_power, + _total_current, + _avg_voltage, + psu_cap, + fault_status, + fault_log, + ) = fields[24:] + + pins = [] + power_total = 0.0 + current_total = 0.0 + for i in range(6): + voltage_v = pin_fields[i * 3] / 1000.0 + current_a = pin_fields[i * 3 + 1] / 1000.0 + power_w = pin_fields[i * 3 + 2] / 1000.0 + pins.append( + { + "voltage_v": round(voltage_v, 3), + "current_a": round(current_a, 3), + "power_w": round(power_w, 3), + } + ) + power_total += voltage_v * current_a + current_total += current_a + + return { + "timestamp": time.time(), + "power_total_w": round(power_total, 3), + "current_total_a": round(current_total, 3), + "voltage_avg_v": ( + round(power_total / current_total, 3) if current_total > 0 else 0.0 + ), + "pins": pins, + "temp_in_c": ts_in / 10.0, + "temp_out_c": ts_out / 10.0, + "temp_ext1_c": ts_ext1 / 10.0, + "temp_ext2_c": ts_ext2 / 10.0, + "fan_duty_pct": fan_duty, + "fault_status": fault_status, + "fault_log": fault_log, + "psu_capability_w": PSU_CAPABILITY_W.get(psu_cap, 0), + } + + +def sensor_frame_is_corrupt(buf: bytes) -> bool: + """Corruption check for a sensor frame: real frames always carry zero + padding bytes and a fan duty <= 100. Wire layout: fan duty at offset + 10, pad1 at 11, pad2 five bytes from the end (before the two 16-bit + fault masks).""" + if len(buf) < SENSOR_STRUCT_SIZE: + return True + return buf[10] > 100 or buf[11] != 0 or buf[SENSOR_STRUCT_SIZE - 5] != 0 + + +# ── Device discovery ────────────────────────────────────────────────────────── + + +def _sysfs_matches_wireview(tty_sysfs_dir: str) -> bool: + """Walk up from a resolved tty sysfs path to the USB device and check + its idVendor/idProduct.""" + path = tty_sysfs_dir + while path and path != "/": + vid_file = os.path.join(path, "idVendor") + pid_file = os.path.join(path, "idProduct") + if os.path.isfile(vid_file) and os.path.isfile(pid_file): + try: + with open(vid_file) as f: + vid = f.read().strip().lower() + with open(pid_file) as f: + pid = f.read().strip().lower() + except OSError: + return False + return vid == USB_VENDOR_ID and pid == USB_PRODUCT_ID + path = os.path.dirname(path) + return False + + +def find_wireview_ports() -> list[str]: + """Find /dev nodes of connected WireView Pro II devices. + + Checks the stable /dev/wireview-pro2 symlink (created by the + 99-wireview.rules udev rule) and falls back to a sysfs scan of all + ttyACM* ports matched by USB VID/PID. + """ + ports: set[str] = set() + + link = "/dev/wireview-pro2" + if os.path.islink(link) or os.path.exists(link): + try: + target = os.path.realpath(link) + if os.path.exists(target): + ports.add(target) + except OSError: + pass + + sys_class = "/sys/class/tty" + if os.path.isdir(sys_class): + try: + entries = os.listdir(sys_class) + except OSError: + entries = [] + for entry in entries: + if not entry.startswith("ttyACM"): + continue + tty_dir = os.path.join(sys_class, entry) + try: + resolved = os.path.realpath(tty_dir) + except OSError: + continue + if _sysfs_matches_wireview(resolved): + ports.add(f"/dev/{entry}") + + return sorted(ports) + + +def find_hwmon_path() -> str | None: + """Find the wireview-hwmon sysfs node, if the kernel module is loaded.""" + base = "/sys/class/hwmon" + if not os.path.isdir(base): + return None + try: + entries = os.listdir(base) + except OSError: + return None + for entry in entries: + name_path = os.path.join(base, entry, "name") + try: + with open(name_path) as f: + if f.read().strip().lower() == "wireview": + return os.path.join(base, entry) + except OSError: + continue + return None + + +# ── Serial transport ────────────────────────────────────────────────────────── + + +class WireViewSerialDevice: + """Direct serial access to a WireView Pro II. + + The port is opened and closed per transaction (open → flush → write → + read → close), matching the proven behavior of the exporter: it keeps + the port unheld between polls so other tools (official GUI, wireviewd) + can share the device, and a fresh open realigns a desynced stream. + """ + + def __init__(self, port: str, baud: int = BAUD_RATE) -> None: + self._port = port + self._baud = baud + self._connected = False + self._rejected = False + self._vendor_id = 0 + self._product_id = 0 + self._firmware_version = "" + self._uid = "" + self._build = "" + self._config_version = -1 + + # ── Identity ── + + @property + def connected(self) -> bool: + return self._connected + + @property + def rejected(self) -> bool: + """True when the device reported an unsupported product id. Callers + can memoize this so the port is not re-probed (and re-logged) on + every watchdog tick.""" + return self._rejected + + @property + def transport(self) -> str: + return "serial" + + @property + def port(self) -> str: + return self._port + + def info(self) -> dict: + return { + "device_name": DEVICE_NAMES.get( + self._product_id, "WireView Pro II" + ), + "hw_rev": f"{self._vendor_id:02X}{self._product_id:02X}", + "firmware_version": self._firmware_version, + "uid": self._uid, + "build": self._build, + "transport": self.transport, + "port": self._port, + } + + def node_exists(self) -> bool: + """Whether the serial device node still exists (unplug check).""" + return os.path.exists(self._port) + + # ── Connection ── + + def connect(self) -> bool: + """Identify the device and prepare it for sensor reads. + + The welcome handshake (RTS edge) is the primary identification, + but the firmware occasionally stops answering it while still + answering every command — a supported vendor-data reply is + equally conclusive, so accept either. + """ + if self._connected: + return True + + # The welcome handshake (RTS edge) is the primary identification, but + # the firmware occasionally stops answering it while still answering + # every command — the supported vendor-data reply below is equally + # conclusive, so the welcome is read for logging only. + welcome = self._read_welcome() + if welcome and welcome != WELCOME_MESSAGE: + log.debug( + "WireView: unexpected welcome string %r on %s", + welcome, + self._port, + ) + + vd = self._transaction(bytes([CMD_READ_VENDOR_DATA]), 3) + if vd is None or len(vd) < 3: + return False + vendor, product, fw = vd[0], vd[1], vd[2] + if not is_supported_product(vendor, product): + self._rejected = True + log.info( + "WireView: unsupported product %02X%02X on %s, skipped", + vendor, + product, + self._port, + ) + return False + + self._vendor_id = vendor + self._product_id = product + self._firmware_version = str(fw) + + cfg = self._transaction(bytes([CMD_READ_CONFIG]), 4) + if cfg is None or len(cfg) < 3: + return False + self._config_version = cfg[2] + + uid = self._transaction(bytes([CMD_READ_UID]), 12) + if uid is not None and len(uid) == 12: + self._uid = uid.hex().upper() + + # Enable display updates just in case. + self._transaction( + bytes([CMD_SCREEN_CHANGE, SCREEN_RESUME_UPDATES]), 0 + ) + + build = self._transaction(bytes([CMD_READ_BUILD_INFO]), BUILD_STRUCT_SIZE) + if build is not None and len(build) >= BUILD_INFO_OFFSET + 1: + self._build = ( + build[BUILD_INFO_OFFSET : BUILD_INFO_OFFSET + 32] + .split(b"\x00")[0] + .decode("ascii", errors="replace") + ) + + self._connected = True + log.info( + "WireView connected on %s (%s, fw %s)", + self._port, + self.info()["hw_rev"], + self._firmware_version, + ) + return True + + def close(self) -> None: + self._connected = False + + # ── Sensor reads ── + + def read_sample(self) -> dict | None: + """Read one sensor sample, or None when the device is unresponsive + or the frame is corrupt.""" + if not self._connected: + return None + buf = self._transaction(bytes([CMD_READ_SENSOR_VALUES]), SENSOR_STRUCT_SIZE) + if buf is None or sensor_frame_is_corrupt(buf): + return None + return parse_sensor_struct(buf) + + # ── 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.""" + ser = self._open_port() + if ser is None: + return None + try: + ser.reset_input_buffer() + ser.rts = False + time.sleep(0.01) + ser.rts = True + time.sleep(0.01) + buf = bytearray() + deadline = time.monotonic() + READ_TIMEOUT_S + while len(buf) < MAX_WELCOME_LENGTH: + remaining = deadline - time.monotonic() + if remaining <= 0: + break + ser.timeout = min(READ_TIMEOUT_S, remaining) + chunk = ser.read(MAX_WELCOME_LENGTH - len(buf)) + if not chunk: + break + buf.extend(chunk) + if b"\x00" in chunk: + break + time.sleep(0.01) + ser.rts = False + nul = buf.find(b"\x00") + if nul >= 0: + return bytes(buf[:nul]).decode("ascii", errors="replace") + return None + except OSError: # SerialException is an OSError + return None + finally: + ser.close() + + def _transaction(self, cmd: bytes, response_size: int) -> bytes | None: + """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.""" + ser = self._open_port() + if ser is None: + return None + try: + ser.reset_input_buffer() + if cmd: + ser.write(cmd) + if response_size == 0: + return b"" + return self._read_exact(ser, response_size) + except OSError: # SerialException is an OSError + return None + finally: + ser.close() + + @staticmethod + 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 + while len(buf) < size: + remaining = deadline - time.monotonic() + if remaining <= 0: + return None + ser.timeout = min(READ_TIMEOUT_S, remaining) + chunk = ser.read(size - len(buf)) + if not chunk: + return None + buf.extend(chunk) + return bytes(buf) + + +# ── hwmon (sysfs) transport ─────────────────────────────────────────────────── + + +class WireViewHwmonDevice: + """Reads the wireview-hwmon sysfs node (kernel module + wireviewd). + + Used when the module is loaded: the daemon owns the serial port in + that case, so direct serial would corrupt frames. + """ + + def __init__(self, hwmon_path: str) -> None: + self._path = hwmon_path + self._connected = False + + @property + def connected(self) -> bool: + return self._connected + + @property + def rejected(self) -> bool: + return False + + @property + def transport(self) -> str: + return "hwmon" + + @property + def port(self) -> str: + return self._path + + def info(self) -> dict: + return { + "device_name": "WireView Pro II", + "hw_rev": "", + "firmware_version": "", + "uid": "", + "build": "", + "transport": self.transport, + "port": self._path, + } + + def node_exists(self) -> bool: + return os.path.isdir(self._path) + + def connect(self) -> bool: + if self._connected: + return True + name_path = os.path.join(self._path, "name") + try: + with open(name_path) as f: + if f.read().strip().lower() != "wireview": + return False + # Probe that the node actually serves data. + with open(os.path.join(self._path, "in0_input")) as f: + f.read().strip() + except OSError: + return False + self._connected = True + log.info("WireView connected via hwmon (%s)", self._path) + return True + + def close(self) -> None: + self._connected = False + + def read_sample(self) -> dict | None: + if not self._connected: + return None + try: + pin_voltage = [ + self._read_int(f"in{i}_input") / 1000.0 for i in range(6) + ] + pin_current = [ + self._read_int(f"curr{i + 1}_input") / 1000.0 for i in range(6) + ] + temp_in = self._read_temp("temp1_input") + temp_out = self._read_temp("temp2_input") + temp_ext1 = self._read_temp("temp3_input") + temp_ext2 = self._read_temp("temp4_input") + + fault_status = self._read_int_or("fault_status_raw") + if fault_status is None: + fault_status = 0xFFFF if self._read_int("intrusion0_alarm") else 0 + fault_log = self._read_int_or("fault_log_raw") + if fault_log is None: + fault_log = 0xFFFF if self._read_int("intrusion1_alarm") else 0 + + psu_cap_uw = self._read_int_or("power1_cap") + if psu_cap_uw is not None: + psu_capability = int(round(psu_cap_uw / 1_000_000.0)) + else: + psu_cap = self._read_int_or("psu_cap") + psu_capability = PSU_CAPABILITY_W.get(psu_cap or 0, 0) + + pwm = self._read_int_or("pwm1") + if pwm is not None: + fan_duty = int(round(min(255, max(0, pwm)) * 100 / 255.0)) + else: + fan_duty = self._read_int("fan1_input") + + power_total = sum( + v * i + for v, i in zip(pin_voltage, pin_current, strict=True) + ) + current_total = sum(pin_current) + + return { + "timestamp": time.time(), + "power_total_w": round(power_total, 3), + "current_total_a": round(current_total, 3), + "voltage_avg_v": ( + round(power_total / current_total, 3) + if current_total > 0 + else 0.0 + ), + "pins": [ + { + "voltage_v": round(v, 3), + "current_a": round(i, 3), + "power_w": round(v * i, 3), + } + for v, i in zip(pin_voltage, pin_current, strict=True) + ], + "temp_in_c": temp_in, + "temp_out_c": temp_out, + "temp_ext1_c": temp_ext1, + "temp_ext2_c": temp_ext2, + "fan_duty_pct": fan_duty, + "fault_status": fault_status, + "fault_log": fault_log, + "psu_capability_w": psu_capability, + } + except OSError: + return None + + def _read_int(self, filename: str) -> int: + """Read an integer sysfs attribute; 0 when missing or unreadable + (matches the exporter's ReadIntFile).""" + try: + with open(os.path.join(self._path, filename)) as f: + return int(f.read().strip()) + except (OSError, ValueError): + return 0 + + def _read_int_or(self, filename: str) -> int | None: + try: + with open(os.path.join(self._path, filename)) as f: + return int(f.read().strip()) + except (OSError, ValueError): + return None + + def _read_temp(self, filename: str) -> float | None: + """Read a temperature sysfs attribute (m°C); None when missing or + unreadable. NaN would poison the whole sample: the WebSocket + serializer emits a bare NaN token (invalid JSON) and the REST + JSONResponse rejects it with a 500.""" + try: + with open(os.path.join(self._path, filename)) as f: + return int(f.read().strip()) / 1000.0 + except (OSError, ValueError): + return None + + +def create_device() -> WireViewSerialDevice | WireViewHwmonDevice | None: + """Create a device for the first available WireView, or None. + + Preference: the wireview-hwmon sysfs node (the wireviewd daemon owns + the serial port in that case), otherwise direct serial on the first + matching /dev/ttyACM*. + """ + hwmon_path = find_hwmon_path() + if hwmon_path: + return WireViewHwmonDevice(hwmon_path) + ports = find_wireview_ports() + if ports: + return WireViewSerialDevice(ports[0]) + return None diff --git a/pyproject.toml b/pyproject.toml index 6a0d03e..c3afc02 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -14,6 +14,7 @@ dependencies = [ "pydantic>=2.0", "httpx>=0.27", "bcrypt>=4.0", + "pyserial>=3.5", ] [project.scripts] diff --git a/tests/test_wireview.py b/tests/test_wireview.py new file mode 100644 index 0000000..2f2a8fb --- /dev/null +++ b/tests/test_wireview.py @@ -0,0 +1,528 @@ +"""Tests for the native WireView Pro II reader (nvcurve/wireview.py). + +Standalone (no pytest required): + + python tests/test_wireview.py + +Also works under pytest. Covers the sensor-frame parser, corruption +detection, fault decoding, USB port discovery, and the hwmon (sysfs) +transport. +""" + +import builtins +import io +import os +import struct +import sys +import tempfile +from unittest import mock + +sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..")) + +from nvcurve import wireview as wv # noqa: E402 + +PASS = 0 +FAIL = 0 + + +def check(name: str, cond: bool) -> None: + global PASS, FAIL + if cond: + PASS += 1 + print(f" PASS {name}") + else: + FAIL += 1 + print(f" FAIL {name}") + + +def make_frame( + ts=(392, 352, 273, 395), + vdd=11900, + fan=0, + pins=((11904, 6221, 74055),) * 6, + total_power=450568, + total_current=37863, + avg_voltage=11900, + psu_cap=0, + fault_status=0, + fault_log=0, + pad1=0, + pad2=0, +) -> bytes: + """Build a 100-byte sensor frame with controllable fields.""" + frame = struct.pack("<4hHB", *ts, vdd, fan) + frame += bytes([pad1]) + for voltage, current, power in pins: + frame += struct.pack(" {watts}W", s["psu_capability_w"] == watts) + + +def test_parse_negative_temps(): + s = wv.parse_sensor_struct(make_frame(ts=(-10, 0, 555, 999))) + check("negative temp", abs(s["temp_in_c"] - (-1.0)) < 1e-9) + check("zero temp", s["temp_out_c"] == 0.0) + check("high temp", abs(s["temp_ext2_c"] - 99.9) < 1e-9) + + +# ── Corruption detection ────────────────────────────────────────────────────── + + +def test_corruption_check(): + check("clean frame accepted", not wv.sensor_frame_is_corrupt(make_frame())) + check( + "fan duty > 100 rejected", + wv.sensor_frame_is_corrupt(make_frame(fan=101)), + ) + check( + "fan duty 100 accepted", + not wv.sensor_frame_is_corrupt(make_frame(fan=100)), + ) + check("pad1 dirty rejected", wv.sensor_frame_is_corrupt(make_frame(pad1=1))) + check("pad2 dirty rejected", wv.sensor_frame_is_corrupt(make_frame(pad2=1))) + check( + "short frame rejected", + wv.sensor_frame_is_corrupt(make_frame()[:50]), + ) + + +# ── Fault decoding ──────────────────────────────────────────────────────────── + + +def test_decode_faults(): + check("no faults", wv.decode_faults(0) == []) + check( + "chip over-temp", + wv.decode_faults(1) == ["Chip over-temperature"], + ) + check( + "over-power", + wv.decode_faults(16) == ["Over-power (OPP)"], + ) + check( + "multiple faults", + wv.decode_faults(1 | 32) + == ["Chip over-temperature", "Current imbalance"], + ) + check( + "unknown bits ignored", + wv.decode_faults(1 | 0x80) == ["Chip over-temperature"], + ) + + +def test_product_support(): + check("Pro II supported", wv.is_supported_product(0xEF, 0x05)) + check("Noctua Edition supported", wv.is_supported_product(0xEF, 0x06)) + check("WireView II unsupported", not wv.is_supported_product(0xEF, 0x07)) + check("other vendor unsupported", not wv.is_supported_product(0x12, 0x05)) + + +# ── USB port discovery ──────────────────────────────────────────────────────── + + +def test_find_wireview_ports(): + """Fake a sysfs tree with one WireView (ttyACM0) and one other CDC + device (ttyACM1), plus the udev symlink.""" + contents = { + "/sys/devices/fake/usb0/idVendor": "0483\n", + "/sys/devices/fake/usb0/idProduct": "5740\n", + "/sys/devices/fake/usb1/idVendor": "1234\n", + "/sys/devices/fake/usb1/idProduct": "5678\n", + } + + def fake_islink(p): + return p == "/dev/wireview-pro2" + + def fake_realpath(p): + return { + "/dev/wireview-pro2": "/dev/ttyACM0", + "/sys/class/tty/ttyACM0": "/sys/devices/fake/usb0", + "/sys/class/tty/ttyACM1": "/sys/devices/fake/usb1", + }.get(p, p) + + def fake_isdir(p): + return p == "/sys/class/tty" + + def fake_listdir(p): + return ["ttyACM0", "ttyACM1"] if p == "/sys/class/tty" else [] + + def fake_isfile(p): + return p in contents + + def fake_exists(p): + return p == "/dev/ttyACM0" + + def fake_open(p, *a, **k): + if p in contents: + return io.StringIO(contents[p]) + return real_open(p, *a, **k) + + real_open = open + with mock.patch.object(wv.os.path, "islink", fake_islink), mock.patch.object( + wv.os.path, "realpath", fake_realpath + ), mock.patch.object(wv.os.path, "isdir", fake_isdir), mock.patch.object( + wv.os, "listdir", fake_listdir + ), mock.patch.object(wv.os.path, "isfile", fake_isfile), mock.patch.object( + wv.os.path, "exists", fake_exists + ), mock.patch.object( + wv.os.path, "dirname", os.path.dirname + ), mock.patch.object( + builtins, "open", fake_open + ): + ports = wv.find_wireview_ports() + check("exactly one port found", ports == ["/dev/ttyACM0"]) + + +def test_find_wireview_ports_none(): + with mock.patch.object(wv.os.path, "islink", lambda p: False), mock.patch.object( + wv.os.path, "isdir", lambda p: False + ), mock.patch.object(wv.os.path, "exists", lambda p: False): + check("no ports", wv.find_wireview_ports() == []) + + +def test_find_hwmon_path(): + with tempfile.TemporaryDirectory() as tmp: + # A non-wireview hwmon and a wireview one. + for name, dev in (("hwmon0", "coretemp"), ("hwmon1", "wireview")): + d = os.path.join(tmp, name) + os.makedirs(d) + with open(os.path.join(d, "name"), "w") as f: + f.write(dev + "\n") + + real_join = os.path.join + real_listdir = os.listdir + + def fake_join(*parts): + if parts and parts[0] == "/sys/class/hwmon": + parts = (tmp,) + parts[1:] + return real_join(*parts) + + def fake_listdir(p): + if p == "/sys/class/hwmon": + return real_listdir(tmp) + return real_listdir(p) + + with mock.patch.object(wv.os.path, "join", fake_join), mock.patch.object( + wv.os, "listdir", fake_listdir + ): + found = wv.find_hwmon_path() + check("hwmon path found", found == os.path.join(tmp, "hwmon1")) + + with mock.patch.object(wv.os.path, "isdir", lambda p: False): + check("hwmon absent", wv.find_hwmon_path() is None) + + +# ── hwmon transport ─────────────────────────────────────────────────────────── + + +def test_hwmon_device(): + with tempfile.TemporaryDirectory() as tmp: + p = os.path.join(tmp, "hwmon2") + os.makedirs(p) + with open(os.path.join(p, "name"), "w") as f: + f.write("wireview\n") + for i in range(6): + with open(os.path.join(p, f"in{i}_input"), "w") as f: + f.write("11904\n") + with open(os.path.join(p, f"curr{i + 1}_input"), "w") as f: + f.write("6221\n") + for name, val in ( + ("temp1_input", "39200"), + ("temp2_input", "35200"), + ("temp3_input", "27300"), + ("temp4_input", "39500"), + ("fault_status_raw", "0"), + ("fault_log_raw", "0"), + ("power1_cap", "600000000"), + ("pwm1", "128"), + ): + with open(os.path.join(p, name), "w") as f: + f.write(val + "\n") + + dev = wv.WireViewHwmonDevice(p) + check("hwmon connect", dev.connect()) + check("hwmon node exists", dev.node_exists()) + s = dev.read_sample() + check("hwmon sample", s is not None) + if s: + check("hwmon temp in", abs(s["temp_in_c"] - 39.2) < 1e-9) + check("hwmon fan duty ~50%", 49 <= s["fan_duty_pct"] <= 51) + check("hwmon psu cap", s["psu_capability_w"] == 600) + check("hwmon 6 pins", len(s["pins"]) == 6) + check( + "hwmon total current", + abs(s["current_total_a"] - round(6 * 6.221, 3)) < 1e-9, + ) + dev.close() + check("hwmon closed", not dev.connected) + + # A missing temp channel must yield null (not NaN): NaN would break + # the WebSocket JSON (bare NaN token) and 500 the REST endpoint. + os.remove(os.path.join(p, "temp3_input")) + dev3 = wv.WireViewHwmonDevice(p) + dev3.connect() + s3 = dev3.read_sample() + check("hwmon missing temp sample", s3 is not None) + if s3: + check("hwmon missing temp is None", s3["temp_ext1_c"] is None) + check("hwmon missing temp present", s3["temp_in_c"] == 39.2) + import json + + check( + "hwmon sample JSON-safe", + json.dumps(s3, allow_nan=False) is not None, + ) + dev3.close() + + # Wrong device name is rejected. + with open(os.path.join(p, "name"), "w") as f: + f.write("coretemp\n") + dev2 = wv.WireViewHwmonDevice(p) + check("wrong name rejected", not dev2.connect()) + + +def test_hwmon_legacy_attrs(): + """Older modules expose psu_cap/fan1_input instead of power1_cap/pwm1.""" + with tempfile.TemporaryDirectory() as tmp: + p = os.path.join(tmp, "hwmon3") + os.makedirs(p) + with open(os.path.join(p, "name"), "w") as f: + f.write("wireview\n") + for i in range(6): + with open(os.path.join(p, f"in{i}_input"), "w") as f: + f.write("12000\n") + with open(os.path.join(p, f"curr{i + 1}_input"), "w") as f: + f.write("1000\n") + for name, val in ( + ("temp1_input", "30000"), + ("temp2_input", "30000"), + ("temp3_input", "30000"), + ("temp4_input", "30000"), + ("psu_cap", "2"), + ("fan1_input", "42"), + ("intrusion0_alarm", "0"), + ("intrusion1_alarm", "0"), + ): + with open(os.path.join(p, name), "w") as f: + f.write(val + "\n") + + dev = wv.WireViewHwmonDevice(p) + check("legacy connect", dev.connect()) + s = dev.read_sample() + check("legacy sample", s is not None) + if s: + check("legacy psu cap 300W", s["psu_capability_w"] == 300) + check("legacy fan pct", s["fan_duty_pct"] == 42) + + +# ── Serial transport (pty-based fake device) ────────────────────────────────── + + +def _build_info_struct(product_name: str, build_info: str) -> bytes: + """BuildStruct: VendorData(3) + ProductName(32) + BuildInfo(32) + NameLength(1).""" + return ( + bytes([0xEF, 0x05, 1]) + + product_name.encode().ljust(32, b"\x00") + + build_info.encode().ljust(32, b"\x00") + + bytes([len(product_name)]) + ) + + +def _pty_fake_device(responses: dict): + """Create a pty pair whose master side answers protocol commands in a + background thread. Returns (slave_path, master_fd, thread). + + The fake never sends the welcome string, so the device exercises its + documented fallback: identification via the vendor-data reply. + """ + import pty + import threading + + master, slave = pty.openpty() + slave_path = os.ttyname(slave) + + def run(): + while True: + try: + cmd = os.read(master, 1) + except OSError: + return + if not cmd: + return + resp = responses.get(cmd[0]) + if resp: + try: + os.write(master, resp) + except OSError: + return + + t = threading.Thread(target=run, daemon=True) + t.start() + return slave_path, master, t + + +def test_serial_device_protocol(): + """Full connect handshake + sensor read against a fake device.""" + uid = bytes.fromhex("A7003100015045324B383120") + responses = { + wv.CMD_READ_VENDOR_DATA: bytes([0xEF, 0x05, 1]), + wv.CMD_READ_CONFIG: bytes([0, 0, 1, 0]), # version at offset 2 + wv.CMD_READ_UID: uid, + wv.CMD_READ_BUILD_INFO: _build_info_struct( + "WireView Pro II", "TG-WV-PRO2-FW_20251211_1547" + ), + wv.CMD_READ_SENSOR_VALUES: make_frame(), + # CMD_SCREEN_CHANGE expects no response. + } + slave_path, master, t = _pty_fake_device(responses) + try: + dev = wv.WireViewSerialDevice(slave_path) + check("serial connect", dev.connect()) + check("serial not rejected", not dev.rejected) + info = dev.info() + check("serial device name", info["device_name"] == "WireView Pro II") + check("serial hw_rev", info["hw_rev"] == "EF05") + check("serial firmware", info["firmware_version"] == "1") + check("serial uid", info["uid"] == "A7003100015045324B383120") + check("serial build", info["build"] == "TG-WV-PRO2-FW_20251211_1547") + check("serial transport", info["transport"] == "serial") + + s = dev.read_sample() + check("serial sample", s is not None) + if s: + check("serial sample temp", abs(s["temp_in_c"] - 39.2) < 1e-9) + check("serial sample pins", len(s["pins"]) == 6) + + dev.close() + check("serial closed", not dev.connected) + finally: + os.close(master) + t.join(timeout=2) + + +def test_serial_device_rejected_product(): + """An unsupported product id is rejected and memoized via .rejected.""" + responses = {wv.CMD_READ_VENDOR_DATA: bytes([0xEF, 0x07, 1])} + slave_path, master, t = _pty_fake_device(responses) + try: + dev = wv.WireViewSerialDevice(slave_path) + check("serial reject connect", not dev.connect()) + check("serial rejected flag", dev.rejected) + finally: + os.close(master) + t.join(timeout=2) + + +def test_serial_device_no_response(): + """A silent port fails the connect (vendor-data read times out).""" + slave_path, master, t = _pty_fake_device({}) + try: + dev = wv.WireViewSerialDevice(slave_path) + check("serial no-response connect", not dev.connect()) + check("serial no-response not rejected", not dev.rejected) + check("serial no-response sample", dev.read_sample() is None) + finally: + os.close(master) + 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() + test_parse_psu_capabilities() + test_parse_negative_temps() + test_corruption_check() + test_decode_faults() + test_product_support() + test_find_wireview_ports() + test_find_wireview_ports_none() + test_find_hwmon_path() + test_hwmon_device() + test_hwmon_legacy_attrs() + 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 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/uv.lock b/uv.lock index 30d4f5b..2d16afb 100644 --- a/uv.lock +++ b/uv.lock @@ -246,6 +246,7 @@ dependencies = [ { name = "httpx" }, { name = "nvidia-ml-py" }, { name = "pydantic" }, + { name = "pyserial" }, { name = "uvicorn", extra = ["standard"] }, ] @@ -261,6 +262,7 @@ requires-dist = [ { name = "httpx", specifier = ">=0.27" }, { name = "nvidia-ml-py", specifier = ">=12.0" }, { name = "pydantic", specifier = ">=2.0" }, + { name = "pyserial", specifier = ">=3.5" }, { name = "uvicorn", extras = ["standard"], specifier = ">=0.30" }, ] @@ -389,6 +391,15 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/f7/07/34573da085946b6a313d7c42f82f16e8920bfd730665de2d11c0c37a74b5/pydantic_core-2.41.5-graalpy312-graalpy250_312_native-manylinux_2_17_x86_64.manylinux2014_x86_64.whl", hash = "sha256:76d0819de158cd855d1cbb8fcafdf6f5cf1eb8e470abe056d5d161106e38062b", size = 2139017, upload-time = "2025-11-04T13:42:59.471Z" }, ] +[[package]] +name = "pyserial" +version = "3.5" +source = { registry = "https://pypi.org/simple" } +sdist = { url = "https://files.pythonhosted.org/packages/1e/7d/ae3f0a63f41e4d2f6cb66a5b57197850f919f59e558159a4dd3a818f5082/pyserial-3.5.tar.gz", hash = "sha256:3c77e014170dfffbd816e6ffc205e9842efb10be9f58ec16d3e8675b4925cddb", size = 159125, upload-time = "2020-11-23T03:59:15.045Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/07/bc/587a445451b253b285629263eb51c2d8e9bcea4fc97826266d186f96f558/pyserial-3.5-py2.py3-none-any.whl", hash = "sha256:c4451db6ba391ca6ca299fb3ec7bae67a5c55dde170964c7a14ceefec02f2cf0", size = 90585, upload-time = "2020-11-23T03:59:13.41Z" }, +] + [[package]] name = "python-dotenv" version = "1.2.2"