Raspberry Pi CM5 - Frigate NVR Hailo-8L Benchmarks

Hailo-8L Thermal Logging

The Hailo-8L module that I have is an extended temperature version with an industrial operating temperature range (-40C to 85C).

Part Number: HM21LB1C2LAE
Product Name: HAILO-8L AI ACC M.2 B+M KEY MODULE EXT TMP

It has two temperature zones, TS0 and TS1, which are monitored by on-die thermal sensors.  TS0 (Thermal Sensor 0) is positioned near the core logic processing clusters and TS1 (Thermal Sensor 1) is positioned near the memory interface / system peripheral blocks.

Operational Thermal Thresholds

Operating Zone   Internal Junction   Temp Range Behavior
Green Zone          Under 104°C            Normal operation; no throttling occurs.
Orange Zone        104°C to 119°C        Frequency throttling active; performance automatically scales down dynamically by up to 50% to prevent burnout.
Red Zone             120°C and Above     Emergency shutdown; all input/output media streams are instantly killed by the hardware firmware driver.

My initial attempt at logging the Hailo temperature sensors failed because Frigate takes exclusive control of the Hailo device, so that you cannot access it when Frigate is running.  So, the method needs to query the pyhailort within the Frigate container using the docker interface.

Here is the updated logging program:

cm5_power_temp_logger_hailo.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 glob

# 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,
)

# HailoRT Python binding
try:
    from hailo_platform.pyhailort import _pyhailort
    HAILO_AVAILABLE = True
except ImportError:
    _pyhailort = None
    HAILO_AVAILABLE = False


# ------------------------------------------------------------
# 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."""
    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 CM5 system metrics."""

    # 1. CPU temperature
    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
    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
    # Find the hwmon device named "pwmfan" dynamically.
    pwm = ""
    rpm = ""

    for hwmon in Path("/sys/class/hwmon").glob("hwmon*"):
        name = read_file(hwmon / "name")
        if name == "pwmfan":
            pwm = read_file(hwmon / "pwm1")
            rpm = read_file(hwmon / "fan1_input")
            break

    # 4. Throttling
    throttled = get_throttled()

    # 5. 1-minute load average
    load_raw = read_file("/proc/loadavg")
    load = load_raw.split()[0] if load_raw else ""

    return temp, freq, pwm, rpm, throttled, load


# ------------------------------------------------------------
# Hailo Temperature
# ------------------------------------------------------------

def get_hailo_temperature():
    """Read Hailo-8L temperatures through the running Frigate container."""
    try:
        cmd = [
            "docker", "exec", "frigate-hailo423",
            "python3", "-c",
            (
                "from hailo_platform.pyhailort import _pyhailort; "
                "ids=_pyhailort.Device.scan(); "
                "d=_pyhailort.Device.create(ids[0]); "
                "t=d.get_chip_temperature(); "
                "print(f'{t.ts0_temperature:.2f},{t.ts1_temperature:.2f}'); "
                "d.release()"
            ),
        ]

        result = subprocess.run(
            cmd,
            capture_output=True,
            text=True,
            timeout=2,
            check=True,
        )

        line = result.stdout.strip().splitlines()[-1]
        ts0, ts1 = line.split(",")

        return float(ts0), float(ts1)

    except Exception:
        return None, None


# ------------------------------------------------------------
# Main Execution
# ------------------------------------------------------------

def main():

    parser = argparse.ArgumentParser(
        description="Combined CM5, Hailo & FNB58 Power/Thermal 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)

    # --------------------------------------------------------
    # Output file
    # --------------------------------------------------------

    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
    # --------------------------------------------------------

    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)

    # --------------------------------------------------------
    # Timing
    # --------------------------------------------------------

    start_time = time.time()
    next_poll = start_time + args.interval
    count = 0

    # --------------------------------------------------------
    # CSV header
    # --------------------------------------------------------

    header = [
        "timestamp",
        "elapsed_s",

        "voltage_V",
        "current_A",
        "power_W",

        "cm5_temp_C",
        "cpu_freq_MHz",
        
        "fan_rpm",

        "throttled",
        "cpu_load_1min",
        
        "hailo_ts0_C",
        "hailo_ts1_C"
    ]

    # --------------------------------------------------------
    # Open CSV
    # --------------------------------------------------------

    with open(log_file, "w", newline="") as csvfile:

        writer = csv.writer(csvfile)

        writer.writerow(header)
        csvfile.flush()

        print()
        print("Combined CM5 + Hailo + FNB58 Logger Started")
        print(f"Logging to: {log_file}")
        print(f"Interval:   {args.interval} seconds")
        print()
        print("Press Ctrl-C to stop.")
        print()

        try:

            while (
                args.count is None
                or count < args.count
            ):

                # --------------------------------------------
                # Maintain regular sampling interval
                # --------------------------------------------

                now = time.time()

                if now < next_poll:
                    time.sleep(next_poll - now)

                poll_time = time.time()

                elapsed = poll_time - start_time

                # --------------------------------------------
                # CM5 metrics
                # --------------------------------------------

                (
                    cm5_temp,
                    freq,
                    pwm,
                    rpm,
                    throttled,
                    load
                ) = get_cm5_metrics()

                # --------------------------------------------
                # Hailo temperature
                # --------------------------------------------

                (
                    hailo_ts0,
                    hailo_ts1
                ) = get_hailo_temperature()

                # --------------------------------------------
                # FNB58
                # --------------------------------------------

                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)
                )

                # --------------------------------------------
                # FNB58 failed
                # --------------------------------------------

                if samples is None:

                    print(
                        f"{elapsed:8.1f}s  "
                        "WARNING: FNB58 no response "
                        "(CM5/Hailo logged)",
                        file=sys.stderr
                    )

                    writer.writerow([
                        timestamp,
                        f"{elapsed:.2f}",

                        "",
                        "",
                        "",

                        cm5_temp,

                        freq,
                        rpm,

                        throttled,
                        load,
                        
                        hailo_ts0 if hailo_ts0 is not None else "",
                        hailo_ts1 if hailo_ts1 is not None else ""                        
                        
                    ])

                # --------------------------------------------
                # Normal sample
                # --------------------------------------------

                else:

                    # Use first FNB58 sample, matching your
                    # existing logger.
                    s = samples[0]

                    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,
                        pwm,

                        throttled,
                        load,
                        
                        hailo_ts0 if hailo_ts0 is not None else "",
                        hailo_ts1 if hailo_ts1 is not None else ""
                    ])

                    # ----------------------------------------
                    # Terminal display
                    # ----------------------------------------

                    hailo_display = (
                        f"{hailo_ts0:.2f}C / {hailo_ts1:.2f}C"
                        if hailo_ts0 is not None
                    else "N/A"
                    )

                    print(
                        f"{elapsed:6.1f}s | "
                        f"Power: "
                        f"{s['voltage_V']:.2f}V "
                        f"{s['current_A']:.2f}A "
                        f"{s['power_W']:.2f}W | "
                        f"CM5: "
                        f"{cm5_temp:>5}C "
                        f"{freq:>4}MHz "
                        f"{rpm:>5}RPM | "
                        f"Hailo: {hailo_display}"
                    )

                csvfile.flush()

                count += 1

                next_poll = poll_time + args.interval

        except KeyboardInterrupt:

            print("\nLogging stopped by user.")

        finally:

            # Close FNB58
            os.close(fd)


if __name__ == "__main__":
    main()

And the logger output:

You'll notice that for the first couple of seconds that the Hailo temperatures are not available.  That's because I had started the logger before Frigate had finished starting.

Comparison Data for Different modes

 A quick jump up:

• The Hailo-8L temperature isn't an issue in this configuration, even without a heatsink. Temperatures are around 40 C.
• The CM5 temperatures are also reduced due to offloading of inference to the Hailo.  The active cooler fan stays in the second operating zone (3500 rpm).
• Detection_FPS (inference) is able to scale as cameras are added.  The CPU detection limited it to 15 FPS.
• Skipped_FPS only becomes an issue with 8 cameras.
• There is some anomalous behavior with the CPU_Load (1 minute average) numbers from Linux.  I'll need to rerun some of the tests to see if I can get repeatability.  It might be caused by the asynchronous camera processing and inference threads and the other processes running on the CM5.  In the 8 camera test, the load numbers increase rapidly about 8 minutes into the run.  If that is real, that mode may not be usable.

Frigate Hailo-8L Detection 8 camera Benchmark

This is my final post before the Roadtest Review.  I'll need to rerun some of the Hailo camera tests and try to run some storage benchmarks.