Thermal Tests continued
I had forgotten to mention in my previous post that after the roadtest benchmarking is complete that I still want to monitor thermal and power parameters on the Frigate NVR. To that end, I'm going to create MQTT sensor entities for those parameters so that I can monitor them in Home Assistant and also display them on a Node-Red dashboard.
I installed the Mosquitto MQTT client on the CM5 and added a crontab entry to publish the CM5 CPU Temperature once a minute
.The crontab syntax is:
* * * * * mosquitto_pub -h localhost -t "mytopic" -m "my_message"
mytopic is cm5/hardware/cpu_temp
my_message is $(( $(cat /sys/class/thermal/thermal_zone0/temp) / 1000 ))
Over on my RPi CM4 which is running Home Assistant, I added an mqtt sensor entity in my configuration.yaml
mqtt:
sensor:
- name: "CM5 CPU Temperature"
state_topic: "cm5/hardware/cpu_temp"
unit_of_measurement: "°C"
device_class: temperature
state_class: measurement
And I can then view the sensor data directly by name.
The plot is showing the temperature over a couple of days. In the first third of the plot, Frigate is running initially with a single camera. Then I stop Frigate and run the stress test (the large spike). Then the system is just idling. The flat line is a period when the system was turned off (it retains the last received value). Then system turns on, Frigate runs single camera then back to idle.
Sensor entities are pretty useful to examine data timelines independently from a dashboard, so I'll create them for parameters that I want to watch, but I'll also create a Node-Red dashboard for real-time monitoring.
Power Logging
To do the power logging using the FNIRSI FNB58 I am going to use OpenFNB58 which is an open-source logger written by Libor Tomsik. In particular I am going to use the Python CLI reader fnb58.py. .This reader can interact with the FNB58 using USB HID or Bluetooth LE, but I'm going to use USB for speed and reliability. The FNB58 is quite a capable tool, it can sample at 100Hz which might be good for looking at quick events but I don't expect to run faster than 1 Hz.
First thing to do is to verify that Linux can see the FNB58 when you plug it in.
Everything looks good, so any required drivers are already available.
Clone the OpenFNB58 repo from github.
All FNB58 programs will be run from this directory to insure we have the correct path to required files.
Quick test of fnb58.py to verify it is operating okay with a single measurement test.
For reference here is the fnb58.py file from the repo:
#!/usr/bin/env python3
"""
FNB58 USB Power Meter reader
Supports two transport modes:
--usb (default) USB HID via pyusb (interface 3, endpoints 0x03/0x83)
--ble Bluetooth LE via BlueZ D-Bus (ffe0 GATT service)
=== USB HID protocol (64-byte reports) ===
Init: aa 81 00..00 8e / aa 82 00..00 96 (×2 for FNB58)
Poll: aa 83 00..00 9e (every ~1 s)
Data packet [0]=0xaa [1]=0x04, then 4×15-byte samples:
[0..3] voltage uint32-LE / 100000 V
[4..7] current uint32-LE / 100000 A
[8..9] D+ uint16-LE / 1000 V
[10..11] D- uint16-LE / 1000 V
[12] unknown
[13..14] temp uint16-LE / 10 °C
=== BLE protocol (GATT, service ffe0, RFstar transparent UART) ===
Device: FNB58-XXXXXX (default MAC set by BLE_DEFAULT_MAC below)
Char ffe9 (write / write-without-response): send commands
Char ffe4 (notify): receive streaming data
Init sequence (4-byte commands sent to ffe9):
aa 81 00 f4 — wake up / enter measurement mode
aa 82 00 a7 — start data stream (send ~2 s after first command)
Stream packet framing: aa [type] [data_len] [data...] [checksum]
total bytes = data_len + 4
Packet types in the stream:
0x07 VBUS/IBUS — data_len=4
data[0..1] uint16-LE / 1000 VBUS (V)
data[2..3] uint16-LE / 1000 IBUS (A)
0x06 D+/D- — data_len=6
data[0..1] uint16-LE / 1000 D+ (V)
data[2..3] uint16-LE / 1000 D- (V)
data[4..5] unknown
0x04 Accumulated — data_len=12
data[0..3] uint32-LE accumulated charge (unit TBD)
0x03 Device info — data_len=14 (sent once on init)
0x05 Status/misc — data_len=7
0x08 Config/misc — data_len=17
Multiple packet types may be concatenated in a single BLE notification chunk.
"""
import sys
import os
import errno
import glob
import select
import subprocess
import time
import struct
import argparse
VID_FNB58 = "2e3c"
PID_FNB58 = "5558"
SAMPLES_PER_SECOND = 100
TIME_INTERVAL = 1.0 / SAMPLES_PER_SECOND
CMD_INIT1 = b"\xaa\x81" + b"\x00" * 61 + b"\x8e"
CMD_INIT2 = b"\xaa\x82" + b"\x00" * 61 + b"\x96"
CMD_POLL = b"\xaa\x83" + b"\x00" * 61 + b"\x9e"
# ── Protocol definitions ──────────────────────────────────────────────────
# Matching the FNIRSI Windows app's protocol list.
# Protocol IDs are used by the trigger command (cmd 0x85).
# NOTE: The trigger command byte (0x85) and payload format are derived from
# reverse engineering; verify with a QC/PD-capable charger.
PROTOCOLS = {
# id: (name, default_voltage_mv, [available_mv])
1: ("QC 2.0 5V", 5000, [5000]),
2: ("QC 2.0 9V", 9000, [9000]),
3: ("QC 2.0 12V", 12000, [12000]),
4: ("QC 2.0 20V", 20000, [20000]),
5: ("QC 3.0", 9000, list(range(3600, 20001, 200))),
6: ("FCP/AFC 9V", 9000, [9000]),
7: ("FCP/AFC 12V", 12000, [12000]),
8: ("Huawei FCP", 9000, [9000, 12000]),
9: ("Huawei SCP", 5000, list(range(4000, 8001, 100))),
10: ("Samsung 2A", 5000, [5000]),
11: ("Samsung AFC", 9000, [9000, 12000]),
12: ("Apple 2.1A", 5000, [5000]),
13: ("Apple 2.4A", 5000, [5000]),
14: ("PD/MTK", 9000, [5000, 9000, 12000, 15000, 20000]),
}
CMD_TRIGGER_BYTE = 0x85 # single trigger (best-guess; not officially documented)
CMD_RELEASE_BYTE = 0x86 # release trigger (best-guess)
def crc8_fnb58(data: bytes) -> int:
"""
CRC-8 over the given bytes.
Parameters: poly=0x39, init=0x42, refin=False, refout=False, xorout=0x00.
For a 64-byte USB HID frame, pass bytes[1:63].
"""
poly, crc = 0x39, 0x42
for b in data:
crc ^= b
for _ in range(8):
crc = ((crc << 1) ^ poly) & 0xFF if (crc & 0x80) else (crc << 1) & 0xFF
return crc
def make_hid_cmd(cmd_byte: int, payload: bytes = b"") -> bytes:
"""Build a 64-byte USB HID command with the correct CRC-8 checksum."""
inner = bytes([cmd_byte]) + bytes(payload) + bytes(61 - len(payload))
return bytes([0xAA]) + inner + bytes([crc8_fnb58(inner)])
def make_trigger_cmd(proto_id: int, voltage_mv: int = 0) -> bytes:
"""
Build a trigger command for the given protocol ID and target voltage.
voltage_mv is encoded as a LE uint16 in 10 mV units (e.g. 9000 mV → 0x0384).
"""
v10 = voltage_mv // 10 # 10 mV resolution
payload = bytes([proto_id, v10 & 0xFF, (v10 >> 8) & 0xFF])
return make_hid_cmd(CMD_TRIGGER_BYTE, payload)
def make_release_cmd() -> bytes:
"""Build the trigger release command."""
return make_hid_cmd(CMD_RELEASE_BYTE)
def find_hidraw(wait: float = 10.0) -> str:
"""
Find the /dev/hidrawN node for the FNB58 by scanning sysfs.
Waits up to `wait` seconds for the device to appear (it may be rebooting).
Returns the device path, or exits with an error if not found.
"""
deadline = time.time() + wait
while True:
for hidraw_dir in glob.glob("/sys/class/hidraw/hidraw*"):
uevent_path = os.path.join(hidraw_dir, "device", "..", "uevent")
try:
uevent = open(uevent_path).read()
except OSError:
continue
product = ""
for line in uevent.splitlines():
if line.startswith("PRODUCT="):
product = line.split("=", 1)[1].lower()
if product.startswith(f"{VID_FNB58}/{PID_FNB58}/"):
return f"/dev/{os.path.basename(hidraw_dir)}"
if time.time() >= deadline:
break
print("Waiting for FNB58 HID device...", file=sys.stderr)
time.sleep(1.0)
print(
"FNB58 HID device not found after waiting. Is it connected?",
file=sys.stderr,
)
sys.exit(1)
def open_hid(path: str) -> int:
"""Open the hidraw device; returns a file descriptor, or exits on error."""
try:
return os.open(path, os.O_RDWR)
except PermissionError:
print(f"Cannot open {path}: permission denied.", file=sys.stderr)
sys.exit(1)
except OSError as e:
print(f"Cannot open {path}: {e}", file=sys.stderr)
sys.exit(1)
def hid_write(fd: int, cmd: bytes, retries: int = 5) -> bool:
"""
Write a HID command. Retries on ETIMEDOUT (device busy streaming).
Returns True on success, False on hard failure.
"""
for _ in range(retries):
try:
os.write(fd, cmd)
return True
except OSError as e:
if e.errno in (errno.ETIMEDOUT, errno.EPROTO, errno.ENODEV):
time.sleep(0.15)
continue
return False
return False
def hid_drain(fd: int, max_packets: int = 200):
"""
Drain pending HID input reports so the device isn't backlogged when we write.
Stops when no packet arrives within 20 ms or max_packets reached.
"""
for _ in range(max_packets):
r, _, _ = select.select([fd], [], [], 0.02)
if not r:
break
try:
os.read(fd, 64)
except OSError:
break
def infer_protocol(dp: float, dn: float, vbus: float = 5.0) -> str:
"""
Infer the charging protocol from D+/D- voltages measured by the FNB58.
These voltages reflect what the charger (upstream side) is asserting.
Returns a short protocol name string.
"""
tol = 0.18 # ±180 mV tolerance
def near(v, target):
return abs(v - target) < tol
def near2(v1, t1, v2, t2):
return near(v1, t1) and near(v2, t2)
# No voltage on either line → SDP or no charger detected
if dp < 0.35 and dn < 0.35:
return "USB SDP"
# Both lines equal and low → BC 1.2 DCP / Samsung-style charger
if near(dp, dn) and 0.5 < dp < 2.5:
if near2(dp, 2.0, dn, 2.0):
return "Apple 1A"
if near2(dp, 2.7, dn, 2.0):
return "Apple 2.1A"
if near2(dp, 2.0, dn, 2.7):
return "Apple 2.4A"
if 1.0 < dp < 1.8:
return "DCP / Samsung"
if 1.8 < dp < 2.3:
return "Apple 1A"
return f"DCP ({dp:.2f}V)"
# QC 2.0 voltage codes (D+, D-)
if near2(dp, 0.6, dn, 0.0):
return "QC 2.0 5V"
if near2(dp, 3.3, dn, 0.6):
return "QC 2.0 9V"
if near2(dp, 0.6, dn, 0.6):
return "QC 2.0 12V / CDP"
if near2(dp, 3.3, dn, 3.3):
return "QC 2.0 20V"
# QC 3.0 continuous mode: D+ = 0.6 V, D- toggling
if near(dp, 0.6) and 0.0 < dn < 3.6:
return "QC 3.0"
# Huawei FCP: D+ ≈ 0.325 V used as handshake initiation signal
if 0.2 < dp < 0.45 and dn < 0.3:
return "Huawei FCP"
# If VBUS is elevated but D+/D- still show 5V pattern → USB PD (CC pins used)
if vbus > 5.5 and dp < 0.5 and dn < 0.5:
return f"USB PD ({vbus:.0f}V)"
if vbus > 5.5:
return f"Fast Charge ({vbus:.0f}V)"
return f"Unknown (D+={dp:.2f}V D-={dn:.2f}V)"
def decode_packet(data):
"""Decode a 64-byte data packet into a list of sample dicts."""
if data[0] != 0xAA or data[1] != 0x04:
return None # not a data packet
samples = []
for i in range(4):
off = 2 + 15 * i
voltage = struct.unpack_from("<I", data, off)[0] / 100000.0
current = struct.unpack_from("<I", data, off + 4)[0] / 100000.0
dp = struct.unpack_from("<H", data, off + 8)[0] / 1000.0
dn = struct.unpack_from("<H", data, off + 10)[0] / 1000.0
temp = struct.unpack_from("<H", data, off + 13)[0] / 10.0
power = voltage * current
samples.append({
"voltage_V": voltage,
"current_A": current,
"power_W": power,
"dp_V": dp,
"dn_V": dn,
"temp_C": temp,
"protocol": infer_protocol(dp, dn, voltage),
})
return samples
def hid_read(fd: int, timeout: float = 2.0) -> bytes | None:
"""Read one 64-byte HID report with a timeout. Returns None on timeout."""
r, _, _ = select.select([fd], [], [], timeout)
if not r:
return None
try:
return os.read(fd, 64)
except OSError:
return None
def _hid_init(fd: int) -> bool:
"""
Drain any pending data, then send the init sequence.
Returns True if init commands were written successfully.
"""
hid_drain(fd)
if not hid_write(fd, CMD_INIT1):
return False
time.sleep(0.05)
if not hid_write(fd, CMD_INIT2):
return False
time.sleep(0.05)
if not hid_write(fd, CMD_INIT2):
return False
return True
def read_once(fd: int) -> list | None:
"""
Read one measurement. Tries progressively more aggressive init sequences:
1. Drain + poll only (device already streaming from a prior run)
2. Drain + INIT2×2 (gentle restart, avoids the aa81 reboot risk)
3. Drain + full INIT (aa81 + aa82×2, only if gentler methods fail)
"""
for phase, cmds in enumerate([
[CMD_POLL],
[CMD_INIT2, CMD_INIT2],
[CMD_INIT1, CMD_INIT2, CMD_INIT2],
]):
hid_drain(fd)
for cmd in cmds:
if not hid_write(fd, cmd):
return None
time.sleep(0.05)
deadline = time.time() + (1.5 if phase == 0 else 4.0)
while time.time() < deadline:
data = hid_read(fd, timeout=min(1.5, deadline - time.time()))
if data is None:
break
samples = decode_packet(data)
if samples:
return samples
return None
def run_monitor(fd: int, interval=1.0, count=None):
"""Continuously poll and print measurements."""
hid_drain(fd)
if not hid_write(fd, CMD_INIT1) or not hid_write(fd, CMD_INIT2) or not hid_write(fd, CMD_INIT2):
print("Failed to initialize device.", file=sys.stderr)
sys.exit(1)
print(f"{'Time':>10} {'Voltage':>9} {'Current':>9} {'Power':>8} {'D+':>6} {'D-':>6} {'Temp':>7} Protocol")
print(f"{'(s)':>10} {'(V)':>9} {'(A)':>9} {'(W)':>8} {'(V)':>6} {'(V)':>6} {'(°C)':>7}")
print("-" * 90)
next_poll = time.time() + interval
start = time.time()
n = 0
try:
while count is None or n < count:
now = time.time()
if now >= next_poll:
hid_write(fd, CMD_POLL)
next_poll = now + interval
data = hid_read(fd, timeout=min(1.0, max(0.05, next_poll - time.time())))
if data is None:
continue
samples = decode_packet(data)
if samples is None:
continue
t = time.time() - start
s = samples[0]
print(
f"{t:10.2f} {s['voltage_V']:9.5f} {s['current_A']:9.5f}"
f" {s['power_W']:8.5f} {s['dp_V']:6.3f} {s['dn_V']:6.3f} {s['temp_C']:7.1f}"
f" {s['protocol']}"
)
sys.stdout.flush()
n += 1
except KeyboardInterrupt:
print("\nStopped.", file=sys.stderr)
def trigger(fd: int, proto_id: int, voltage_mv: int = 0, hold_secs: float = 3.0):
"""
Send a single trigger command to the FNB58 to negotiate a fast-charge protocol.
proto_id : protocol ID (1–14, see PROTOCOLS dict)
voltage_mv : target voltage in millivolts (0 = use protocol default)
hold_secs : seconds to hold the trigger before releasing (0 = no release)
NOTE: Trigger command bytes are reverse-engineered / best-guess.
Test with a QC/PD-capable charger and report results.
"""
proto = PROTOCOLS.get(proto_id)
if proto is None:
print(f"Unknown protocol ID {proto_id}. Valid IDs: {list(PROTOCOLS.keys())}", file=sys.stderr)
sys.exit(1)
name, default_mv, _ = proto
if voltage_mv == 0:
voltage_mv = default_mv
print(f"Triggering: {name} @ {voltage_mv/1000:.1f} V (proto_id={proto_id})", file=sys.stderr)
hid_write(fd, make_trigger_cmd(proto_id, voltage_mv))
if hold_secs > 0:
time.sleep(hold_secs)
hid_write(fd, make_release_cmd())
print("Trigger released.", file=sys.stderr)
# ──────────────────────────────────────────────────────────────────────────────
# BLE transport (BlueZ D-Bus, service ffe0 / RFstar transparent UART)
# ──────────────────────────────────────────────────────────────────────────────
BLE_DEFAULT_MAC = "BA:03:18:7A:23:DF"
# UUIDs for the transparent UART characteristics
UUID_NOTIFY = "0000ffe4-0000-1000-8000-00805f9b34fb" # ffe4 — notifications
UUID_WRITE = "0000ffe9-0000-1000-8000-00805f9b34fb" # ffe9 — write
# 4-byte init commands sent over ffe9 to start the measurement stream
BLE_CMD_INIT1 = bytes([0xaa, 0x81, 0x00, 0xf4])
BLE_CMD_INIT2 = bytes([0xaa, 0x82, 0x00, 0xa7])
# Stream packet data_len by type (used for framing)
_BLE_PKT_LENS = {0x03: 14, 0x04: 12, 0x05: 7, 0x06: 6, 0x07: 4, 0x08: 17}
def _ble_ensure_connected(mac: str):
"""Connect to BLE device via bluetoothctl if not already connected."""
import dbus
try:
bus = dbus.SystemBus()
dev_path = f"/org/bluez/hci0/dev_{mac.replace(':', '_').upper()}"
props = dbus.Interface(bus.get_object("org.bluez", dev_path), "org.freedesktop.DBus.Properties")
if props.Get("org.bluez.Device1", "Connected"):
return # already connected
except Exception:
pass
print(f"Connecting to {mac}...", file=sys.stderr)
result = subprocess.run(
["bluetoothctl", "connect", mac],
capture_output=True, text=True, timeout=15
)
if "Connection successful" not in result.stdout:
print(f"Could not connect: {result.stdout.strip()}", file=sys.stderr)
sys.exit(1)
time.sleep(1.5) # give BlueZ time to resolve GATT services
def _ble_find_chars(bus, mac: str) -> tuple[str, str]:
"""
Enumerate BlueZ GATT objects to find the D-Bus paths for ffe4 and ffe9
characteristics under the given device MAC. Returns (notify_path, write_path).
"""
import dbus
mgr = dbus.Interface(bus.get_object("org.bluez", "/"), "org.freedesktop.DBus.ObjectManager")
objs = mgr.GetManagedObjects()
dev_prefix = f"/org/bluez/hci0/dev_{mac.replace(':', '_').upper()}"
notify_path = write_path = None
for path, ifaces in objs.items():
if not str(path).startswith(dev_prefix):
continue
char = ifaces.get("org.bluez.GattCharacteristic1")
if not char:
continue
uuid = str(char.get("UUID", ""))
if uuid == UUID_NOTIFY:
notify_path = str(path)
elif uuid == UUID_WRITE:
write_path = str(path)
if not notify_path or not write_path:
print(
f"Could not find ffe4/ffe9 characteristics for {mac}.\n"
"Is the device connected? Try: bluetoothctl connect " + mac,
file=sys.stderr,
)
sys.exit(1)
return notify_path, write_path
def _parse_ble_stream(chunk: bytes) -> list[dict]:
"""
Parse a BLE notification chunk that may contain multiple concatenated packets.
Each packet: aa [type] [data_len] [data...] [checksum]
Returns decoded temperature, D+/D-, and VBUS/IBUS measurement dicts.
"""
results = []
i = 0
while i < len(chunk):
if chunk[i] != 0xAA:
i += 1
continue
if i + 2 >= len(chunk):
break
ptype = chunk[i + 1]
dlen = chunk[i + 2]
expected = _BLE_PKT_LENS.get(ptype)
if expected is None or dlen != expected:
i += 1
continue
total = dlen + 4 # aa + type + len + data + checksum
if i + total > len(chunk):
break
data = chunk[i + 3 : i + 3 + dlen]
if ptype == 0x07:
vbus = struct.unpack_from("<H", data, 0)[0] / 1000.0
ibus = struct.unpack_from("<H", data, 2)[0] / 1000.0
results.append({"type": "vi", "vbus_V": vbus, "ibus_A": ibus, "power_W": vbus * ibus})
elif ptype == 0x05:
temp = struct.unpack_from("<H", data, 5)[0] / 10.0
results.append({"type": "temp", "temp_C": temp})
elif ptype == 0x06:
dp = struct.unpack_from("<H", data, 0)[0] / 1000.0
dn = struct.unpack_from("<H", data, 2)[0] / 1000.0
results.append({"type": "dp", "dp_V": dp, "dn_V": dn})
i += total
return results
def _ble_require_imports():
try:
import dbus
import dbus.mainloop.glib
from gi.repository import GLib
return dbus, GLib
except ImportError:
print("BLE mode requires: sudo apt install python3-dbus python3-gi", file=sys.stderr)
sys.exit(1)
def ble_read_once(mac: str) -> dict | None:
"""
Start the BLE stream, collect the first VBUS/IBUS, D+/D-, and temperature
readings, then return them.
"""
dbus, GLib = _ble_require_imports()
import dbus.mainloop.glib
dbus.mainloop.glib.DBusGMainLoop(set_as_default=True)
_ble_ensure_connected(mac)
bus = dbus.SystemBus()
notify_path, write_path = _ble_find_chars(bus, mac)
ni = dbus.Interface(bus.get_object("org.bluez", notify_path), "org.bluez.GattCharacteristic1")
wi = dbus.Interface(bus.get_object("org.bluez", write_path), "org.bluez.GattCharacteristic1")
result = {}
loop = GLib.MainLoop()
def on_data(iface, changed, inv):
if "Value" not in changed:
return
for pkt in _parse_ble_stream(bytes(changed["Value"])):
result.update(pkt)
if "vbus_V" in result and "dp_V" in result and "temp_C" in result:
loop.quit()
bus.add_signal_receiver(on_data,
dbus_interface="org.freedesktop.DBus.Properties",
signal_name="PropertiesChanged", path=notify_path)
try: ni.StopNotify()
except Exception: pass
ni.StartNotify()
def _send_init2():
wi.WriteValue(dbus.Array([dbus.Byte(b) for b in BLE_CMD_INIT2], signature="y"), {})
wi.WriteValue(dbus.Array([dbus.Byte(b) for b in BLE_CMD_INIT1], signature="y"), {})
GLib.timeout_add(2000, lambda: _send_init2() or False)
GLib.timeout_add(8000, lambda: loop.quit() or False)
loop.run()
try: ni.StopNotify()
except Exception: pass
return result if "vbus_V" in result else None
def ble_monitor(mac: str, count: int | None = None):
"""Stream measurements continuously from BLE and print each VBUS/IBUS packet."""
dbus, GLib = _ble_require_imports()
import dbus.mainloop.glib
dbus.mainloop.glib.DBusGMainLoop(set_as_default=True)
_ble_ensure_connected(mac)
bus = dbus.SystemBus()
notify_path, write_path = _ble_find_chars(bus, mac)
ni = dbus.Interface(bus.get_object("org.bluez", notify_path), "org.bluez.GattCharacteristic1")
wi = dbus.Interface(bus.get_object("org.bluez", write_path), "org.bluez.GattCharacteristic1")
print(f"BLE mode — device {mac}")
print(f"{'Time':>10} {'VBUS':>9} {'IBUS':>9} {'Power':>9} {'D+':>7} {'D-':>7} {'Temp':>7}")
print(f"{'(s)':>10} {'(V)':>9} {'(A)':>9} {'(W)':>9} {'(V)':>7} {'(V)':>7} {'(°C)':>7}")
print("-" * 77)
start = time.time()
n = [0]
last_dp = [None]
last_temp = [None]
loop = GLib.MainLoop()
def on_data(iface, changed, inv):
if "Value" not in changed:
return
for pkt in _parse_ble_stream(bytes(changed["Value"])):
if pkt["type"] == "dp":
last_dp[0] = pkt
elif pkt["type"] == "temp":
last_temp[0] = pkt["temp_C"]
elif pkt["type"] == "vi":
t = time.time() - start
dp = last_dp[0]["dp_V"] if last_dp[0] else float("nan")
dn = last_dp[0]["dn_V"] if last_dp[0] else float("nan")
temp = last_temp[0] if last_temp[0] is not None else float("nan")
print(
f"{t:10.2f} {pkt['vbus_V']:9.3f} {pkt['ibus_A']:9.3f}"
f" {pkt['power_W']:9.5f} {dp:7.3f} {dn:7.3f} {temp:7.1f}"
)
sys.stdout.flush()
n[0] += 1
if count is not None and n[0] >= count:
loop.quit()
bus.add_signal_receiver(on_data,
dbus_interface="org.freedesktop.DBus.Properties",
signal_name="PropertiesChanged", path=notify_path)
try: ni.StopNotify()
except Exception: pass
ni.StartNotify()
def _send_init2():
wi.WriteValue(dbus.Array([dbus.Byte(b) for b in BLE_CMD_INIT2], signature="y"), {})
wi.WriteValue(dbus.Array([dbus.Byte(b) for b in BLE_CMD_INIT1], signature="y"), {})
GLib.timeout_add(2000, lambda: _send_init2() or False)
try:
loop.run()
except KeyboardInterrupt:
print("\nStopped.", file=sys.stderr)
try: ni.StopNotify()
except Exception: pass
# ──────────────────────────────────────────────────────────────────────────────
# Entry point
# ──────────────────────────────────────────────────────────────────────────────
def main():
parser = argparse.ArgumentParser(
description="Read VBUS/IBUS and more from an FNIRSI FNB58 USB power meter.",
formatter_class=argparse.ArgumentDefaultsHelpFormatter,
)
transport = parser.add_mutually_exclusive_group()
transport.add_argument(
"--usb", action="store_true", default=True,
help="Use USB HID (default)"
)
transport.add_argument(
"--ble", action="store_true",
help="Use Bluetooth LE (requires python3-dbus, python3-gi)"
)
parser.add_argument(
"--mac", default=BLE_DEFAULT_MAC,
help="BLE device MAC address"
)
parser.add_argument(
"-n", "--count", type=int, default=None,
help="Number of readings (default: run forever, Ctrl-C to stop)"
)
parser.add_argument(
"--once", action="store_true",
help="Take a single snapshot and exit"
)
parser.add_argument(
"--interval", type=float, default=1.0,
help="Poll interval in seconds"
)
parser.add_argument(
"--trigger", type=int, default=None, metavar="PROTO_ID",
help=(
"Send a fast-charge trigger and release it after --hold seconds. "
"PROTO_ID: " + ", ".join(f"{k}={v[0]}" for k,v in PROTOCOLS.items())
)
)
parser.add_argument(
"--voltage", type=float, default=0,
help="Trigger target voltage in V (0 = protocol default)"
)
parser.add_argument(
"--hold", type=float, default=3.0,
help="Seconds to hold the trigger before releasing (0 = no release)"
)
args = parser.parse_args()
# ── BLE mode ──
if args.ble:
if args.once:
s = ble_read_once(args.mac)
if s:
dp = s.get('dp_V', float('nan'))
dn = s.get('dn_V', float('nan'))
vbs = s.get('vbus_V', float('nan'))
print(f"VBUS : {vbs:.3f} V")
print(f"IBUS : {s.get('ibus_A', float('nan')):.3f} A")
print(f"Power : {s.get('power_W', float('nan')):.5f} W")
print(f"D+ : {dp:.3f} V")
print(f"D- : {dn:.3f} V")
print(f"Temp : {s.get('temp_C', float('nan')):.1f} °C")
print(f"Protocol : {infer_protocol(dp, dn, vbs)}")
else:
print("No BLE data received.", file=sys.stderr)
sys.exit(1)
else:
ble_monitor(args.mac, count=args.count)
return
# ── USB HID mode ──
path = find_hidraw()
fd = open_hid(path)
try:
if args.trigger is not None:
# Ensure device is streaming first
read_once(fd)
trigger(fd, args.trigger, int(args.voltage * 1000), args.hold)
elif args.once:
samples = read_once(fd)
if samples is None:
# Device may be mid-reboot; wait for it to come back and retry
os.close(fd)
path = find_hidraw(wait=20.0)
fd = open_hid(path)
samples = read_once(fd)
if samples:
s = samples[0]
print(f"VBUS : {s['voltage_V']:.5f} V")
print(f"IBUS : {s['current_A']:.5f} A")
print(f"Power : {s['power_W']:.5f} W")
print(f"D+ : {s['dp_V']:.3f} V")
print(f"D- : {s['dn_V']:.3f} V")
print(f"Temp : {s['temp_C']:.1f} °C")
print(f"Protocol : {s['protocol']}")
else:
print("No data received.", file=sys.stderr)
sys.exit(1)
else:
run_monitor(fd, interval=args.interval, count=args.count)
finally:
os.close(fd)
if __name__ == "__main__":
main()
The program allows for using USB_HID or BLE for communicating with the FNB58. Since I'll be using USB, I can simplify the code by deleting the BLE sections when I create the benchmark logger. For the initial test we can use the program unchanged just by passing the usb argument at the command line. The program uses polling for the USB_HID protocol. The default is 1 second interval.
Here are the results of the first few seconds: (fnb58_debug.py just adds debug printouts to the original program)
ralph@CM5-Frigate:~/OpenFNB58 $ python3 -u fnb58_debug.py --usb --interval 1
Time Voltage Current Power D+ D- Temp Protocol
(s) (V) (A) (W) (V) (V) (°C)
------------------------------------------------------------------------------------------
0.04 5.12432 0.63418 3.24974 0.034 0.059 27.2 USB SDP
0.08 5.12432 0.66473 3.40629 0.034 0.059 27.2 USB SDP
0.12 5.13784 0.55195 2.83583 0.034 0.059 27.2 USB SDP
0.16 5.12436 0.55394 2.83859 0.033 0.058 27.2 USB SDP
0.20 5.12436 0.56313 2.88568 0.032 0.058 27.2 USB SDP
0.24 5.12444 0.57743 2.95901 0.033 0.058 27.2 USB SDP
0.28 5.12444 0.55518 2.84499 0.033 0.058 27.2 USB SDP
0.32 5.12448 0.56015 2.87048 0.033 0.058 27.2 USB SDP
0.36 5.12448 0.55593 2.84885 0.033 0.057 27.2 USB SDP
0.40 5.12448 0.57754 2.95959 0.033 0.058 27.2 USB SDP
0.44 5.12440 0.55220 2.82969 0.033 0.058 27.2 USB SDP
0.48 5.12440 0.57729 2.95826 0.033 0.058 27.2 USB SDP
0.52 5.12436 0.56760 2.90859 0.032 0.058 27.2 USB SDP
0.56 5.12436 0.57763 2.95998 0.033 0.058 27.2 USB SDP
0.60 5.12436 0.55096 2.82332 0.033 0.058 27.2 USB SDP
0.64 5.12440 0.56288 2.88442 0.033 0.058 27.2 USB SDP
0.68 5.12440 0.54723 2.80423 0.033 0.058 27.2 USB SDP
0.72 5.12448 0.57766 2.96021 0.033 0.058 27.2 USB SDP
0.76 5.12448 0.54922 2.81447 0.033 0.058 27.2 USB SDP
0.80 5.12448 0.55717 2.85521 0.033 0.058 27.2 USB SDP
0.84 5.12455 0.55146 2.82598 0.033 0.058 27.2 USB SDP
0.88 5.12455 0.57760 2.95994 0.033 0.058 27.2 USB SDP
0.92 5.12471 0.54897 2.81331 0.033 0.058 27.2 USB SDP
0.96 5.11152 0.60958 3.11588 0.033 0.058 27.2 USB SDP
1.00 5.12471 0.57752 2.95962 0.033 0.058 27.2 USB SDP
DEBUG: sending POLL
DEBUG: POLL write result = True
1.04 5.12487 0.58425 2.99421 0.033 0.058 27.2 USB SDP
1.08 5.12487 0.56697 2.90565 0.033 0.058 27.2 USB SDP
1.12 5.12491 0.55873 2.86344 0.033 0.058 27.2 USB SDP
1.16 5.12491 0.55873 2.86344 0.032 0.057 27.2 USB SDP
1.20 5.12491 0.57034 2.92294 0.033 0.058 27.2 USB SDP
1.24 5.12471 0.55608 2.84975 0.033 0.058 27.2 USB SDP
1.28 5.12471 0.55608 2.84975 0.033 0.058 27.2 USB SDP
1.32 5.12487 0.57282 2.93563 0.032 0.057 27.2 USB SDP
1.36 5.12487 0.56711 2.90637 0.032 0.058 27.2 USB SDP
1.40 5.12487 0.55596 2.84922 0.033 0.058 27.2 USB SDP
1.44 5.12495 0.55617 2.85034 0.033 0.058 27.2 USB SDP
1.48 5.12495 0.56686 2.90513 0.033 0.058 27.2 USB SDP
1.52 5.12506 0.58971 3.02230 0.033 0.058 27.2 USB SDP
1.56 5.12506 0.55630 2.85107 0.033 0.058 27.2 USB SDP
1.60 5.12506 0.55630 2.85107 0.033 0.058 27.2 USB SDP
1.64 5.12518 0.57282 2.93581 0.033 0.058 27.2 USB SDP
1.68 5.12518 0.58176 2.98162 0.033 0.058 27.2 USB SDP
1.72 5.12518 0.55636 2.85145 0.033 0.058 27.2 USB SDP
1.76 5.12518 0.55636 2.85145 0.033 0.058 27.2 USB SDP
1.80 5.12518 0.56760 2.90905 0.033 0.058 27.2 USB SDP
1.84 5.12510 0.57307 2.93704 0.033 0.058 27.2 USB SDP
1.88 5.12510 0.55666 2.85294 0.033 0.058 27.2 USB SDP
1.92 5.12502 0.55666 2.85289 0.033 0.058 27.2 USB SDP
1.96 5.12502 0.60362 3.09356 0.033 0.058 27.2 USB SDP
2.00 5.12502 0.60139 3.08214 0.032 0.057 27.2 USB SDP
DEBUG: sending POLL
DEBUG: POLL write result = True
2.04 5.12498 0.55684 2.85379 0.033 0.058 27.2 USB SDP
2.08 5.12498 0.55684 2.85379 0.033 0.057 27.2 USB SDP
2.12 5.12502 0.55685 2.85387 0.033 0.058 27.2 USB SDP
2.16 5.12502 0.58623 3.00444 0.032 0.058 27.2 USB SDP
2.20 5.12502 0.55685 2.85387 0.033 0.058 27.2 USB SDP
2.24 5.12502 0.55696 2.85443 0.033 0.058 27.2 USB SDP
2.28 5.12502 0.55696 2.85443 0.033 0.058 27.2 USB SDP
2.32 5.12502 0.60834 3.11775 0.033 0.058 27.2 USB SDP
2.36 5.12502 0.55713 2.85530 0.033 0.058 27.2 USB SDP
2.40 5.12502 0.55713 2.85530 0.033 0.058 27.2 USB SDP
2.44 5.12498 0.55713 2.85528 0.033 0.058 27.2 USB SDP
2.48 5.12498 0.60288 3.08975 0.032 0.058 27.2 USB SDP
2.52 5.12491 0.55741 2.85668 0.032 0.057 27.2 USB SDP
2.56 5.12491 0.55741 2.85668 0.033 0.058 27.2 USB SDP
2.60 5.12491 0.55741 2.85668 0.033 0.058 27.2 USB SDP
2.64 5.12491 0.59170 3.03241 0.033 0.058 27.2 USB SDP
2.68 5.12491 0.55692 2.85416 0.033 0.058 27.2 USB SDP
2.72 5.12491 0.55665 2.85278 0.033 0.058 27.2 USB SDP
2.76 5.12491 0.55665 2.85278 0.033 0.058 27.2 USB SDP
2.80 5.12491 0.59642 3.05660 0.033 0.058 27.2 USB SDP
2.84 5.12491 0.55650 2.85201 0.033 0.058 27.2 USB SDP
2.88 5.12491 0.55650 2.85201 0.033 0.058 27.2 USB SDP
2.92 5.12483 0.55670 2.85299 0.033 0.058 27.2 USB SDP
2.96 5.12483 0.59418 3.04507 0.033 0.058 27.2 USB SDP
3.00 5.12483 0.56909 2.91649 0.033 0.058 27.2 USB SDP
DEBUG: sending POLL
DEBUG: POLL write result = True
3.04 5.12471 0.55696 2.85426 0.033 0.058 27.2 USB SDP
3.08 5.12471 0.55696 2.85426 0.033 0.058 27.2 USB SDP
3.12 5.12463 0.57977 2.97111 0.033 0.058 27.2 USB SDP
3.16 5.12463 0.55610 2.84981 0.033 0.058 27.2 USB SDP
3.20 5.12463 0.55610 2.84981 0.033 0.058 27.2 USB SDP
3.24 5.12459 0.55599 2.84922 0.033 0.058 27.2 USB SDP
3.28 5.12459 0.58946 3.02074 0.032 0.058 27.2 USB SDP
3.32 5.12448 0.55615 2.84998 0.033 0.058 27.2 USB SDP
3.36 5.12448 0.55615 2.84998 0.033 0.058 27.2 USB SDP
3.40 5.12448 0.55615 2.84998 0.032 0.058 27.2 USB SDP
3.44 5.12444 0.57903 2.96720 0.033 0.058 27.2 USB SDP
3.48 5.12444 0.56686 2.90484 0.033 0.058 27.2 USB SDP
3.52 5.12444 0.57282 2.93538 0.033 0.058 27.2 USB SDP
3.56 5.12444 0.57034 2.92267 0.033 0.058 27.2 USB SDP
3.60 5.12444 0.58002 2.97228 0.033 0.058 27.2 USB SDP
3.64 5.12444 0.55578 2.84806 0.032 0.058 27.2 USB SDP
3.68 5.12444 0.55578 2.84806 0.032 0.057 27.2 USB SDP
3.72 5.12428 0.55568 2.84746 0.033 0.058 27.2 USB SDP
3.76 5.12428 0.57853 2.96455 0.032 0.057 27.2 USB SDP
3.80 5.12428 0.55568 2.84746 0.033 0.058 27.2 USB SDP
3.84 5.12420 0.55578 2.84793 0.033 0.058 27.2 USB SDP
3.88 5.12420 0.57158 2.92889 0.033 0.058 27.2 USB SDP
3.92 5.12424 0.58325 2.98871 0.032 0.057 27.2 USB SDP
3.96 5.12424 0.57257 2.93399 0.033 0.058 27.2 USB SDP
4.00 5.12424 0.57804 2.96202 0.033 0.058 27.2 USB SDP
DEBUG: sending POLL
DEBUG: POLL write result = True
The FNB58 is sampling at 100Hz, so when we poll at 1Hz we get 25 packets with 4 samples per packet. The printout is only showing the first sample in each packet.
So, the stock program is working well. When moving on to creating a standalone logging program I ran into problems trying to reduce the amount of data being ingested. When I increased the polling interval to 2 or higher, the polling interface would hang. After chasing this for a while I decided to move on and leave the polling interval at 1 second.
So, after a few iterations deciding what data and format that I wanted - here is the first "release" of the logging program cm5_power_temp_logger.py.
#!/usr/bin/env python3
from datetime import datetime
from pathlib import Path
import subprocess
import time
import sys
import os
import csv
import select
import struct
import argparse
# Import FNB58 protocol definitions (requires fnb58.py in the same directory)
from fnb58 import (
CMD_INIT1,
CMD_INIT2,
CMD_POLL,
open_hid,
hid_write,
hid_drain,
infer_protocol,
)
# --- FNB58 Helper Functions ---
def decode_packet(data):
"""Decode a 64-byte FNB58 data packet."""
if len(data) < 62:
return None
if data[0] != 0xAA or data[1] != 0x04:
return None
samples = []
for i in range(4):
off = 2 + 15 * i
voltage = struct.unpack_from("<I", data, off)[0] / 100000.0
current = struct.unpack_from("<I", data, off + 4)[0] / 100000.0
dp = struct.unpack_from("<H", data, off + 8)[0] / 1000.0
dn = struct.unpack_from("<H", data, off + 10)[0] / 1000.0
temp = struct.unpack_from("<H", data, off + 13)[0] / 10.0
samples.append({
"voltage_V": voltage,
"current_A": current,
"power_W": voltage * current,
"dp_V": dp,
"dn_V": dn,
"temp_C": temp,
"protocol": infer_protocol(dp, dn, voltage),
})
return samples
def hid_read(fd, timeout=2.0):
"""Read one HID report with a timeout."""
r, _, _ = select.select([fd], [], [], timeout)
if not r:
return None
try:
return os.read(fd, 64)
except OSError:
return None
def initialize_fnb58(fd):
"""Initialize the FNB58 using the known-working sequence."""
hid_drain(fd)
if not hid_write(fd, CMD_INIT1):
return False
time.sleep(0.05)
if not hid_write(fd, CMD_INIT2):
return False
time.sleep(0.05)
if not hid_write(fd, CMD_INIT2):
return False
return True
def get_fnb58_packet(fd, deadline):
"""Wait for a valid FNB58 data packet until deadline."""
while time.time() < deadline:
remaining = deadline - time.time()
data = hid_read(fd, timeout=min(1.0, remaining))
if data is None:
continue
samples = decode_packet(data)
if samples:
return samples
return None
# --- CM5 System / Thermal Helper Functions ---
def read_file(path):
"""Safely read a file's contents, returning an empty string if it fails."""
try:
return Path(path).read_text().strip()
except Exception:
return ""
def get_throttled():
"""Run vcgencmd to get the throttled status."""
try:
res = subprocess.run(
["vcgencmd", "get_throttled"],
capture_output=True,
text=True,
check=True,
)
return res.stdout.strip().split("=")[1]
except Exception:
return ""
def get_cm5_metrics():
"""Gather all CM5 system metrics."""
# 1. Temperature (C)
temp_raw = read_file("/sys/class/thermal/thermal_zone0/temp")
try:
temp = f"{float(temp_raw) / 1000:.2f}" if temp_raw else ""
except ValueError:
temp = ""
# 2. CPU Frequency (MHz)
freq_raw = read_file("/sys/devices/system/cpu/cpu0/cpufreq/scaling_cur_freq")
try:
freq = f"{float(freq_raw) / 1000:.0f}" if freq_raw else ""
except ValueError:
freq = ""
# 3. Fan PWM and RPM
pwm = read_file("/sys/devices/platform/cooling_fan/hwmon/hwmon2/pwm1")
rpm = read_file("/sys/devices/platform/cooling_fan/hwmon/hwmon2/fan1_input")
# 4. Throttling status
throttled = get_throttled()
# 5. 1-minute CPU load average
load_raw = read_file("/proc/loadavg")
load = load_raw.split()[0] if load_raw else ""
return temp, freq, pwm, rpm, throttled, load
# --- Main Execution ---
def main():
parser = argparse.ArgumentParser(description="Combined CM5 System & FNB58 Power Logger")
parser.add_argument("--usb", action="store_true", default=True, help="Use USB HID interface (default)")
parser.add_argument("--interval", type=float, default=1.0, help="Seconds between polls (default: 1)")
parser.add_argument("--count", type=int, default=None, help="Number of measurements, or run until Ctrl-C")
parser.add_argument("--output", type=str, default=None, help="Custom CSV output filename")
args = parser.parse_args()
if not args.usb:
print("Error: --usb flag is required to talk to the FNB58 meter.", file=sys.stderr)
sys.exit(1)
# Setup directories and default output filename if not provided
log_dir = Path.home() / "cm5_benchmark"
log_dir.mkdir(parents=True, exist_ok=True)
if args.output:
log_file = Path(args.output)
else:
timestamp_str = datetime.now().strftime("%Y%m%d_%H%M%S")
log_file = log_dir / f"combined_benchmark_{timestamp_str}.csv"
# Open FNB58 HID device
fd = open_hid("/dev/hidraw0")
if fd < 0:
print("Unable to open FNB58 HID device at /dev/hidraw0.", file=sys.stderr)
sys.exit(1)
if not initialize_fnb58(fd):
print("Failed to initialize FNB58.", file=sys.stderr)
os.close(fd)
sys.exit(1)
start_time = time.time()
next_poll = start_time + args.interval
count = 0
# Write combined CSV header
header = [
"timestamp",
"elapsed_s",
"voltage_V",
"current_A",
"power_W",
"cm5_temp_C",
"cpu_freq_MHz",
"fan_rpm",
"throttled",
"cpu_load_1min"
]
with open(log_file, "w", newline="") as csvfile:
writer = csv.writer(csvfile)
writer.writerow(header)
csvfile.flush()
print("Combined CM5 & FNB58 Logger Started")
print(f"Logging to: {log_file}")
print(f"Interval: {args.interval} seconds")
print("Press Ctrl-C to stop.\n")
try:
while args.count is None or count < args.count:
now = time.time()
if now < next_poll:
time.sleep(next_poll - now)
poll_time = time.time()
elapsed = poll_time - start_time
# 1. Gather CM5 Metrics
cm5_temp, freq, pwm, rpm, throttled, load = get_cm5_metrics()
# 2. Gather FNB58 Power Metrics
if not hid_write(fd, CMD_POLL):
print("WARNING: FNB58 POLL write failed", file=sys.stderr)
next_poll = poll_time + args.interval
continue
samples = get_fnb58_packet(fd, poll_time + 1.5)
timestamp = time.strftime("%Y-%m-%d %H:%M:%S", time.localtime(poll_time))
if samples is None:
print(f"{elapsed:8.1f}s WARNING: FNB58 no response (CM5 logged)", file=sys.stderr)
# Log row with empty power values if meter drops out
writer.writerow([
timestamp, f"{elapsed:.2f}", "", "", "", cm5_temp, freq, rpm,
throttled, load
])
else:
s = samples[0]
# Write full combined row
writer.writerow([
timestamp,
f"{elapsed:.2f}",
f"{s['voltage_V']:.5f}",
f"{s['current_A']:.5f}",
f"{s['power_W']:.5f}",
cm5_temp,
freq,
rpm,
throttled,
load
])
# Print combined quick view to terminal
print(
f"{elapsed:6.1f}s | Power: {s['voltage_V']:.2f}V {s['current_A']:.2f}A {s['power_W']:.2f}W | "
f"CM5: {cm5_temp:>5}C {freq:>4}MHz {rpm:>4}RPM"
)
csvfile.flush()
count += 1
next_poll = poll_time + args.interval
except KeyboardInterrupt:
print("\nLogging stopped by user.")
finally:
os.close(fd)
if __name__ == "__main__":
main()
It is about a third of the length of fnb58.py. I'll need to circle back and add comments and options as required.
Here is a test run for about 30 seconds with the system at idle:
The printed output is missing a few parameters on purpose.
Here is the CSV file that is created:
As a final check I ran another stress test, this time including memory stress and running for 30 minutes (logger runs for 45).
.
The data plots from the CSV file.
At this point I'm ready to start trying different performance benchmarks.