feat: add NL-52 USB bulk probe implementation

fix: jinja2 dict hash bug
This commit is contained in:
2026-06-08 17:57:30 +00:00
parent 0fcab0d0dc
commit 49c79e9c2a
2 changed files with 130 additions and 2 deletions
+2 -2
View File
@@ -76,12 +76,12 @@ app.include_router(routers.router)
@app.get("/", response_class=HTMLResponse) @app.get("/", response_class=HTMLResponse)
def index(request: Request): def index(request: Request):
return templates.TemplateResponse("index.html", {"request": request}) return templates.TemplateResponse(request, "index.html")
@app.get("/roster", response_class=HTMLResponse) @app.get("/roster", response_class=HTMLResponse)
def roster(request: Request): def roster(request: Request):
return templates.TemplateResponse("roster.html", {"request": request}) return templates.TemplateResponse(request, "roster.html")
@app.get("/health") @app.get("/health")
+128
View File
@@ -0,0 +1,128 @@
#!/usr/bin/env python3
"""
RION NL-42/NL-52 USB bulk probe — talks to the meter via libusb directly
(ctypes, no pyusb/pip needed).
The NL-52's USB serial interface (interface 0) is a plain vendor class with
two bulk endpoints and NO control protocol — not FTDI/CDC, so no kernel
serial driver applies. We just claim interface 0 and move ASCII over the
bulk pipe: write command -> EP 0x03 OUT, read reply <- EP 0x84 IN.
Needs write access to the USB device node (root, or a udev rule granting the
plugdev group). Run: sudo python3 nl52/usb_bulk_probe.py
"""
import ctypes as C
import time
VID, PID = 0x0EA3, 0x000F
EP_OUT, EP_IN = 0x03, 0x84
IFACE = 0
LIBUSB_ERROR_TIMEOUT = -7
lib = C.CDLL("libusb-1.0.so.0")
lib.libusb_init.argtypes = [C.POINTER(C.c_void_p)]
lib.libusb_open_device_with_vid_pid.argtypes = [C.c_void_p, C.c_uint16, C.c_uint16]
lib.libusb_open_device_with_vid_pid.restype = C.c_void_p
lib.libusb_set_auto_detach_kernel_driver.argtypes = [C.c_void_p, C.c_int]
lib.libusb_claim_interface.argtypes = [C.c_void_p, C.c_int]
lib.libusb_release_interface.argtypes = [C.c_void_p, C.c_int]
lib.libusb_set_interface_alt_setting.argtypes = [C.c_void_p, C.c_int, C.c_int]
lib.libusb_clear_halt.argtypes = [C.c_void_p, C.c_ubyte]
lib.libusb_close.argtypes = [C.c_void_p]
lib.libusb_exit.argtypes = [C.c_void_p]
lib.libusb_bulk_transfer.argtypes = [
C.c_void_p, C.c_ubyte, C.POINTER(C.c_ubyte), C.c_int, C.POINTER(C.c_int), C.c_uint
]
lib.libusb_strerror.argtypes = [C.c_int]
lib.libusb_strerror.restype = C.c_char_p
def err(code):
return lib.libusb_strerror(code).decode(errors="replace")
def main():
ctx = C.c_void_p()
if lib.libusb_init(C.byref(ctx)) != 0:
print("[!] libusb_init failed")
return 1
h = lib.libusb_open_device_with_vid_pid(ctx, VID, PID)
if not h:
print(f"[!] open {VID:04x}:{PID:04x} failed — device present? running as root?")
lib.libusb_exit(ctx)
return 1
lib.libusb_set_auto_detach_kernel_driver(h, 1)
rc = lib.libusb_claim_interface(h, IFACE)
if rc != 0:
print(f"[!] claim_interface({IFACE}) failed: {err(rc)}")
lib.libusb_close(h); lib.libusb_exit(ctx)
return 1
print(f"[*] Claimed interface {IFACE} on {VID:04x}:{PID:04x} — bulk OUT 0x{EP_OUT:02x}, IN 0x{EP_IN:02x}")
# Select alt setting 0 explicitly and clear any endpoint halts left over
# from prior (wrong-driver) probing.
rc = lib.libusb_set_interface_alt_setting(h, IFACE, 0)
print(f"[*] set_interface_alt_setting(0): {err(rc) if rc else 'ok'}")
for ep in (EP_OUT, EP_IN):
rc = lib.libusb_clear_halt(h, ep)
print(f"[*] clear_halt(0x{ep:02x}): {err(rc) if rc else 'ok'}")
print()
def bulk_out(data: bytes, timeout=2000):
buf = (C.c_ubyte * len(data)).from_buffer_copy(data)
actual = C.c_int(0)
rc = lib.libusb_bulk_transfer(h, EP_OUT, buf, len(data), C.byref(actual), timeout)
return rc, actual.value
def bulk_in(n=512, timeout=2000):
buf = (C.c_ubyte * n)()
actual = C.c_int(0)
rc = lib.libusb_bulk_transfer(h, EP_IN, buf, n, C.byref(actual), timeout)
return rc, bytes(buf[:actual.value])
def exchange(cmd: str):
rc, n = bulk_out((cmd + "\r\n").encode("ascii"))
if rc != 0:
print(f" > {cmd!r:24} OUT failed: {err(rc)}")
return
# Read until idle (collect result code + data lines)
chunks = bytearray()
end = time.time() + 2.0
while time.time() < end:
rrc, data = bulk_in(512, 600)
if rrc == 0 and data:
chunks.extend(data)
end = min(end, time.time() + 0.4) # brief idle window then stop
elif rrc == LIBUSB_ERROR_TIMEOUT:
if chunks:
break
else:
if rrc != 0:
break
print(f" > {cmd!r:24} -> {bytes(chunks)!r}")
# Read-first sanity check: is the IN endpoint alive / does the device send
# anything unsolicited?
rrc, data = bulk_in(512, 1000)
print(f"[*] read-first IN: rc={err(rrc) if rrc else 'ok'} data={data!r}\n")
try:
for cmd in ["System Version?NL", "Measure?", "DOD?", "Battery Type?"]:
exchange(cmd)
time.sleep(0.3)
finally:
lib.libusb_release_interface(h, IFACE)
lib.libusb_close(h)
lib.libusb_exit(ctx)
print("\n[done]")
return 0
if __name__ == "__main__":
raise SystemExit(main())