diff --git a/src/zebtrack/core/controller.py b/src/zebtrack/core/controller.py
index 783ee52..4d30b00 100644
--- a/src/zebtrack/core/controller.py
+++ b/src/zebtrack/core/controller.py
@@ -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()
@@ -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(
@@ -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):
@@ -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")
@@ -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:
diff --git a/src/zebtrack/core/detector.py b/src/zebtrack/core/detector.py
index 69b30fd..bb70a2a 100644
--- a/src/zebtrack/core/detector.py
+++ b/src/zebtrack/core/detector.py
@@ -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
@@ -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)
@@ -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
@@ -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,
diff --git a/src/zebtrack/io/arduino.py b/src/zebtrack/io/arduino.py
index d2488c0..b3b3206 100644
--- a/src/zebtrack/io/arduino.py
+++ b/src/zebtrack/io/arduino.py
@@ -1,3 +1,5 @@
+import queue
+import threading
import time
from types import TracebackType
from typing import Optional, Type
@@ -23,6 +25,10 @@ def __init__(self, port: str, baud_rate: int):
self.port = port
self.baud_rate = baud_rate
self.ser: Optional[serial.Serial] = None
+ # Trigger dispatch runs on its own thread (see send_command_async).
+ self._tx_queue: "queue.Queue" = queue.Queue(maxsize=8)
+ self._tx_stop = threading.Event()
+ self._tx_thread: Optional[threading.Thread] = None
log.info("arduino.init", port=self.port, baud_rate=self.baud_rate)
def connect(self) -> bool:
@@ -33,7 +39,17 @@ def connect(self) -> bool:
log.info("arduino.connect.already_connected")
return True
try:
- self.ser = serial.Serial(self.port, self.baud_rate, timeout=2)
+ # 250 ms is ~17x the 14.6 ms an ACK line takes at 9600 baud, so a
+ # healthy reply always arrives, while a lost one costs 0.25 s
+ # instead of stalling the closed loop for 2 s.
+ # write_timeout is NOT optional here. pyserial defaults it to None,
+ # which on Windows means WriteFile + GetOverlappedResult(bWait=True)
+ # -- a wait with no bound. A single stalled write then blocks the
+ # calling thread until the port is closed, which is exactly what
+ # froze the live preview for 60 s on 2026-08-08.
+ self.ser = serial.Serial(
+ self.port, self.baud_rate, timeout=0.25, write_timeout=0.25
+ )
# Opening the port auto-resets the Arduino Uno (DTR toggle). Wait
# for the board to finish booting before treating it as available.
time.sleep(2)
@@ -67,9 +83,86 @@ def __exit__(
"""Exit the runtime context and close the connection."""
self.close()
- def send_command(self, box_number: int) -> bool:
+ # ------------------------------------------------------------------
+ # Asynchronous dispatch
+ # ------------------------------------------------------------------
+ # The closed-loop analysis thread must never touch the serial port. Even
+ # with every timeout set, a USB-CDC device can stall in the driver, and a
+ # stall on the analysis thread stops detection, the preview and recording
+ # all at once. Triggers are queued here and written by a dedicated thread;
+ # a stalled port now costs dropped triggers (counted, logged) instead of a
+ # frozen application. The latency columns are unaffected: t_send and t_ack
+ # are still taken around the real write, and the queue hop shows up in
+ # decision_to_send_ms, where it is visible rather than hidden.
+
+ def _ensure_dispatcher(self) -> None:
+ if self._tx_thread is not None and self._tx_thread.is_alive():
+ return
+ self._tx_stop.clear()
+ self._tx_thread = threading.Thread(
+ target=self._tx_loop, name="ArduinoTxThread", daemon=True
+ )
+ self._tx_thread.start()
+
+ def _tx_loop(self) -> None:
+ while not self._tx_stop.is_set():
+ try:
+ box_number, kwargs = self._tx_queue.get(timeout=0.5)
+ except queue.Empty:
+ continue
+ try:
+ self.send_command(box_number, **kwargs)
+ except Exception: # noqa: BLE001 - a dead tx thread is silent
+ log.exception("arduino.tx_thread.error", command=box_number)
+ log.info("arduino.tx_thread.finished")
+
+ def send_command_async(self, box_number: int, **kwargs) -> bool:
+ """Queue a trigger for the dispatcher thread. Never blocks the caller.
+
+ Returns True if the trigger was queued, False if it was dropped because
+ the port is not keeping up.
+ """
+ self._ensure_dispatcher()
+ try:
+ self._tx_queue.put_nowait((box_number, kwargs))
+ return True
+ except queue.Full:
+ # Counted in 8_LatencyMeta as trigger_queue_drops. A stimulus the
+ # port was too slow to accept must not vanish from the record.
+ log.error("arduino.command.dropped_queue_full", command=box_number)
+ latency_logging.note_drop("trigger")
+ return False
+
+ def stop_dispatcher(self, timeout: float = 1.0) -> None:
+ """Stops the dispatcher thread, without waiting on a wedged write."""
+ self._tx_stop.set()
+ thread = self._tx_thread
+ if thread and thread.is_alive():
+ thread.join(timeout=timeout)
+ if thread.is_alive():
+ log.error("arduino.tx_thread.join_timeout", timeout_s=timeout)
+ self._tx_thread = None
+
+ def send_command(
+ self,
+ box_number: int,
+ *,
+ frame_t0: Optional[float] = None,
+ t_decision: Optional[float] = None,
+ frame: Optional[int] = None,
+ cam_seq: Optional[int] = None,
+ roi: Optional[int] = None,
+ edge: Optional[str] = None,
+ ) -> bool:
"""
Sends a command to the Arduino and waits for an acknowledgment.
+
+ The keyword arguments carry the identity and timing of the frame that
+ produced this decision. They are optional so that existing callers and
+ the module self-test keep working, but the live loop must pass them:
+ without ``frame_t0`` the end-to-end latency column cannot be computed,
+ and reading a stale global instead (as this method used to) silently
+ measures the wrong frame.
"""
try:
command_num = int(box_number)
@@ -79,35 +172,73 @@ def send_command(self, box_number: int) -> bool:
if self.ser and self.ser.is_open:
command = f"{command_num}\n"
- t_send = time.perf_counter()
+ # Bound before the try: the timeout handler below reports it, and
+ # the drain could in principle raise before it is assigned.
+ t_send = None
try:
+ # Drain BEFORE the write. Draining after the read (as this code
+ # previously did) means the next readline returns the ACK of the
+ # PREVIOUS command, already buffered — an off-by-one that made
+ # 97.7% of the logged serial round trips sub-millisecond and
+ # therefore physically impossible at 9600 baud.
+ self.ser.reset_input_buffer()
+ t_send = time.perf_counter()
self.ser.write(command.encode("utf-8"))
log.info("arduino.command.sent", command=command_num)
+ except serial.SerialTimeoutException:
+ # The write did not complete within write_timeout. Report the
+ # trigger as lost rather than waiting on the driver -- but still
+ # record the row. A trigger the firmware may never have acted on
+ # has to stay visible in 6_Latency_.csv and in n_triggers,
+ # otherwise the lost stimulus is indistinguishable from one that
+ # was never attempted.
+ log.error("arduino.command.write_timeout", command=command_num)
+ latency_logging.log_trigger(
+ command_num,
+ t_send,
+ None,
+ frame_t0,
+ ack_ok=False,
+ ack_text="WRITE_TIMEOUT",
+ frame=frame,
+ cam_seq=cam_seq,
+ roi=roi,
+ edge=edge,
+ t_decision=t_decision,
+ )
+ return False
+ except serial.SerialException as e:
+ log.error("arduino.command.send_error", exc_info=e)
+ return False
+ try:
response = self.ser.readline().decode("utf-8", errors="replace").strip()
- # Drain any extra lines the firmware may emit per command so they
- # don't leak into the next command's response.
t_ack = time.perf_counter()
- self.ser.reset_input_buffer()
+ ack_ok = bool(response)
latency_logging.log_trigger(
command_num,
t_send,
t_ack,
- latency_logging.FRAME_T0,
+ frame_t0,
+ ack_ok=ack_ok,
+ ack_text=response,
+ frame=frame,
+ cam_seq=cam_seq,
+ roi=roi,
+ edge=edge,
+ t_decision=t_decision,
)
- if response == "OK":
+ # The LED firmware replies with a human-readable line such as
+ # "Red LED 1 ON" — never the literal "OK" this branch used to
+ # require, so every command was previously logged as a nack and
+ # returned False even though the LED did fire. Any non-empty
+ # reply is an acknowledgment.
+ if ack_ok:
log.info(
"arduino.command.ack", command=command_num, response=response
)
return True
- if response:
- log.warning(
- "arduino.command.nack",
- command=command_num,
- response=response,
- )
- else:
- log.warning("arduino.command.no_response", command=command_num)
+ log.warning("arduino.command.no_response", command=command_num)
return False
except serial.SerialException as e:
log.error("arduino.command.send_error", exc_info=e)
@@ -120,6 +251,7 @@ def close(self) -> None:
"""
Closes the serial connection.
"""
+ self.stop_dispatcher()
if self.ser and self.ser.is_open:
self.ser.close()
log.info("arduino.connection.closed")
diff --git a/src/zebtrack/io/camera.py b/src/zebtrack/io/camera.py
index f77412a..3abca0f 100644
--- a/src/zebtrack/io/camera.py
+++ b/src/zebtrack/io/camera.py
@@ -6,7 +6,6 @@
import numpy as np
import structlog
-from zebtrack import latency_logging
from zebtrack.io.frame_source import FrameSource
from zebtrack.settings import settings
@@ -27,6 +26,9 @@ def __init__(self):
self.actual_width = int(self.cap.get(cv2.CAP_PROP_FRAME_WIDTH))
self.actual_height = int(self.cap.get(cv2.CAP_PROP_FRAME_HEIGHT))
+ # Read once, here, before the reader thread exists. Querying the capture
+ # backend from another thread while it is inside cap.read() can block.
+ self._declared_fps = self.cap.get(cv2.CAP_PROP_FPS)
log.info(
"camera.initialized",
index=self._camera_index,
@@ -36,6 +38,18 @@ def __init__(self):
self._lock = threading.Lock()
self._latest_frame: Tuple[bool, np.ndarray | None] = (False, None)
+ # Capture timestamp and sequence number of the frame currently held in
+ # ``_latest_frame``. These travel WITH the frame; the old module-level
+ # ``latency_logging.FRAME_T0`` global is gone (it was overwritten by
+ # this thread on every read, including frames never consumed).
+ self._latest_t0: float | None = None
+ self._latest_seq: int = 0
+ self._cam_seq: int = 0
+ # Running estimate of the achieved camera rate, used to write the video
+ # container at the true fps instead of the declared one.
+ self._fps_t_first: float | None = None
+ self._fps_t_last: float | None = None
+ self._fps_n: int = 0
self._stopped = threading.Event()
self._thread = threading.Thread(target=self._reader_thread, daemon=True)
self._thread.start()
@@ -59,8 +73,14 @@ def _reader_thread(self):
time.sleep(2)
continue
+ t_before = time.perf_counter()
ret, frame = self.cap.read()
- latency_logging.FRAME_T0 = time.perf_counter()
+ t0 = time.perf_counter()
+ # A read that takes this long is not a slow frame, it is a stalled
+ # device. Say so, instead of letting the preview silently freeze on
+ # the last good frame.
+ if t0 - t_before > 2.0:
+ log.warning("camera.read.stalled", seconds=round(t0 - t_before, 2))
if not ret:
self.cap.release()
@@ -70,7 +90,14 @@ def _reader_thread(self):
continue
with self._lock:
+ self._cam_seq += 1
self._latest_frame = (ret, frame)
+ self._latest_t0 = t0
+ self._latest_seq = self._cam_seq
+ if self._fps_t_first is None:
+ self._fps_t_first = t0
+ self._fps_t_last = t0
+ self._fps_n += 1
log.info("camera.reader_thread.stopped")
def get_frame(self) -> Tuple[bool, np.ndarray | None]:
@@ -81,12 +108,43 @@ def get_frame(self) -> Tuple[bool, np.ndarray | None]:
ret, frame = self._latest_frame
return ret, frame.copy() if ret else None
+ def get_frame_ts(self):
+ """Like :meth:`get_frame`, but also returns the frame's own capture
+ timestamp and the camera reader's sequence number for it.
+
+ Returns ``(ret, frame, t_capture_perf, cam_seq)``. ``cam_seq`` counts
+ camera reads, so the gap between consecutive consumed values is the
+ number of camera frames skipped by the "latest frame wins" policy.
+ """
+ with self._lock:
+ ret, frame = self._latest_frame
+ t0, seq = self._latest_t0, self._latest_seq
+ return ret, (frame.copy() if ret else None), t0, seq
+
+ def measured_fps(self) -> float | None:
+ """Achieved camera rate since start-up, or None if not yet estimable.
+
+ Measured, not declared. ``settings.video_processing.fps`` is a request,
+ not an observation, and on this rig the two differ by ~30%.
+ """
+ with self._lock:
+ if self._fps_n < 2 or self._fps_t_first is None:
+ return None
+ span = self._fps_t_last - self._fps_t_first
+ return (self._fps_n - 1) / span if span > 0 else None
+
def release(self) -> None:
"""
Signals the reader thread to stop and releases the camera resource.
"""
self._stopped.set()
self._thread.join(timeout=2)
+ if self._thread.is_alive():
+ # The reader is still inside cap.read(). Calling cap.release() now
+ # would block on the same backend and hang the UI thread. The thread
+ # is a daemon and the handle is freed at process exit.
+ log.error("camera.release.reader_still_running")
+ return
if self.cap.isOpened():
self.cap.release()
log.info("camera.released")
@@ -98,7 +156,10 @@ def get_properties(self) -> Dict[str, Any]:
return {
"width": self.actual_width,
"height": self.actual_height,
- "fps": self.cap.get(cv2.CAP_PROP_FPS) or settings.video_processing.fps,
+ "fps": self._declared_fps or settings.video_processing.fps,
+ # Observed rate of the reader thread. Prefer this over "fps" for
+ # anything that converts frames to milliseconds.
+ "fps_measured": self.measured_fps(),
}
diff --git a/src/zebtrack/io/recorder.py b/src/zebtrack/io/recorder.py
index f6953e9..3b62395 100644
--- a/src/zebtrack/io/recorder.py
+++ b/src/zebtrack/io/recorder.py
@@ -28,7 +28,12 @@ def __init__(self):
self.recording_start_frame = 0
def start_recording(
- self, output_folder, frame_width, frame_height, is_video_file=False
+ self,
+ output_folder,
+ frame_width,
+ frame_height,
+ is_video_file=False,
+ fps=None,
):
"""
Prepares and starts a new recording session.
@@ -38,6 +43,10 @@ def start_recording(
frame_width (int): The width of the video frames.
frame_height (int): The height of the video frames.
is_video_file (bool): If True, skips video file creation.
+ fps (float | None): Frame rate to stamp into the video container.
+ Pass the *measured* camera rate. Falls back to the configured
+ value, which is a request rather than an observation and on
+ this rig differs from the achieved rate by about 30%.
Returns:
bool: True if recording started successfully, False otherwise.
@@ -55,10 +64,24 @@ def start_recording(
if not is_video_file:
video_filename = os.path.join(output_folder, f"{self.base_name}.mp4")
fourcc = cv2.VideoWriter_fourcc(*"mp4v")
+ # Explicit about None, but still rejecting non-positive values: a
+ # container stamped with fps=0 cannot be timed by any player, so a
+ # bad measurement has to fall back rather than pass through.
+ use_measured = fps is not None and float(fps) > 0
+ write_fps = (
+ float(fps) if use_measured else float(settings.video_processing.fps)
+ )
+ self.video_fps = write_fps
+ log_context.info(
+ "recorder.video_fps",
+ fps_used=write_fps,
+ fps_configured=settings.video_processing.fps,
+ measured=use_measured,
+ )
self.video_writer = cv2.VideoWriter(
video_filename,
fourcc,
- settings.video_processing.fps,
+ write_fps,
(frame_width, frame_height),
)
if not self.video_writer.isOpened():
diff --git a/src/zebtrack/latency_logging.py b/src/zebtrack/latency_logging.py
index ee700cb..e4c4939 100644
--- a/src/zebtrack/latency_logging.py
+++ b/src/zebtrack/latency_logging.py
@@ -1,63 +1,299 @@
-"""latency_logging.py — instrumentacao de latencia do loop fechado (PyZebArdYolo).
-Coloque este arquivo em src/zebtrack/ . Nao tem dependencias alem da stdlib.
+"""latency_logging.py — closed-loop latency instrumentation (PyZebArdYolo).
-Mede dois tempos por gatilho (entrada/saida de ROI -> comando ao Arduino):
- - serial_act_ms : t_ack - t_send (transmissao serial + atuacao do Arduino/LED)
- - frame_to_ack_ms : t_ack - FRAME_T0
- (captura do frame -> LED aceso = ponta-a-ponta de software)
+Stdlib only. Writes three files per recording session, all into the session's
+output folder:
-FRAME_T0 e atualizado pela alca de captura (uma linha; ver PASSO_A_PASSO.md).
-Saida: CSV definido em PYZEB_LATENCY_CSV (default ./latency_log.csv).
+ 6_Latency_.csv one row per Arduino trigger
+ 7_FrameLedger_.csv one row per frame handed to the video writer
+ 8_LatencyMeta_.json session-level metadata (measured fps, drops, config)
+
+Design notes
+------------
+The previous version exposed a module-level ``FRAME_T0`` that the camera reader
+thread overwrote on every ``cap.read()``. Because the reader runs free and
+``Camera.get_frame()`` returns only the most recent frame, ``FRAME_T0`` almost
+never referred to the frame that actually produced the decision. The resulting
+``frame_to_ack_ms`` column was near-uniform over one frame interval and
+correlated negatively with its own serial leg. That global is gone: the capture
+timestamp is now carried with the frame and passed explicitly.
+
+The frame ledger exists because neither ``live_frame_count`` nor the video frame
+index is a camera frame index — frames are skipped at ``get_frame()`` and
+dropped again at both queues. The ledger makes the video<->pipeline mapping
+exact and makes drops visible instead of silent.
"""
import csv
+import json
import os
import threading
import time
-FRAME_T0 = None # setado na alca de captura: time.perf_counter()
-_PATH = os.environ.get("PYZEB_LATENCY_CSV", "latency_log.csv")
+# Deprecated. Kept only so that any stale import does not raise. Nothing in the
+# application writes or reads it any more; passing it to log_trigger is a no-op.
+FRAME_T0 = None
+
_LOCK = threading.Lock()
-_f = None
-_w = None
-
-
-def _ensure():
- global _f, _w
- if _w is None:
- new = (not os.path.exists(_PATH)) or os.path.getsize(_PATH) == 0
- _f = open(_PATH, "a", newline="", encoding="utf-8")
- _w = csv.writer(_f)
- if new:
- _w.writerow(
- [
- "wall_iso",
- "cmd",
- "t_send_perf",
- "t_ack_perf",
- "serial_act_ms",
- "frame_to_ack_ms",
- ]
- )
+_TRIG_COLUMNS = [
+ "event_id",
+ "wall_iso",
+ "frame", # live_frame_count of the frame that produced the decision
+ "cam_seq", # camera reader sequence number of that same frame
+ "roi", # square index (1-based), or "" if unknown
+ "edge", # "enter" | "exit"
+ "token", # command byte sent to the Arduino
+ "t_capture_perf", # perf_counter immediately after cap.read() returned
+ "t_decision_perf", # perf_counter at the end of Detector.process_frame
+ "t_send_perf", # perf_counter immediately before ser.write
+ "t_ack_perf", # perf_counter immediately after ser.readline returned
+ "ack_ok", # True if readline returned a non-empty line
+ "ack_text",
+ "capture_to_decision_ms",
+ "decision_to_send_ms",
+ "serial_act_ms",
+ "frame_to_ack_ms",
+]
+
+_LEDGER_COLUMNS = ["video_write_index", "frame", "cam_seq", "t_capture_perf"]
+
+
+def _fmt(value, decimals=6):
+ """Format an optional float for CSV: empty cell for None, never for 0.0."""
+ return "" if value is None else f"{value:.{decimals}f}"
+
+
+class _Session:
+ """Holds the open file handles and counters for one recording session."""
+
+ def __init__(self, folder, base_name, meta=None):
+ self.folder = folder
+ self.base_name = base_name
+ self.meta = dict(meta or {})
+ self.event_id = 0
+ self.video_write_index = 0
+ self.video_drops = 0
+ self.analysis_drops = 0
+ self.trigger_drops = 0
+ self.t_first_capture = None
+ self.t_last_capture = None
+ self.n_captures = 0
+
+ self._tf = open(
+ os.path.join(folder, f"6_Latency_{base_name}.csv"),
+ "w", newline="", encoding="utf-8",
+ )
+ self._tw = csv.writer(self._tf)
+ self._tw.writerow(_TRIG_COLUMNS)
+ self._tf.flush()
+
+ self._lf = open(
+ os.path.join(folder, f"7_FrameLedger_{base_name}.csv"),
+ "w", newline="", encoding="utf-8",
+ )
+ self._lw = csv.writer(self._lf)
+ self._lw.writerow(_LEDGER_COLUMNS)
+ self._lf.flush()
+
+ def close(self):
+ for f in (self._tf, self._lf):
+ try:
+ f.flush()
+ f.close()
+ except Exception:
+ pass
+
+
+_SESSION: "_Session | None" = None
+
+
+def start_session(folder, base_name, meta=None):
+ """Open the latency files for a recording session. Safe to call twice."""
+ global _SESSION
+ with _LOCK:
+ if _SESSION is not None:
+ _SESSION.close()
+ try:
+ os.makedirs(folder, exist_ok=True)
+ _SESSION = _Session(folder, base_name, meta)
+ except Exception:
+ _SESSION = None
+ return _SESSION is not None
+
+
+def stop_session(extra_meta=None):
+ """Flush and close the session, writing the metadata sidecar.
-def log_trigger(cmd, t_send, t_ack, frame_t0=None):
- """Grava uma linha por gatilho. Chamado de dentro de Arduino.send_command()."""
+ Teardown happens entirely under ``_LOCK``. Clearing ``_SESSION`` first and
+ closing the handles afterwards would let a logging call that had already
+ read the old handle write to a closed file, and would snapshot the counters
+ while a worker was still incrementing them. Every logger below therefore
+ reads ``_SESSION`` under the same lock, so each call either completes
+ against a live session or sees None and skips.
+ """
+ global _SESSION
+ with _LOCK:
+ s = _SESSION
+ _SESSION = None
+ if s is None:
+ return
+ meta = dict(s.meta)
+ meta.update(extra_meta or {})
+ meta.update(
+ {
+ "n_triggers": s.event_id,
+ "n_video_frames_written": s.video_write_index,
+ "video_queue_drops": s.video_drops,
+ "analysis_queue_drops": s.analysis_drops,
+ "trigger_queue_drops": s.trigger_drops,
+ "n_captures_consumed": s.n_captures,
+ "fps_measured": measured_fps_from(s),
+ "t_first_capture_perf": s.t_first_capture,
+ "t_last_capture_perf": s.t_last_capture,
+ }
+ )
+ try:
+ path = os.path.join(s.folder, f"8_LatencyMeta_{s.base_name}.json")
+ with open(path, "w", encoding="utf-8") as fh:
+ json.dump(meta, fh, indent=1, default=str)
+ except Exception:
+ pass
+ s.close()
+
+
+def measured_fps_from(s):
+ """Achieved consumption rate over the session, or None if undeterminable."""
+ if s is None or s.n_captures < 2:
+ return None
+ if s.t_first_capture is None or s.t_last_capture is None:
+ return None
+ span = s.t_last_capture - s.t_first_capture
+ return (s.n_captures - 1) / span if span > 0 else None
+
+
+def note_capture(t_capture):
+ """Record a consumed frame's capture timestamp (for the fps estimate)."""
+ with _LOCK:
+ s = _SESSION
+ if s is None:
+ return
+ if s.t_first_capture is None:
+ s.t_first_capture = t_capture
+ s.t_last_capture = t_capture
+ s.n_captures += 1
+
+
+def note_drop(kind):
+ """Count a dropped frame. kind is 'video', 'analysis' or 'trigger'."""
+ with _LOCK:
+ s = _SESSION
+ if s is None:
+ return
+ if kind == "video":
+ s.video_drops += 1
+ elif kind == "trigger":
+ s.trigger_drops += 1
+ else:
+ s.analysis_drops += 1
+
+
+def log_video_frame(frame, cam_seq, t_capture):
+ """One ledger row per frame accepted by the video queue.
+
+ Returns the 0-based index this frame will occupy in the mp4, which is what
+ makes the video<->pipeline mapping exact.
+ """
try:
with _LOCK:
- _ensure()
- serial_ms = (t_ack - t_send) * 1000.0
- f2a = ((t_ack - frame_t0) * 1000.0) if frame_t0 else ""
- _w.writerow(
+ s = _SESSION
+ if s is None:
+ return None
+ idx = s.video_write_index
+ s.video_write_index += 1
+ s._lw.writerow([idx, frame, cam_seq, _fmt(t_capture)])
+ s._lf.flush()
+ return idx
+ except Exception:
+ return None
+
+
+def log_trigger(
+ cmd,
+ t_send,
+ t_ack,
+ frame_t0=None,
+ *,
+ ack_ok=None,
+ ack_text="",
+ frame=None,
+ cam_seq=None,
+ roi=None,
+ edge=None,
+ t_decision=None,
+):
+ """One row per Arduino trigger.
+
+ ``frame_t0`` must be the capture timestamp of the frame that produced this
+ decision, passed explicitly by the caller. The old module-level global is
+ no longer consulted.
+
+ ``t_ack`` may be None for a trigger that never got as far as a reply (a
+ write timeout, say). The row is still written, with the acknowledgement
+ columns empty and ``ack_ok`` false, because a trigger the firmware may
+ never have acted on is a data point, not an absence of one.
+
+ All timestamps are tested against None rather than for truthiness:
+ perf_counter's epoch is arbitrary, so 0.0 is a legal reading, and treating
+ it as "missing" would silently blank a latency column.
+ """
+ try:
+ with _LOCK:
+ s = _SESSION
+ if s is None:
+ return
+ s.event_id += 1
+ eid = s.event_id
+ serial_ms = (
+ (t_ack - t_send) * 1000.0
+ if (t_ack is not None and t_send is not None)
+ else None
+ )
+ f2a = (
+ (t_ack - frame_t0) * 1000.0
+ if (t_ack is not None and frame_t0 is not None)
+ else None
+ )
+ c2d = (
+ (t_decision - frame_t0) * 1000.0
+ if (t_decision is not None and frame_t0 is not None)
+ else None
+ )
+ d2s = (
+ (t_send - t_decision) * 1000.0
+ if (t_send is not None and t_decision is not None)
+ else None
+ )
+ s._tw.writerow(
[
+ eid,
time.strftime("%Y-%m-%dT%H:%M:%S"),
+ "" if frame is None else frame,
+ "" if cam_seq is None else cam_seq,
+ "" if roi is None else roi,
+ "" if edge is None else edge,
cmd,
- f"{t_send:.6f}",
- f"{t_ack:.6f}",
- f"{serial_ms:.3f}",
- f"{f2a:.3f}" if f2a != "" else "",
+ _fmt(frame_t0),
+ _fmt(t_decision),
+ _fmt(t_send),
+ _fmt(t_ack),
+ "" if ack_ok is None else bool(ack_ok),
+ ack_text,
+ _fmt(c2d, 3),
+ _fmt(d2s, 3),
+ _fmt(serial_ms, 3),
+ _fmt(f2a, 3),
]
)
- _f.flush()
+ s._tf.flush()
except Exception:
- pass # nunca derrubar o loop por causa de logging
+ pass # never take down the loop for logging
diff --git a/src/zebtrack/ui/gui.py b/src/zebtrack/ui/gui.py
index b35f610..ffec7e5 100644
--- a/src/zebtrack/ui/gui.py
+++ b/src/zebtrack/ui/gui.py
@@ -26,6 +26,7 @@
import structlog
# Import custom modules
+from zebtrack import latency_logging
from zebtrack.core.detector import Detector, draw_overlay
from zebtrack.io.camera import Camera
from zebtrack.io.video_source import VideoFileSource
@@ -58,6 +59,9 @@ def __init__(self, root, controller):
self.progress_bar = None
self.progress_labels: dict[str, StringVar] = {}
self.video_label: Label | None = None
+ # Live preview hand-off state (see _post_live_preview).
+ self._preview_pending = False
+ self._preview_last_t = 0.0
# User options
self.processing_interval_var = StringVar(
@@ -301,27 +305,55 @@ def _live_frame_capture_loop(self):
"""
Loop para capturar quadros de uma fonte AO VIVO (câmera).
"""
+ try:
+ self._live_frame_capture_loop_body()
+ except Exception: # noqa: BLE001 - a silent thread death freezes the app
+ log.exception("gui.capture_thread.crashed")
+ finally:
+ log.info("gui.live_frame_capture_loop.finished")
+
+ def _live_frame_capture_loop_body(self):
live_frame_count = 0
while not self.controller.program_exit_event.is_set():
- if not self.controller.active_frame_source:
+ source = self.controller.active_frame_source
+ if not source:
time.sleep(0.1)
continue
- ret, frame = self.controller.active_frame_source.get_frame()
+ # Take the frame together with its own capture timestamp and the
+ # camera reader's sequence number. Sources that predate this API
+ # (e.g. VideoFileSource) fall back to the two-tuple form.
+ if hasattr(source, "get_frame_ts"):
+ ret, frame, t_capture, cam_seq = source.get_frame_ts()
+ else:
+ ret, frame = source.get_frame()
+ t_capture, cam_seq = time.perf_counter(), None
if not ret:
log.error("gui.capture_thread.get_frame_failed")
time.sleep(0.5)
continue
live_frame_count += 1
+ latency_logging.note_capture(t_capture)
if not self.controller.frame_queue.full():
- self.controller.frame_queue.put((live_frame_count, frame.copy()))
- if (
- self.controller.is_capturing_for_video
- and not self.controller.video_queue.full()
- ):
- self.controller.video_queue.put(frame.copy())
+ self.controller.frame_queue.put(
+ (live_frame_count, frame.copy(), t_capture, cam_seq)
+ )
+ else:
+ latency_logging.note_drop("analysis")
+ if self.controller.is_capturing_for_video:
+ if not self.controller.video_queue.full():
+ self.controller.video_queue.put(frame.copy())
+ # One ledger row per frame that will actually be written,
+ # in write order. This is what makes the video frame index
+ # recoverable: neither live_frame_count nor the camera
+ # sequence indexes the mp4 once a drop occurs.
+ latency_logging.log_video_frame(
+ live_frame_count, cam_seq, t_capture
+ )
+ else:
+ latency_logging.note_drop("video")
time.sleep(1 / (settings.video_processing.fps * 1.5))
@@ -329,18 +361,40 @@ def _live_processing_loop(self):
"""
Loop para processar quadros de uma fonte AO VIVO.
"""
+ try:
+ self._live_processing_loop_body()
+ except Exception: # noqa: BLE001 - see _live_frame_capture_loop
+ log.exception("gui.processing_thread.crashed")
+ finally:
+ log.info("gui.live_processing_loop.finished")
+
+ def _live_processing_loop_body(self):
while not self.controller.program_exit_event.is_set():
try:
- frame_number, frame = self.controller.frame_queue.get(timeout=1)
+ item = self.controller.frame_queue.get(timeout=1)
except queue.Empty:
continue
+ if len(item) == 4:
+ frame_number, frame, t_capture, cam_seq = item
+ else: # tolerate the legacy two-tuple
+ frame_number, frame = item
+ t_capture, cam_seq = None, None
if self.controller.is_processing:
- detections, command = self.controller.detector.process_frame(
- frame, "live"
- )
+ detector = self.controller.detector
+ detections, command = detector.process_frame(frame, "live")
if command is not None:
- self.controller.arduino.send_command(command)
+ meta = detector.last_decision or {}
+ # Async: the analysis thread must not block on the port.
+ self.controller.arduino.send_command_async(
+ command,
+ frame_t0=t_capture,
+ t_decision=detector.last_decision_perf,
+ frame=frame_number,
+ cam_seq=cam_seq,
+ roi=meta.get("roi"),
+ edge=meta.get("edge"),
+ )
if self.controller.is_recording and detections:
timestamp = time.time() - self.controller.recorder.start_time
self.controller.recorder.write_detection_data(
@@ -348,12 +402,60 @@ def _live_processing_loop(self):
)
draw_overlay(frame, detections, self.controller.detector)
- cv2.imshow("Live View", frame)
- if cv2.waitKey(1) & 0xFF == ord("q"):
- self.controller.on_close()
- break
- cv2.destroyAllWindows()
- log.info("gui.live_processing_loop.finished")
+ # Preview is rendered by Tk on the main thread. It must NOT be done
+ # with cv2.imshow/cv2.waitKey from here: HighGUI drives a Win32
+ # message loop owned by the calling thread, and while Tk runs its
+ # own (modal dialogs included) on the main thread the two queues can
+ # become attached and waitKey never returns. The symptom is exactly
+ # this: the live window stops updating on a correctly drawn frame,
+ # the Tk window stays responsive, and shutdown hangs forever in
+ # processing_thread.join().
+ self._post_live_preview(frame)
+
+ # Preview cadence for the live view. The analysis loop runs as fast as the
+ # detector allows; the screen does not need to.
+ _PREVIEW_MIN_INTERVAL_S = 1 / 20
+ _PREVIEW_MAX_W = 800
+
+ def _post_live_preview(self, frame):
+ """Hand a frame to the Tk main thread for display, dropping if behind.
+
+ Called from the processing thread. Never blocks it: if the previous
+ frame has not been drawn yet, or the cadence budget is not met, the
+ frame is simply skipped.
+ """
+ now = time.perf_counter()
+ if self._preview_pending:
+ return
+ if now - self._preview_last_t < self._PREVIEW_MIN_INTERVAL_S:
+ return
+ self._preview_last_t = now
+ self._preview_pending = True
+ # Downscale here, on the worker thread, so the main thread only pays for
+ # the Tk blit. A 1280x720 label would also oversize the window.
+ h, w = frame.shape[:2]
+ if w > self._PREVIEW_MAX_W:
+ scale = self._PREVIEW_MAX_W / w
+ shown = cv2.resize(frame, (self._PREVIEW_MAX_W, int(h * scale)))
+ else:
+ shown = frame.copy()
+ try:
+ self.root.after(0, self._draw_live_preview, shown)
+ except Exception: # noqa: BLE001 - root may be tearing down
+ self._preview_pending = False
+
+ def _draw_live_preview(self, frame):
+ """Main-thread half of :meth:`_post_live_preview`."""
+ try:
+ self.show_video_area()
+ self.display_frame(frame)
+ finally:
+ self._preview_pending = False
+
+ def show_video_area(self):
+ """Makes the preview/progress area visible (idempotent)."""
+ if self.progress_frame and not self.progress_frame.winfo_viewable():
+ self.progress_frame.pack(pady=5, fill="x", padx=10)
def _file_processing_loop(self):
"""
diff --git a/tests/test_latency_instrumentation.py b/tests/test_latency_instrumentation.py
new file mode 100644
index 0000000..d318b13
--- /dev/null
+++ b/tests/test_latency_instrumentation.py
@@ -0,0 +1,277 @@
+"""Regression tests for the closed-loop latency instrumentation.
+
+These exist because two defects shipped undetected and were only caught by
+statistical audit of the output CSV, months later:
+
+ D1 ``reset_input_buffer()`` was called *after* ``readline()``, so every
+ trigger but the first measured a buffered read of the previous command's
+ ACK (~0.5 ms) instead of a serial round trip. 346 of 354 logged rows were
+ physically impossible at 9600 baud.
+ D2 the frame capture timestamp was a module-level global overwritten by a
+ free-running camera thread, so it referred to a later frame than the one
+ that produced the decision.
+
+Both are cheap to assert and neither was covered.
+"""
+
+import csv
+import json
+import os
+import time
+
+import pytest
+
+from zebtrack import latency_logging
+from zebtrack.io.arduino import Arduino
+
+
+class FakeSerial:
+ """Minimal serial stand-in that records the order of operations.
+
+ Models the real failure mode: the device replies asynchronously, so a line
+ from the *previous* command is still sitting in the buffer unless the caller
+ drains it before writing.
+ """
+
+ def __init__(self, ack=b"Red LED 1 ON\n", ack_delay_s=0.015):
+ self.is_open = True
+ self.calls = []
+ self._buffer = [b"stale line from previous command\n"]
+ self._ack = ack
+ self._ack_delay = ack_delay_s
+
+ def write(self, payload):
+ self.calls.append("write")
+ time.sleep(self._ack_delay) # transit + firmware turnaround
+ self._buffer.append(self._ack)
+ return len(payload)
+
+ def readline(self):
+ self.calls.append("readline")
+ return self._buffer.pop(0) if self._buffer else b""
+
+ def reset_input_buffer(self):
+ self.calls.append("reset_input_buffer")
+ self._buffer.clear()
+
+ def close(self):
+ self.is_open = False
+
+
+def _make_arduino():
+ ard = Arduino(port="FAKE", baud_rate=9600)
+ ard.ser = FakeSerial()
+ return ard
+
+
+def test_buffer_is_drained_before_write_not_after():
+ """D1 regression: draining after the read yields the previous ACK."""
+ ard = _make_arduino()
+ ard.send_command(1)
+ order = ard.ser.calls
+ assert order.index("reset_input_buffer") < order.index("write"), (
+ f"buffer must be drained before the write; got {order}"
+ )
+
+
+def test_serial_leg_reflects_a_real_round_trip(tmp_path):
+ """The measured serial leg must include the device turnaround, not a
+ buffered read. Sub-millisecond values are the D1 signature."""
+ latency_logging.start_session(str(tmp_path), "unit")
+ ard = _make_arduino()
+ t0 = time.perf_counter()
+ for _ in range(5):
+ assert ard.send_command(1, frame_t0=t0, t_decision=t0 + 0.01, frame=10) is True
+ latency_logging.stop_session()
+
+ rows = list(csv.DictReader(open(tmp_path / "6_Latency_unit.csv", encoding="utf-8")))
+ assert len(rows) == 5
+ serial_ms = [float(r["serial_act_ms"]) for r in rows]
+ assert all(s > 10.0 for s in serial_ms), (
+ f"sub-millisecond serial legs indicate a stale buffered read: {serial_ms}"
+ )
+ assert all(r["ack_ok"] == "True" for r in rows)
+ assert all(r["ack_text"] == "Red LED 1 ON" for r in rows)
+
+
+def test_non_ok_ack_is_accepted():
+ """The LED firmware never replies 'OK'; any non-empty line is an ack."""
+ ard = _make_arduino()
+ ard.ser._ack = b"Blue LED OFF\n"
+ assert ard.send_command(4) is True
+
+
+def test_missing_ack_is_reported_not_silently_timed_out(tmp_path):
+ latency_logging.start_session(str(tmp_path), "unit")
+ ard = _make_arduino()
+ ard.ser._ack = b"" # device says nothing
+ assert ard.send_command(1, frame_t0=time.perf_counter()) is False
+ latency_logging.stop_session()
+ rows = list(csv.DictReader(open(tmp_path / "6_Latency_unit.csv", encoding="utf-8")))
+ assert rows[0]["ack_ok"] == "False"
+
+
+def test_no_module_level_frame_timestamp_is_consulted(tmp_path):
+ """D2 regression: the capture time must come from the argument, never from
+ module state, so that a concurrently advancing camera cannot poison it."""
+ latency_logging.start_session(str(tmp_path), "unit")
+ ard = _make_arduino()
+ latency_logging.FRAME_T0 = time.perf_counter() + 999 # absurd global
+ explicit_t0 = time.perf_counter()
+ ard.send_command(1, frame_t0=explicit_t0)
+ latency_logging.stop_session()
+ row = next(csv.DictReader(open(tmp_path / "6_Latency_unit.csv", encoding="utf-8")))
+ assert 0 < float(row["frame_to_ack_ms"]) < 1000, (
+ "end-to-end latency was computed from the stale global, not the argument"
+ )
+
+
+def test_legs_sum_to_end_to_end(tmp_path):
+ latency_logging.start_session(str(tmp_path), "unit")
+ ard = _make_arduino()
+ t_cap = time.perf_counter()
+ time.sleep(0.005)
+ t_dec = time.perf_counter()
+ ard.send_command(1, frame_t0=t_cap, t_decision=t_dec, frame=20, cam_seq=25,
+ roi=1, edge="enter")
+ latency_logging.stop_session()
+ r = next(csv.DictReader(open(tmp_path / "6_Latency_unit.csv", encoding="utf-8")))
+ total = (
+ float(r["capture_to_decision_ms"])
+ + float(r["decision_to_send_ms"])
+ + float(r["serial_act_ms"])
+ )
+ assert total == pytest.approx(float(r["frame_to_ack_ms"]), abs=0.01)
+ assert r["roi"] == "1" and r["edge"] == "enter" and r["frame"] == "20"
+
+
+def test_frame_ledger_indexes_written_frames_contiguously(tmp_path):
+ latency_logging.start_session(str(tmp_path), "unit")
+ idx = [
+ latency_logging.log_video_frame(
+ frame=f, cam_seq=f + 3, t_capture=time.perf_counter()
+ )
+ for f in range(1, 6)
+ ]
+ latency_logging.note_drop("video")
+ latency_logging.stop_session()
+ assert idx == [0, 1, 2, 3, 4]
+ ledger = tmp_path / "7_FrameLedger_unit.csv"
+ rows = list(csv.DictReader(open(ledger, encoding="utf-8")))
+ assert [int(r["video_write_index"]) for r in rows] == [0, 1, 2, 3, 4]
+ meta = json.load(open(tmp_path / "8_LatencyMeta_unit.json", encoding="utf-8"))
+ assert meta["video_queue_drops"] == 1
+ assert meta["n_video_frames_written"] == 5
+
+
+def test_measured_fps_is_recorded(tmp_path):
+ latency_logging.start_session(str(tmp_path), "unit")
+ t = time.perf_counter()
+ for k in range(40):
+ latency_logging.note_capture(t + k / 39.0) # simulate 39 fps
+ latency_logging.stop_session()
+ meta = json.load(open(tmp_path / "8_LatencyMeta_unit.json", encoding="utf-8"))
+ assert meta["fps_measured"] == pytest.approx(39.0, rel=1e-6)
+
+
+def test_logging_never_raises_without_a_session():
+ """Logging must never take down the closed loop."""
+ latency_logging.stop_session()
+ latency_logging.log_trigger(1, 0.0, 0.1)
+ latency_logging.log_video_frame(1, 1, 0.0)
+ latency_logging.note_capture(0.0)
+ latency_logging.note_drop("video")
+
+
+def test_session_files_land_in_the_recording_folder(tmp_path):
+ folder = tmp_path / "Grupo_1"
+ latency_logging.start_session(str(folder), "Grupo_1")
+ latency_logging.stop_session()
+ for name in ("6_Latency_Grupo_1.csv", "7_FrameLedger_Grupo_1.csv",
+ "8_LatencyMeta_Grupo_1.json"):
+ assert os.path.exists(folder / name)
+
+
+# --- Review follow-ups (PR #21) -------------------------------------------
+
+
+def test_zero_is_a_legal_timestamp_not_a_missing_one(tmp_path):
+ """perf_counter's epoch is arbitrary, so 0.0 is a reading, not an absence.
+
+ Truthiness checks blanked the latency columns for it, which corrupts the
+ measurement silently -- the failure mode this whole module exists to stop.
+ """
+ latency_logging.start_session(str(tmp_path), "unit")
+ latency_logging.log_trigger(
+ 1, 0.0, 0.5, 0.0, ack_ok=True, ack_text="Red LED 1 ON", t_decision=0.0
+ )
+ latency_logging.stop_session()
+ row = next(csv.DictReader(open(tmp_path / "6_Latency_unit.csv", encoding="utf-8")))
+ assert row["t_capture_perf"] == "0.000000"
+ assert row["t_decision_perf"] == "0.000000"
+ assert row["t_send_perf"] == "0.000000"
+ assert float(row["frame_to_ack_ms"]) == pytest.approx(500.0)
+ assert float(row["capture_to_decision_ms"]) == pytest.approx(0.0)
+
+
+def test_a_trigger_with_no_ack_still_produces_a_row(tmp_path):
+ """A write timeout is a data point: the stimulus may never have fired."""
+ latency_logging.start_session(str(tmp_path), "unit")
+ latency_logging.log_trigger(
+ 7, 1.0, None, 0.5, ack_ok=False, ack_text="WRITE_TIMEOUT", roi=4, edge="enter"
+ )
+ latency_logging.stop_session()
+ row = next(csv.DictReader(open(tmp_path / "6_Latency_unit.csv", encoding="utf-8")))
+ assert row["ack_text"] == "WRITE_TIMEOUT"
+ assert row["ack_ok"] == "False"
+ assert row["t_ack_perf"] == ""
+ assert row["serial_act_ms"] == ""
+ assert row["frame_to_ack_ms"] == ""
+ # It still counts as an attempted trigger.
+ meta = json.load(open(tmp_path / "8_LatencyMeta_unit.json", encoding="utf-8"))
+ assert meta["n_triggers"] == 1
+
+
+def test_stop_session_does_not_race_concurrent_logging(tmp_path):
+ """stop_session must quiesce logging, not close files under a live writer.
+
+ Reading _SESSION outside the lock let a call that had already taken the
+ handle write to a closed file after teardown.
+ """
+ import threading
+
+ latency_logging.start_session(str(tmp_path), "unit")
+ stop = threading.Event()
+ errors = []
+
+ def hammer():
+ while not stop.is_set():
+ try:
+ latency_logging.note_capture(time.perf_counter())
+ latency_logging.log_video_frame(1, 1, time.perf_counter())
+ latency_logging.log_trigger(1, 0.0, 0.1, 0.0)
+ except Exception as exc: # noqa: BLE001
+ errors.append(exc)
+ return
+
+ workers = [threading.Thread(target=hammer) for _ in range(4)]
+ for w in workers:
+ w.start()
+ time.sleep(0.2)
+ latency_logging.stop_session()
+ stop.set()
+ for w in workers:
+ w.join(timeout=2)
+
+ assert not errors
+ # The sidecar must exist and its counters must be internally consistent
+ # with the rows that were actually written.
+ meta = json.load(open(tmp_path / "8_LatencyMeta_unit.json", encoding="utf-8"))
+ ledger = list(
+ csv.DictReader(open(tmp_path / "7_FrameLedger_unit.csv", encoding="utf-8"))
+ )
+ triggers = list(
+ csv.DictReader(open(tmp_path / "6_Latency_unit.csv", encoding="utf-8"))
+ )
+ assert meta["n_video_frames_written"] == len(ledger)
+ assert meta["n_triggers"] == len(triggers)