The Latency Bottleneck in Edge Vision
Running object detection models like YOLOv8 on edge hardware requires careful memory and frame-rate management. Processing every single frame from a camera stream often saturates the CPU/GPU without adding meaningful diagnostic value—especially when monitoring slow-changing physical environments like plant health or industrial gauges.
In this guide, we will implement a frame-sampling and asynchronous inference wrapper in Python to keep latency low and resource usage predictable.
Designing a Non-Blocking Inference Queue
Instead of running model inference directly inside the camera capture loop, we decouple frame capture from model execution using a bounded queue.Queue of size 1. This ensures the model always evaluates the most recent frame while dropping stale frames automatically.
import threading
import queue
import time
class EdgeInferencePipeline:
def __init__(self, model, target_fps: float = 2.0):
self.model = model
self.interval = 1.0 / target_fps
self.frame_queue = queue.Queue(maxsize=1)
self.running = False
def push_frame(self, frame):
if self.frame_queue.full():
try:
self.frame_queue.get_nowait()
except queue.Empty:
pass
self.frame_queue.put(frame)
def _worker(self):
while self.running:
start_time = time.perf_counter()
try:
frame = self.frame_queue.get(timeout=1.0)
results = self.model.predict(frame, verbose=False)
self.handle_detections(results)
except queue.Empty:
continue
elapsed = time.perf_counter() - start_time
sleep_time = max(0.0, self.interval - elapsed)
time.sleep(sleep_time)
def handle_detections(self, results):
for r in results:
boxes = r.boxes
print(f"Detected {len(boxes)} objects at {time.strftime('%X')}")
def start(self):
self.running = True
threading.Thread(target=self._worker, daemon=True).start()
def stop(self):
self.running = False
Adding Basic Telemetry Metrics
To monitor how healthy our inference service is in production, we should track average inference duration and frame drop rates:
def measure_inference(model, sample_image):
t0 = time.perf_counter()
_ = model.predict(sample_image, verbose=False)
latency_ms = (time.perf_counter() - t0) * 1000
print(f"Inference latency: {latency_ms:.2f} ms")
return latency_ms
Conclusion
Decoupling frame capture from model inference prevents memory leaks and keeps edge hardware responsive. Combined with lightweight latency logging, you can reliably deploy computer vision models on local hardware without cloud round-trip delays.
Top comments (2)
Nice pattern: a queue of one is the right default when only the latest frame matters. Two things I'd add to the telemetry part, since the drop rate you mention isn't counted yet. First, increment a counter in push_frame every time it discards a frame, and log it per minute next to latency. With a queue of one, a slower model never shows up as backlog or memory growth, only as more frames dropped, so that counter is the number that tells you the device is falling behind.
Second, measure once the device is warm, not on one call. On a Pi 5 with a fan heatsink, YOLO26n on ONNX Runtime in a single process hit the soft temperature limit at 7.8 minutes and throttled at 17.5, and while the real clock dropped to 1,500 MHz, Linux kept reporting 2,400. A single measure_inference call at startup would never see that. I'd log p50 and p95 per minute for the first half hour. Have you run it long enough on your hardware to see whether the latency stays flat?
Thanks a lot for the detailed feedback, Israel! You are spot on—with a bounded queue of size 1, silent frame drops are the real indicator of backpressure, so tracking a drop counter per minute alongside p50/p95 latency makes total sense.
In my current deployment (a hydroponic greenhouse monitoring system), I avoid thermal throttling by running event-driven and scheduled snapshot captures rather than continuous 24/7 video streaming, which keeps the hardware cool and latency flat over long periods. However, adding a
dropped_framesmetric and running a 30-minute thermal soak test with p50/p95 logging is a great addition for continuous-stream setups. Appreciate you sharing those real-world Raspberry Pi 5 benchmarks!