[Python] Why is my inference process operating on outdated data instead of real-time data,...

I have a Python script running real-time data collection and inference using Hailo on a Raspberry Pi 5. The system uses a circular buffer to collect sensor data, processes the most recent data batch (size 200), and performs inference on it. However, the inference is often performed on old data instead of the latest real-time data, causing significant delays. i.e. the inference is performed for all the data without skipping. This creates a backlog of data batches for which inference is to be performed, creating incremental inference delay as time goes on. How can I remedy it?

How can I modify my code to ensure that inference is always performed only on the latest available data, ignoring any backlog?

Code:

#!/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
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
start_index = (write_index - batch_size) % CIRCULAR_BUFFER_CAPACITY
return circular_buffer[start_index:start_index + batch_size]

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()
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 != 200:
continue

batch_data = recent_data.reshape((10, 20, 1))

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)



I was originally trying a simple data buffer but then I tried using the circular data buffer. I also used get_recent_data(batch_size) to ensure that inference is performed only on the real-time data.

Continue reading...