diff --git a/app/main.py b/app/main.py index ffce6a2..4db5847 100644 --- a/app/main.py +++ b/app/main.py @@ -76,12 +76,12 @@ app.include_router(routers.router) @app.get("/", response_class=HTMLResponse) def index(request: Request): - return templates.TemplateResponse("index.html", {"request": request}) + return templates.TemplateResponse(request, "index.html") @app.get("/roster", response_class=HTMLResponse) def roster(request: Request): - return templates.TemplateResponse("roster.html", {"request": request}) + return templates.TemplateResponse(request, "roster.html") @app.get("/health") diff --git a/nl52/usb_bulk_probe.py b/nl52/usb_bulk_probe.py new file mode 100644 index 0000000..ae98925 --- /dev/null +++ b/nl52/usb_bulk_probe.py @@ -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())