ARIA
526b71b45f
Standalone support for the ThermalGrizzly WireView Pro II without depending on the external wireview_reporter exporter: - wireview.py: serial protocol (STX/ETX + 16-bit CRC16-CCITT) with vendor-data product identification, config version, UID, build string, screen layout, and temperature/power/current sensor reads; hwmon fallback with per-channel index resolution; udev-based detection (vendor 0x2560 / product 0x0101) with fallback port probing; serial read timeout and watchdog reconnect - server.py: startup detection (root only), 1 Hz poller gated on subscribed clients, WS 'wireview' channel, GET /api/wireview, rejected-product memoization, stale-read race guard - frontend: WireView tab (visible when a device is detected) with live temperature/power/current cards, sparkline history, and device info; nullable temp channels - tests: 76 tests covering parser, CRC, fault classification, hwmon resolution, JSON safety, and the serial transport against a pty-based fake device Verified live: tab appears with the device connected, disappears when unplugged, reconnects on re-plug, no serial traffic when idle.
494 lines
17 KiB
Python
494 lines
17 KiB
Python
"""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("<hxxII", voltage, current, power)
|
|
frame += struct.pack("<IIHB", total_power, total_current, avg_voltage, psu_cap)
|
|
frame += bytes([pad2])
|
|
frame += struct.pack("<HH", fault_status, fault_log)
|
|
assert len(frame) == wv.SENSOR_STRUCT_SIZE
|
|
return frame
|
|
|
|
|
|
# ── Sensor frame parser ───────────────────────────────────────────────────────
|
|
|
|
|
|
def test_parse_sensor_struct():
|
|
s = wv.parse_sensor_struct(make_frame())
|
|
check("temp in", abs(s["temp_in_c"] - 39.2) < 1e-9)
|
|
check("temp out", abs(s["temp_out_c"] - 35.2) < 1e-9)
|
|
check("temp ext1", abs(s["temp_ext1_c"] - 27.3) < 1e-9)
|
|
check("temp ext2", abs(s["temp_ext2_c"] - 39.5) < 1e-9)
|
|
check("fan duty", s["fan_duty_pct"] == 0)
|
|
check("psu capability", s["psu_capability_w"] == 600)
|
|
check("fault status", s["fault_status"] == 0)
|
|
check("fault log", s["fault_log"] == 0)
|
|
check("6 pins", len(s["pins"]) == 6)
|
|
pin = s["pins"][0]
|
|
check("pin voltage", abs(pin["voltage_v"] - 11.904) < 1e-9)
|
|
check("pin current", abs(pin["current_a"] - 6.221) < 1e-9)
|
|
check("pin power", abs(pin["power_w"] - 74.055) < 1e-9)
|
|
# Totals are computed from the pins (exporter behavior), not the
|
|
# device's total fields.
|
|
check(
|
|
"total current",
|
|
abs(s["current_total_a"] - round(6 * 6.221, 3)) < 1e-9,
|
|
)
|
|
check(
|
|
"total power",
|
|
abs(s["power_total_w"] - round(6 * 11.904 * 6.221, 3)) < 1e-9,
|
|
)
|
|
check(
|
|
"avg voltage",
|
|
abs(s["voltage_avg_v"] - 11.904) < 1e-3,
|
|
)
|
|
|
|
|
|
def test_parse_psu_capabilities():
|
|
for cap, watts in ((0, 600), (1, 450), (2, 300), (3, 150)):
|
|
s = wv.parse_sensor_struct(make_frame(psu_cap=cap))
|
|
check(f"psu cap {cap} -> {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 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()
|
|
print(f"\n{PASS} passed, {FAIL} failed")
|
|
return 1 if FAIL else 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main())
|