Skip to content

Fix closed-loop latency instrumentation and two live-mode freezes #21

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
MarkSant merged 2 commits into from
Aug 8, 2026
Merged

Fix closed-loop latency instrumentation and two live-mode freezes #21

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
d8d66f4
Fix closed-loop latency instrumentation and two live-mode freezes
Aug 8, 2026
d906120
Address review: session teardown race, unlogged timeouts, 0.0 timestamps
Aug 8, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
85 changes: 63 additions & 22 deletions src/zebtrack/core/controller.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -17,10 +17,12 @@
import logging as structlog # type: ignore
structlog.get_logger = lambda *a, **k: structlog.getLogger("zebtrack")

from zebtrack import latency_logging
from zebtrack.core.project_manager import ProjectManager
from zebtrack.io.arduino import Arduino
from zebtrack.io.camera import Camera
from zebtrack.io.recorder import Recorder
from zebtrack.settings import settings
from zebtrack.ui.gui import ApplicationGUI

log = structlog.get_logger()
Expand All @@ -34,8 +36,6 @@ def __init__(self, root):
self.view = ApplicationGUI(root, self)

# Backend modules
from zebtrack.settings import settings

self.project_manager = ProjectManager()
self.recorder = Recorder()
self.arduino = Arduino(
Expand Down Expand Up @@ -92,27 +92,27 @@ def on_close(self):
self.root.destroy()
log.info("application.shutdown.complete")

# No join here may be unbounded. Every one of these threads can block in a
# third-party call (OpenCV capture, OpenVINO inference, pyserial), and an
# unbounded join turns one stalled worker into a frozen application with no
# diagnostic. All threads are daemons, so a straggler cannot keep the
# process alive; it only gets named in the log.
_JOIN_TIMEOUT_S = 5

def join_threads(self):
"""Waits for all core threads to finish."""
"""Waits (with a bound) for all core threads to finish."""
log.info("threads.join.start")
if (
hasattr(self, "capture_thread")
and self.capture_thread
and self.capture_thread.is_alive()
):
self.capture_thread.join()
if (
hasattr(self, "processing_thread")
and self.processing_thread
and self.processing_thread.is_alive()
):
self.processing_thread.join()
if (
hasattr(self, "video_thread")
and self.video_thread
and self.video_thread.is_alive()
):
self.video_thread.join(timeout=5)
for name in ("capture_thread", "processing_thread", "video_thread"):
thread = getattr(self, name, None)
if not thread or not thread.is_alive():
continue
thread.join(timeout=self._JOIN_TIMEOUT_S)
if thread.is_alive():
log.error(
"threads.join.timeout",
thread=name,
timeout_s=self._JOIN_TIMEOUT_S,
)
log.info("threads.join.finished")

def stop_recording(self):
Expand All @@ -125,6 +125,14 @@ def stop_recording(self):
self.video_thread.join(timeout=5)

self.recorder.stop_recording()
latency_logging.stop_session(
{
"fps_measured_at_stop": (
self.camera.measured_fps() if self.camera is not None else None
),
"video_fps_written": getattr(self.recorder, "video_fps", None),
}
)

self.view.update_button_state("start_rec", "normal")
self.view.update_button_state("stop_rec", "disabled")
Expand Down Expand Up @@ -201,11 +209,44 @@ def start_recording(self):
)

cam_props = self.camera.get_properties()
# Stamp the container with the rate the camera actually achieves. The
# configured value is a request; on this rig the two differ by ~30%,
# which would make every frame->millisecond conversion wrong.
measured_fps = cam_props.get("fps_measured")
success = self.recorder.start_recording(
output_folder, cam_props["width"], cam_props["height"]
output_folder,
cam_props["width"],
cam_props["height"],
fps=measured_fps,
)

if success:
latency_logging.start_session(
output_folder,
os.path.basename(output_folder),
meta={
"system": "PyZebArdYolo",
"session_id": os.path.basename(output_folder),
"camera_index": settings.camera.index,
"frame_width": cam_props["width"],
"frame_height": cam_props["height"],
"fps_configured": settings.video_processing.fps,
"fps_measured_at_start": measured_fps,
# Live mode analyses every queued frame; the configured
# processing_interval applies only to pre-recorded video.
"analysis_interval_frames": 1,
"arduino_port": settings.arduino.port,
"baud_rate": settings.arduino.baud_rate,
"detector_plugin": type(self.detector.plugin).__name__
if self.detector is not None
else None,
"confidence_threshold": settings.yolo_model.confidence_threshold,
"detection_squares": settings.detection_zones.squares,
"enter_commands": settings.detection_zones.enter_commands,
"exit_commands": settings.detection_zones.exit_commands,
"roi_convention": "bbox_corner",
},
)
with self.frame_queue.mutex:
self.frame_queue.queue.clear()
with self.video_queue.mutex:
Expand Down
22 changes: 22 additions & 0 deletions src/zebtrack/core/detector.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -42,6 +42,9 @@ def __init__(self, plugin: DetectorPlugin):
)
self.base_squares = settings.detection_zones.squares
self.scaled_polygon = self.base_polygon
# Latency instrumentation: populated by process_frame.
self.last_decision: dict | None = None
self.last_decision_perf: float | None = None
self.scaled_squares = self.base_squares
self.update_scaling(
settings.camera.desired_width, settings.camera.desired_height
Expand Down Expand Up @@ -97,6 +100,10 @@ def process_frame(self, frame: np.ndarray, project_type: str):
Processes a single frame for object detection and state tracking.
"""
start_time = time.perf_counter()
# Metadata about the decision taken on this frame, published for the
# latency logger. Kept on the instance rather than added to the return
# tuple so that every existing caller keeps working unchanged.
self.last_decision = None

# 1. Delegate actual detection to the loaded plugin
predictions = self.plugin.detect(frame)
Expand Down Expand Up @@ -124,6 +131,11 @@ def process_frame(self, frame: np.ndarray, project_type: str):
command_to_send = (
settings.detection_zones.enter_commands[index]
)
self.last_decision = {
"roi": index + 1,
"edge": "enter",
"token": command_to_send,
}
found_object_for_state_change = True
break
elif self.flag == 1: # Looking for exit
Expand All @@ -139,10 +151,20 @@ def process_frame(self, frame: np.ndarray, project_type: str):
self.current_square - 1
]
)
self.last_decision = {
"roi": self.current_square,
"edge": "exit",
"token": command_to_send,
}
self.current_square = 0
found_object_for_state_change = True

end_time = time.perf_counter()
# Published for the latency logger: the instant the ROI decision became
# available, i.e. the boundary between the compute leg and the I/O leg.
self.last_decision_perf = end_time
if self.last_decision is not None:
self.last_decision["t_decision_perf"] = end_time
log.debug(
"frame.processing.time",
duration_ms=(end_time - start_time) * 1000,
Expand Down
Loading
Loading