树莓派5+Hailo实时推理存在旧数据积压延迟问题的解决及代码修改咨询

看起来你的问题核心是推理速度跟不上数据采集速度，导致旧的batch不断积压，系统一直在处理滞后的数据，无法响应最新的实时数据。本质原因是当前代码会响应每一次新数据的触发信号，逐个处理所有生成的batch，但我们其实只需要关注当前时刻缓冲区里最新的那一批200个数据，中间的旧batch都可以直接丢弃。

核心解决方案：只处理最新的Batch，丢弃积压请求 #

我们的目标是让系统永远只处理「当前时刻缓冲区里最新的200个数据」，完全忽略中间积压的推理请求。下面是具体的实现步骤和代码修改：

一、关键问题分析 #

原代码中，new_data_flag会在每新增一个数据点时被设置，而主线程处理一个batch的时间里，可能已经有大量新数据点进入，导致flag被多次触发。主线程处理完一个batch后会立刻响应下一次触发，进而处理的是稍旧的batch，最终形成延迟积压。

二、具体修改方案 #

方案1：优化信号触发逻辑（保留实时性） #

通过清空积压的触发信号，确保每次只处理当前最新的batch，同时保留信号触发的及时性：

#!/usr/bin/env python3
import threading
import time
import argparse
import numpy as np
import serial
import csv
import lgpio
from hailo_platform import (HEF, Device, VDevice, HailoStreamInterface, ConfigureParams,
                            InputVStreamParams, OutputVStreamParams, InputVStreams, OutputVStreams, FormatType)

TRIGGER_PIN = 25
OUTPUT_PIN = 24
h = lgpio.gpiochip_open(0)
lgpio.gpio_claim_input(h, TRIGGER_PIN)
lgpio.gpio_claim_output(h, OUTPUT_PIN)
ser = serial.Serial('/dev/ttyAMA0', 1562500, timeout=0.01, parity=serial.PARITY_NONE,
                    stopbits=serial.STOPBITS_ONE, bytesize=serial.EIGHTBITS)

time.sleep(0.1)

parser = argparse.ArgumentParser(description='Run Hailo inference using SCI communication.')
parser.add_argument('hef', help="HEF file path")
parser.add_argument('--data-csv', default="received_data.csv", help="Path to save the last received data batch.")
args = parser.parse_args()

DATA_BUFFER_SIZE = 200
CIRCULAR_BUFFER_CAPACITY = 2000
circular_buffer = np.zeros(CIRCULAR_BUFFER_CAPACITY, dtype=np.float32)
write_index = 0
data_count = 0
received_data_count = 0
new_data_flag = threading.Event()
terminate_flag = False

def post_processing(inference_output):
    class_probabilities = inference_output[0]
    predicted_class = np.argmax(class_probabilities)
    confidence = class_probabilities[predicted_class]
    return predicted_class, confidence

def synchronous_inference(network_group, batch_data):
    input_vstreams_params = InputVStreamParams.make_from_network_group(network_group, quantized=False, format_type=FormatType.FLOAT32)
    output_vstreams_params = OutputVStreamParams.make_from_network_group(network_group, quantized=False, format_type=FormatType.FLOAT32)

    with InputVStreams(network_group, input_vstreams_params) as input_vstreams, \
         OutputVStreams(network_group, output_vstreams_params) as output_vstreams:

        for vstream in input_vstreams:
            formatted_data = np.expand_dims(batch_data, axis=0).astype(np.float32)
            vstream.send(formatted_data)

        inference_dict = {}
        for vstream in output_vstreams:
            received_data = vstream.recv()
            inference_dict[vstream.name] = received_data

        inference_output = list(inference_dict.values())
        return post_processing(inference_output)

def save_data_to_csv(all_batches, output_csv_path):
    if all_batches:
        with open(output_csv_path, mode='w', newline='') as file:
            writer = csv.writer(file)
            for batch in all_batches:
                flattened_batch = [item for sublist in batch for subsublist in sublist for item in subsublist]
                writer.writerow(flattened_batch)

def collect_data():
    global write_index, data_count, received_data_count, terminate_flag
    last_gpio_state = 0

    while not terminate_flag:
        gpio_state = lgpio.gpio_read(h, TRIGGER_PIN)
        if last_gpio_state == 0 and gpio_state == 1:
            if ser.in_waiting > 0:
                received_data = ser.read(2)
                received_value = (int.from_bytes(received_data, byteorder='little') * 0.001)

                circular_buffer[write_index] = received_value
                write_index = (write_index + 1) % CIRCULAR_BUFFER_CAPACITY
                data_count = min(data_count + 1, CIRCULAR_BUFFER_CAPACITY)
                received_data_count += 1
                # 优化：仅当数据量足够一个batch时才触发信号，减少无效触发
                if data_count >= DATA_BUFFER_SIZE:
                    new_data_flag.set()

                if received_data_count >= 10000:
                    terminate_flag = True
                    break

        last_gpio_state = gpio_state

def get_recent_data(batch_size):
    global write_index, data_count
    if data_count < batch_size:
        return None
    # 确保取到绝对最新的batch，处理循环缓冲区的 wrap-around 情况
    start_index = (write_index - batch_size) % CIRCULAR_BUFFER_CAPACITY
    if start_index + batch_size <= CIRCULAR_BUFFER_CAPACITY:
        return circular_buffer[start_index:start_index + batch_size]
    else:
        return np.concatenate([circular_buffer[start_index:], circular_buffer[:start_index + batch_size - CIRCULAR_BUFFER_CAPACITY]])

try:
    hef = HEF(args.hef)
    devices = Device.scan()
    with VDevice(device_ids=devices) as target:
        configure_params = ConfigureParams.create_from_hef(hef, interface=HailoStreamInterface.PCIe)
        network_group = target.configure(hef, configure_params)[0]

        with network_group.activate():
            network_group.wait_for_activation(100)

            all_batches = []

            data_thread = threading.Thread(target=collect_data, daemon=True)
            data_thread.start()

            while not terminate_flag:
                new_data_flag.wait()
                # 清空所有积压的触发信号，确保只处理当前最新的batch
                while new_data_flag.is_set():
                    new_data_flag.clear()

                lgpio.gpio_write(h, OUTPUT_PIN, lgpio.LOW)

                recent_data = get_recent_data(DATA_BUFFER_SIZE)
                if recent_data is None or recent_data.size != DATA_BUFFER_SIZE:
                    continue

                batch_data = recent_data.reshape((10, 20, 1))
                all_batches = [batch_data]  # 只保留最新的batch，丢弃旧的

                predicted_class, confidence = synchronous_inference(network_group, batch_data)

                if predicted_class == 0:
                    lgpio.gpio_write(h, OUTPUT_PIN, lgpio.LOW)
                else:
                    lgpio.gpio_write(h, OUTPUT_PIN, lgpio.HIGH)

            save_data_to_csv(all_batches, args.data_csv)

except KeyboardInterrupt:
    pass
finally:
    terminate_flag = True
    data_thread.join()
    ser.close()
    lgpio.gpiochip_close(h)

方案2：定期轮询取最新Batch（更简单可控） #

如果你的实时性要求允许小的固定延迟，可以去掉信号触发逻辑，让主线程定期检查并取最新的batch，完全避免信号积压：

# 仅修改主线程部分，其余代码同方案1
try:
    hef = HEF(args.hef)
    devices = Device.scan()
    with VDevice(device_ids=devices) as target:
        configure_params = ConfigureParams.create_from_hef(hef, interface=HailoStreamInterface.PCIe)
        network_group = target.configure(hef, configure_params)[0]

        with network_group.activate():
            network_group.wait_for_activation(100)

            all_batches = []

            data_thread = threading.Thread(target=collect_data, daemon=True)
            data_thread.start()

            # 轮询间隔，根据实时需求调整（示例为50ms）
            POLL_INTERVAL = 0.05
            while not terminate_flag:
                time.sleep(POLL_INTERVAL)
                
                lgpio.gpio_write(h, OUTPUT_PIN, lgpio.LOW)

                recent_data = get_recent_data(DATA_BUFFER_SIZE)
                if recent_data is None or recent_data.size != DATA_BUFFER_SIZE:
                    continue

                batch_data = recent_data.reshape((10, 20, 1))
                all_batches = [batch_data]  # 只保留最新的batch

                predicted_class, confidence = synchronous_inference(network_group, batch_data)

                if predicted_class == 0:
                    lgpio.gpio_write(h, OUTPUT_PIN, lgpio.LOW)
                else:
                    lgpio.gpio_write(h, OUTPUT_PIN, lgpio.HIGH)

            save_data_to_csv(all_batches, args.data_csv)
except KeyboardInterrupt:
    pass
finally:
    terminate_flag = True
    data_thread.join()
    ser.close()
    lgpio.gpiochip_close(h)

三、额外优化建议 #

1. 模型量化优化：将HEF模型量化为INT8/INT4格式，Hailo对量化模型的推理速度会提升数倍，同时修改输入输出的FormatType为量化格式，减少数据转换开销。
2. 异步推理优化：如果Hailo支持异步推理，可以将推理任务异步执行，并在新batch到来时取消旧的推理任务，直接执行最新的，进一步降低延迟。
3. 采集线程优化：确保串口采集代码足够高效，避免因为采集线程阻塞导致的数据延迟。

备注：内容来源于stack exchange，提问作者Bijen Mali