From d8d66f4e1ea7bda66fe41267c7f096e2dfd6c6ae Mon Sep 17 00:00:00 2001 From: MarkSant Date: Sat, 8 Aug 2026 17:56:16 -0300 Subject: [PATCH 1/2] Fix closed-loop latency instrumentation and two live-mode freezes The optical validation showed the software log understating physical latency by 2.8x (411 ms measured vs 148 ms logged) and the declared fps (30) not matching the real one (38.95). A statistical audit of latency_log.csv found it was an instrumentation artefact, not a measurement: 346/354 rows with a sub-millisecond serial round trip, an end-to-end column near-uniform over one frame interval, and a NEGATIVE correlation between the serial leg and the total. Both causes were confirmed in code and fixed, along with two freezes found while validating the result on the rig. Instrumentation - arduino.py: reset_input_buffer() ran AFTER readline(), so from the second trigger on, readline returned the PREVIOUS command's ACK, already buffered. Root cause of the 346 impossible rows. Drained before the write instead. - camera.py/arduino.py: the module-level FRAME_T0 was set by the reader thread on every cap.read(), including frames never consumed, so it pointed at a frame later than the one that produced the decision. The global is gone; the capture timestamp travels with the frame. - arduino.py: `if response == "OK"` never matched -- the firmware replies "Red LED 1 ON". Every command was logged as a nack and send_command returned False even though the LED fired. Any non-empty reply is now an ACK. - recorder.py: VideoWriter was stamped with the configured fps on a ~39 fps stream. It now takes the measured rate. - latency_logging.py: rewritten. Three files per session in the recording folder: 6_Latency_.csv (schema compatible with the DRerio unit), 7_FrameLedger_.csv, 8_LatencyMeta_.json. The ledger exists because neither live_frame_count nor the mp4 index is a camera frame index -- frames are skipped at get_frame() and dropped at both queues. Drops are now counted instead of silent. Freezes found on the rig - The live preview ran cv2.imshow/waitKey/destroyAllWindows on the processing thread while Tk owned the main thread. Preview now renders through Tk via root.after, throttled to 20 fps and downscaled to 800 px on the worker, dropping frames rather than blocking. - Arduino.connect() set timeout=0.25 but left write_timeout at pyserial's default of None, which on Windows is WriteFile + GetOverlappedResult(bWait=True) -- an unbounded wait. One stalled write froze the analysis thread for 60 s, until the port was closed. The session log made this unambiguous: arduino.command.sent, which is emitted immediately after ser.write(), appeared at the instant of arduino.connection.closed. write_timeout is now set and SerialTimeoutException is reported as a lost trigger. - Serial I/O moved off the analysis thread entirely. send_command_async() queues (maxsize 8) and a dedicated ArduinoTxThread writes, so a stalled port costs counted, logged trigger drops instead of a frozen application. t_send/t_ack are still taken around the real write; the queue hop shows up in decision_to_send_ms. - join_threads() used unbounded joins, turning any stalled worker into a frozen app with no diagnostic. All joins are bounded at 5 s and name the offending thread. Both live loops now log exceptions instead of dying silently, and Camera.release() no longer calls cap.release() while the reader is still inside cap.read(). CAP_PROP_BUFFERSIZE and the free-running reader thread are deliberately left alone: the architectural difference from the DRerio unit is real and should be measured, not normalised away before measuring. Verified: ruff clean, 44 tests pass, including 10 new ones whose FakeSerial models the stale-buffer ordering -- three of them fail when the old ACK ordering is restored. Co-Authored-By: Claude Opus 5 --- src/zebtrack/core/controller.py | 85 +++++--- src/zebtrack/core/detector.py | 22 +++ src/zebtrack/io/arduino.py | 141 ++++++++++++-- src/zebtrack/io/camera.py | 67 ++++++- src/zebtrack/io/recorder.py | 21 +- src/zebtrack/latency_logging.py | 270 ++++++++++++++++++++++---- src/zebtrack/ui/gui.py | 140 +++++++++++-- tests/test_latency_instrumentation.py | 192 ++++++++++++++++++ 8 files changed, 838 insertions(+), 100 deletions(-) create mode 100644 tests/test_latency_instrumentation.py 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..4e9b414 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,83 @@ 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: + log.error("arduino.command.dropped_queue_full", command=box_number) + 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 +169,53 @@ 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() 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. + log.error("arduino.command.write_timeout", command=command_num) + 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 +228,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..065aeef 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,18 @@ 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") + write_fps = float(fps) if fps 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=bool(fps), + ) 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..51ac1ed 100644 --- a/src/zebtrack/latency_logging.py +++ b/src/zebtrack/latency_logging.py @@ -1,63 +1,257 @@ -"""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"] + + +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.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.""" + 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, + "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).""" + s = _SESSION + if s is None: + return + with _LOCK: + 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' or 'analysis'.""" + s = _SESSION + if s is None: + return + with _LOCK: + if kind == "video": + s.video_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. + """ + s = _SESSION + if s is None: + return None + try: + with _LOCK: + idx = s.video_write_index + s.video_write_index += 1 + s._lw.writerow([idx, frame, cam_seq, f"{t_capture:.6f}"]) + 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. -def log_trigger(cmd, t_send, t_ack, frame_t0=None): - """Grava uma linha por gatilho. Chamado de dentro de Arduino.send_command().""" + ``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. + """ + s = _SESSION + if s is None: + return try: with _LOCK: - _ensure() + s.event_id += 1 + eid = s.event_id serial_ms = (t_ack - t_send) * 1000.0 - f2a = ((t_ack - frame_t0) * 1000.0) if frame_t0 else "" - _w.writerow( + f2a = (t_ack - frame_t0) * 1000.0 if frame_t0 else "" + c2d = (t_decision - frame_t0) * 1000.0 if (t_decision and frame_t0) else "" + d2s = (t_send - t_decision) * 1000.0 if t_decision else "" + 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"{frame_t0:.6f}" if frame_t0 else "", + f"{t_decision:.6f}" if t_decision else "", f"{t_send:.6f}", f"{t_ack:.6f}", + "" if ack_ok is None else bool(ack_ok), + ack_text, + f"{c2d:.3f}" if c2d != "" else "", + f"{d2s:.3f}" if d2s != "" else "", f"{serial_ms:.3f}", f"{f2a:.3f}" if f2a != "" else "", ] ) - _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..69a0e1e --- /dev/null +++ b/tests/test_latency_instrumentation.py @@ -0,0 +1,192 @@ +"""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) From d906120ae0134de2d7bc1f2c37a11e8090ee6752 Mon Sep 17 00:00:00 2001 From: MarkSant Date: Sat, 8 Aug 2026 18:17:02 -0300 Subject: [PATCH 2/2] Address review: session teardown race, unlogged timeouts, 0.0 timestamps Follow-up to the Copilot review on #21. Three of the four points were real; the fourth is addressed differently and the reasoning is recorded here. Session teardown race (latency_logging.py). note_capture, note_drop, log_video_frame and log_trigger all read _SESSION before taking _LOCK, so a call that had already captured the handle could write to it after stop_session cleared the global and closed the files. All four now read _SESSION under the lock, and stop_session holds the lock across the whole teardown -- snapshot, sidecar write and close -- so a logging call either completes against a live session or sees None and skips. The new concurrency test shows the impact was not theoretical: against the old code the ledger holds 13779 rows while the metadata claims 13777. Unlogged write timeouts (arduino.py, latency_logging.py). A SerialTimeoutException returned False without writing a row, so a trigger the firmware may never have acted on was indistinguishable from one never attempted -- in a latency study that is a silently dropped data point. log_trigger now accepts t_ack=None and emits the row with the acknowledgement columns empty and ack_ok false. Triggers dropped because the TX queue is full are counted as trigger_queue_drops in the metadata, closing the same gap on the path added by the async dispatcher. 0.0 as a legal timestamp (latency_logging.py). perf_counter's epoch is arbitrary, so 0.0 is a reading rather than an absence, but frame_t0 and t_decision were tested for truthiness and would have blanked the latency columns for it. All timestamp handling is now explicit about None, via a single _fmt helper. Declined as proposed (recorder.py). The review asked for a bare None check on fps. That would let fps=0.0 through to VideoWriter and produce a container no player can time, which is worse than the truthiness bug it replaces. Written instead as `fps is not None and float(fps) > 0`: explicit about None, still rejecting values that cannot be stamped. Verified: ruff clean, 47 tests pass. The three new tests fail against the pre-review code. Co-Authored-By: Claude Opus 5 --- src/zebtrack/io/arduino.py | 25 ++++- src/zebtrack/io/recorder.py | 10 +- src/zebtrack/latency_logging.py | 142 +++++++++++++++++--------- tests/test_latency_instrumentation.py | 85 +++++++++++++++ 4 files changed, 209 insertions(+), 53 deletions(-) diff --git a/src/zebtrack/io/arduino.py b/src/zebtrack/io/arduino.py index 4e9b414..b3b3206 100644 --- a/src/zebtrack/io/arduino.py +++ b/src/zebtrack/io/arduino.py @@ -127,7 +127,10 @@ def send_command_async(self, box_number: int, **kwargs) -> bool: 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: @@ -169,6 +172,9 @@ def send_command( if self.ser and self.ser.is_open: command = f"{command_num}\n" + # 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 @@ -181,8 +187,25 @@ def send_command( 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. + # 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) diff --git a/src/zebtrack/io/recorder.py b/src/zebtrack/io/recorder.py index 065aeef..3b62395 100644 --- a/src/zebtrack/io/recorder.py +++ b/src/zebtrack/io/recorder.py @@ -64,13 +64,19 @@ 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") - write_fps = float(fps) if fps else float(settings.video_processing.fps) + # 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=bool(fps), + measured=use_measured, ) self.video_writer = cv2.VideoWriter( video_filename, diff --git a/src/zebtrack/latency_logging.py b/src/zebtrack/latency_logging.py index 51ac1ed..e4c4939 100644 --- a/src/zebtrack/latency_logging.py +++ b/src/zebtrack/latency_logging.py @@ -58,6 +58,11 @@ _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.""" @@ -69,6 +74,7 @@ def __init__(self, folder, base_name, meta=None): 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 @@ -116,34 +122,43 @@ def start_session(folder, base_name, meta=None): def stop_session(extra_meta=None): - """Flush and close the session, writing the metadata sidecar.""" + """Flush and close the session, writing the metadata sidecar. + + 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, - "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() + 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): @@ -158,10 +173,10 @@ def measured_fps_from(s): def note_capture(t_capture): """Record a consumed frame's capture timestamp (for the fps estimate).""" - s = _SESSION - if s is None: - return 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 @@ -169,13 +184,15 @@ def note_capture(t_capture): def note_drop(kind): - """Count a dropped frame. kind is 'video' or 'analysis'.""" - s = _SESSION - if s is None: - return + """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 @@ -186,14 +203,14 @@ def log_video_frame(frame, cam_seq, t_capture): Returns the 0-based index this frame will occupy in the mp4, which is what makes the video<->pipeline mapping exact. """ - s = _SESSION - if s is None: - return None try: with _LOCK: + 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, f"{t_capture:.6f}"]) + s._lw.writerow([idx, frame, cam_seq, _fmt(t_capture)]) s._lf.flush() return idx except Exception: @@ -219,18 +236,43 @@ def log_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. """ - s = _SESSION - if s is None: - return 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 - f2a = (t_ack - frame_t0) * 1000.0 if frame_t0 else "" - c2d = (t_decision - frame_t0) * 1000.0 if (t_decision and frame_t0) else "" - d2s = (t_send - t_decision) * 1000.0 if t_decision else "" + 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, @@ -240,16 +282,16 @@ def log_trigger( "" if roi is None else roi, "" if edge is None else edge, cmd, - f"{frame_t0:.6f}" if frame_t0 else "", - f"{t_decision:.6f}" if t_decision else "", - f"{t_send:.6f}", - f"{t_ack:.6f}", + _fmt(frame_t0), + _fmt(t_decision), + _fmt(t_send), + _fmt(t_ack), "" if ack_ok is None else bool(ack_ok), ack_text, - f"{c2d:.3f}" if c2d != "" else "", - f"{d2s:.3f}" if d2s != "" else "", - f"{serial_ms:.3f}", - f"{f2a:.3f}" if f2a != "" else "", + _fmt(c2d, 3), + _fmt(d2s, 3), + _fmt(serial_ms, 3), + _fmt(f2a, 3), ] ) s._tf.flush() diff --git a/tests/test_latency_instrumentation.py b/tests/test_latency_instrumentation.py index 69a0e1e..d318b13 100644 --- a/tests/test_latency_instrumentation.py +++ b/tests/test_latency_instrumentation.py @@ -190,3 +190,88 @@ def test_session_files_land_in_the_recording_folder(tmp_path): 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)