From 800aea21dcda870d022ab9aed6dbf7fb01f8886b Mon Sep 17 00:00:00 2001 From: jaknapper Date: Tue, 12 May 2026 12:03:31 +0100 Subject: [PATCH] Move to async monitoring --- .../things/camera/picamera.py | 150 +++++++++++++----- 1 file changed, 114 insertions(+), 36 deletions(-) diff --git a/src/openflexure_microscope_server/things/camera/picamera.py b/src/openflexure_microscope_server/things/camera/picamera.py index d58d5797..f2735758 100644 --- a/src/openflexure_microscope_server/things/camera/picamera.py +++ b/src/openflexure_microscope_server/things/camera/picamera.py @@ -22,7 +22,6 @@ import json import logging import os import tempfile -import threading import time from contextlib import contextmanager from threading import RLock @@ -169,6 +168,7 @@ class StreamingPiCamera2(BaseCamera): self._sensor_info = SUPPORTED_CAMS_SENSOR_INFO[camera_board] self._picamera_lock = RLock() self._picamera = None + self._framerate_monitor_running = False # Load the tuning file for the specified sensor mode. self.default_tuning = tf_utils.load_default_tuning( @@ -550,15 +550,35 @@ class StreamingPiCamera2(BaseCamera): with self._streaming_picamera() as cam: cam.capture_metadata() - def _framerate_worker( + async def _monitor_framerate( self, duration: float, output_dir: str, sample_interval: float = 0.1, ) -> None: - """Worker thread to measure framerate and save capture metadata.""" + """Monitor the delivered MJPEG frame rate asynchronously. + + This method listens to frames arriving on self.mjpeg_stream and records + timing information without directly interacting with the underlying + ``Picamera2`` instance. + + Writes a CSV log file with timestamp, frame count, and size of the most + recent MJPEG frame in bytes + + Monitoring stops automatically after ``duration`` seconds, or earlier if + ``self._framerate_monitor_running`` is set to ``False``. + + :param duration: Total monitoring duration in seconds. + :param output_dir: Directory in which the CSV log file will be written. + :param sample_interval: Time interval in seconds between FPS samples. + Default = 0.1s. + + :raises Exception: Any unexpected exception is logged before the monitor + exits. + """ try: os.makedirs(output_dir, exist_ok=True) + timestamp = time.strftime("%Y%m%d_%H%M%S") log_path = os.path.join( output_dir, @@ -570,58 +590,116 @@ class StreamingPiCamera2(BaseCamera): last_sample_frames = 0 frames = 0 - with open(log_path, "w", newline="") as f: - writer = csv.writer(f) - writer.writerow(["timestamp", "frame_count", "instant_fps"]) + self._framerate_monitor_running = True - with self._streaming_picamera() as cam: - while time.time() - start_time < duration: - cam.capture_metadata() - frames += 1 + LOGGER.info( + "Starting framerate monitor for %.2fs -> %s", + duration, + log_path, + ) - now = time.time() + with open(log_path, "w", newline="") as csv_file: + writer = csv.writer(csv_file) - if now - last_sample_time >= sample_interval: - interval = now - last_sample_time - instant_fps = (frames - last_sample_frames) / interval + writer.writerow( + [ + "timestamp", + "frame_count", + "frame_size_bytes", + ] + ) - writer.writerow([now, frames, instant_fps]) - f.flush() + async for frame in self.mjpeg_stream.frame_async_generator(): + if not self._framerate_monitor_running: + LOGGER.info("Framerate monitor stopped externally") + break - LOGGER.info(f"[Framerate test] FPS ~ {instant_fps:.2f}") + now = time.time() - last_sample_time = now - last_sample_frames = frames + frames += 1 + frame_size = len(frame) - total_time = time.time() - start_time - avg_fps = frames / total_time + if now - last_sample_time >= sample_interval: + interval = now - last_sample_time - LOGGER.info( - f"[Framerate test complete] " - f"Frames={frames}, Avg FPS={avg_fps:.2f}" - ) + instant_fps = (frames - last_sample_frames) / interval + + writer.writerow( + [ + now, + frames, + frame_size, + ] + ) + + csv_file.flush() + + LOGGER.info( + "[Framerate test] FPS=%.2f, size=%d B", + instant_fps, + frame_size, + ) + + last_sample_time = now + last_sample_frames = frames + + if now - start_time >= duration: + LOGGER.info("Framerate monitor duration reached") + break + + total_time = time.time() - start_time + avg_fps = frames / total_time if total_time > 0 else 0 + + writer.writerow([]) + writer.writerow(["TOTAL_DURATION", total_time]) + writer.writerow(["TOTAL_FRAMES", frames]) + writer.writerow(["AVG_FPS", avg_fps]) + + LOGGER.info( + "[Framerate test complete] Frames=%d, Avg FPS=%.2f", + frames, + avg_fps, + ) except Exception: - LOGGER.exception("Framerate worker crashed") + LOGGER.exception("Framerate monitor crashed") + + finally: + self._framerate_monitor_running = False @lt.action def record_framerate_async( self, duration: float = 5.0, - output_dir: Optional[str] = "framerate", + output_dir: str = "framerate", ) -> str: - """Start a background framerate recording session.""" - if output_dir is None: - output_dir = tempfile.mkdtemp(prefix="framerate_test_") + """Start asynchronous MJPEG framerate monitoring. - thread = threading.Thread( - target=self._framerate_worker, - args=(duration, output_dir), - daemon=True, + This action schedules ``_monitor_framerate`` on the Thing server event loop + and immediately returns. Monitoring occurs in the background and does not + block the calling thread. + + The monitor observes frames delivered through ``self.mjpeg_stream`` rather + than directly querying the camera hardware. This makes it suitable for + measuring effective stream throughput under real operating conditions. + + :param duration: Monitoring duration in seconds. Default = 5.0s. + :param output_dir: Directory where CSV log files will be written. + Default = ``"framerate"``. + + :returns: The output directory containing generated log files. + """ + self._thing_server_interface.start_async_task_soon( + self._monitor_framerate, + duration, + output_dir, ) - thread.start() - LOGGER.info(f"Started framerate test for {duration}s -> {output_dir}") + LOGGER.info( + "Started async framerate monitor for %.2fs -> %s", + duration, + output_dir, + ) return output_dir