|
#!/usr/bin/env python3 |
|
|
|
#MIT License |
|
#Copyright (c) 2026 A C G Mennucci |
|
|
|
""" this script reads the Bluetooth data from the |
|
USB Digital Tester |
|
model J7-c |
|
produced by SJAMING |
|
that appears as a Bluetooth gadget as UC96_SPP |
|
""" |
|
|
|
import os, argparse, csv, datetime, serial, time, subprocess |
|
|
|
|
|
FRAME_LEN = 36 |
|
MAGIC_HEADER = b"\xff\x55" |
|
REPORT_MESSAGE_TYPE = 0x01 |
|
USB_DEVICE_TYPE = 0x03 |
|
|
|
|
|
def u16be(b, i): |
|
return (b[i] << 8) | b[i + 1] |
|
|
|
def u24be(b, i): |
|
return (b[i] << 16) | (b[i + 1] << 8) | b[i + 2] |
|
|
|
def u32be(b, i): |
|
return (b[i] << 24) | (b[i + 1] << 16) | (b[i + 2] << 8) | b[i + 3] |
|
|
|
def frame_checksum(b): |
|
checksum = 0 |
|
# SPP USB reports exclude the magic header and message type from checksum. |
|
for item in b[3:-1]: |
|
checksum = (checksum + item) & 0xff |
|
return checksum ^ 0x44 |
|
|
|
def is_usb_report_frame(b): |
|
return ( |
|
len(b) == FRAME_LEN |
|
and b[:2] == MAGIC_HEADER |
|
and b[2] == REPORT_MESSAGE_TYPE |
|
and b[3] == USB_DEVICE_TYPE |
|
and b[-1] == frame_checksum(b) |
|
) |
|
|
|
def parse_frame(b): |
|
if not is_usb_report_frame(b): |
|
raise ValueError("invalid USB report frame") |
|
|
|
voltage = u24be(b, 4) / 100.0 |
|
current = u24be(b, 7) / 100.0 |
|
power = voltage * current |
|
resistance = voltage / current if current else 0 |
|
data_minus = u16be(b, 17) / 100.0 |
|
data_plus = u16be(b, 19) / 100.0 |
|
|
|
hours = u16be(b, 23) |
|
minutes = b[25] |
|
seconds = b[26] |
|
elapsed = f"{hours:02d}:{minutes:02d}:{seconds:02d}" |
|
|
|
return { |
|
"timestamp": datetime.datetime.now().isoformat(timespec="seconds"), |
|
"voltage_V": voltage, |
|
"current_A": current, |
|
"power_W_calc": power, |
|
"resistance_ohm_calc": resistance, |
|
"capacity_mAh": u24be(b, 10), |
|
"energy_Wh": u32be(b, 13) / 100.0, |
|
"data_plus_V": data_plus, |
|
"data_minus_V": data_minus, |
|
"temperature_C": u16be(b, 21), |
|
'hours' : hours, |
|
'minutes' : minutes, |
|
'seconds' : seconds, |
|
'elapsed' : elapsed, |
|
"backlight_s": b[27], |
|
"const28_29": u16be(b, 28), |
|
"const32_33": u16be(b, 32), |
|
"checksum": b[35], |
|
"checksum_calc": frame_checksum(b), |
|
"raw": b.hex(" "), |
|
} |
|
|
|
def read_frame(ser): |
|
buf = bytearray() |
|
while True: |
|
c = ser.read(1) |
|
if not c: |
|
return None |
|
buf += c |
|
if len(buf) >= 2 and buf[-2:] == MAGIC_HEADER: |
|
frame = bytearray(MAGIC_HEADER) |
|
frame += ser.read(FRAME_LEN - len(MAGIC_HEADER)) |
|
if is_usb_report_frame(frame): |
|
return bytes(frame) |
|
|
|
def main(): |
|
ap = argparse.ArgumentParser() |
|
ap.add_argument("--address", default="A0:55:C1:44:AE:63", help='address of UC96_SPP device') |
|
ap.add_argument("--dev", default="/dev/rfcomm0", help='/dev/rfcommX to use') |
|
ap.add_argument("--baud", type=int, default=9600) |
|
ap.add_argument("--csv", default=None, help='output .csv file') |
|
ap.add_argument("--show-raw", action="store_true", help='print raw frame bytes') |
|
## currently this script does not resolve from name to address |
|
#ap.add_argument("--name", default="UC96_SPP") |
|
args = ap.parse_args() |
|
|
|
P = None |
|
rfcomm_num = None |
|
if not os.path.exists(args.dev): |
|
rfcomm_num = args.dev.replace("/dev/rfcomm", "") |
|
cmd = ["sudo", "rfcomm", "bind", rfcomm_num, args.address , "1"] |
|
print(f' ... binding to {args.address} ...') |
|
P = subprocess.run(cmd, check=False) |
|
if P.returncode: |
|
print("failed to rfcomm using ",cmd) |
|
return 1 |
|
ok = False |
|
for j in range(40): |
|
if os.path.exists(args.dev) and os.access(args.dev, os.R_OK): |
|
ok = True |
|
break |
|
time.sleep(0.2) |
|
if not ok: |
|
print("failed to rfcomm using ",cmd) |
|
return 1 |
|
|
|
ser = serial.Serial(args.dev, args.baud, timeout=2) |
|
|
|
csvfile = None |
|
writer = None |
|
if args.csv: |
|
csvfile = open(args.csv, "a", newline="") |
|
writer = None |
|
try: |
|
while True: |
|
frame = read_frame(ser) |
|
if frame is None: |
|
continue |
|
|
|
d = parse_frame(frame) |
|
|
|
print( |
|
f'{d["timestamp"]}', |
|
'\n ', |
|
f'V={d["voltage_V"]:6.2f} V ' |
|
f'I={d["current_A"]:5.2f} A ' |
|
f'P={d["power_W_calc"]:7.3f} W ' |
|
f'R={d["resistance_ohm_calc"]:7.2f} ohm ' |
|
'\n ', |
|
f'Cap={d["capacity_mAh"]:5d} mAh ' |
|
f'E={d["energy_Wh"]:7.2f} Wh ' |
|
f'T={d["temperature_C"]:3d} C ' |
|
f'elapsed={d["elapsed"]} ' |
|
'\n ', |
|
f'D+={d["data_plus_V"]:5.2f} V ' |
|
f'D-={d["data_minus_V"]:5.2f} V ' |
|
) |
|
if args.show_raw: |
|
print("RAW:", d["raw"]) |
|
|
|
if args.csv: |
|
if writer is None: |
|
writer = csv.DictWriter(csvfile, fieldnames=list(d.keys())) |
|
if csvfile.tell() == 0: |
|
writer.writeheader() |
|
writer.writerow(d) |
|
csvfile.flush() |
|
except (SystemExit,KeyboardInterrupt): |
|
pass |
|
if rfcomm_num is not None: |
|
cmd = ["sudo", "rfcomm", "release", rfcomm_num] |
|
print(f' ... unbinding from {args.address} ...') |
|
P = subprocess.run(cmd, check=False) |
|
|
|
|
|
if __name__ == "__main__": |
|
main() |
|
|