diff --git a/src/instamatic/camera/camera_serval.py b/src/instamatic/camera/camera_serval.py index 89eac22f..8aa775b5 100644 --- a/src/instamatic/camera/camera_serval.py +++ b/src/instamatic/camera/camera_serval.py @@ -1,15 +1,21 @@ from __future__ import annotations import atexit +import contextlib +import json import logging import math +import socket from io import BytesIO -from itertools import batched -from typing import Generator, List, Optional, Sequence, Tuple, Union +from math import prod +from threading import Thread +from typing import Iterator, Optional, Sequence, Tuple +from urllib.parse import urlparse import numpy as np import tifffile from serval_toolkit.camera import Camera as ServalCamera +from typing_extensions import TypeAlias from instamatic.camera.camera_base import CameraBase @@ -20,7 +26,20 @@ # 2. `java -jar .\server\serv-2.1.3.jar` # 3. launch `instamatic` -Ignore = object() # sentinel object: informs `_get_images` to get a single image + +# By default, movies are requested image-by-image client-side which can be slow. +# Setting this var or using `camera.yaml` equivalent causes movies to be streamed +# by server via TCP which is faster, but may require tweaking firewall settings. +STREAM_MOVIES_VIA_TCP: bool = False + +Movie: TypeAlias = Iterator[np.ndarray] + + +def _local_ip_for(remote_host: str, remote_port: int) -> str: + """Return the local IP used to reach (remote_host, remote_port).""" + with contextlib.closing(socket.socket(socket.AF_INET, socket.SOCK_DGRAM)) as s: + s.connect((remote_host, remote_port)) + return s.getsockname()[0] class CameraServal(CameraBase): @@ -31,16 +50,66 @@ class CameraServal(CameraBase): MAX_EXPOSURE = 10.0 BAD_EXPOSURE_MSG = 'Requested exposure exceeds native Serval support (>0-10s)' - def __init__(self, name='serval'): - """Initialize camera module.""" + def __init__(self, name='serval') -> None: + """Initialize camera module, vars, establish connection & cleanup.""" super().__init__(name) - self.establish_connection() - self.dead_time = ( - self.detector_config['TriggerPeriod'] - self.detector_config['ExposureTime'] - ) + + self.tcp_dest: dict[str, str] = {} # used if STREAM_MOVIES_VIA_TCP=True + self.tcp_listener: Optional[socket.socket] = None # use tcp if not None + + dc = self.detector_config # noqa: loaded from a camera/file.yaml + self.dead_time = dc['TriggerPeriod'] - dc['ExposureTime'] + self.movie_bufsize = 2 * 4 * prod(self.dimensions) + self.null_image = np.zeros(shape=self.dimensions, dtype=np.uint32) + + self.previous_config: dict = {} # used to revert to default after movie + self.previous_destination: dict = {} + self.conn: ServalCamera = self.establish_connection() + logger.info(f'Camera {self.get_name()} initialized') atexit.register(self.release_connection) + def establish_connection(self) -> ServalCamera: + """Establish cam connection; "Missing" attrs are read from config.""" + http_url = urlparse(self.url) + http_dest = {'Base': 'http://localhost', 'Format': 'tiff', 'Mode': 'count'} + f = dict(bpc_file_path=self.bpc_file_path, dacs_file_path=self.dacs_file_path) + + conn = ServalCamera() + conn.connect(http_url.geturl()) + conn.set_chip_config_files(**f) + conn.set_detector_config(**self.detector_config) + conn.destination = {'Image': [http_dest]} + + if getattr(self, 'stream_movies_via_tcp') or STREAM_MOVIES_VIA_TCP: + self.tcp_listener = socket.socket(socket.AF_INET, socket.SOCK_STREAM) + self.tcp_listener.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) + self.tcp_listener.settimeout(1.0) + self.tcp_listener.bind(('0.0.0.0', 0)) + self.tcp_listener.listen(1) + tcp_port = self.tcp_listener.getsockname()[1] + local_ip = _local_ip_for(http_url.hostname or 'localhost', tcp_port) + tcp_base = f'tcp://connect@{local_ip}:{tcp_port}' + self.tcp_dest = {'Base': tcp_base, 'Format': 'jsonimage', 'Mode': 'count'} + + return conn + + def release_connection(self) -> None: + """Release the connection to the camera (HTTP & TCP if applicable).""" + self.conn.measurement_stop() + if self.tcp_listener is not None: + self.tcp_listener.close() + logger.info(f"Connection to camera '{self.get_name()}' released") + + def set_detector_config(self, **kwargs) -> None: + """Set detector config while inferring about missing config params.""" + if 'TriggerMode' not in kwargs: + tm = 'AUTOTRIGSTART_TIMERSTOP' if self.dead_time else 'CONTINUOUS' + kwargs['TriggerMode'] = tm + if 'TriggerPeriod' not in kwargs and 'ExposureTime' in kwargs: + kwargs['TriggerPeriod'] = kwargs['ExposureTime'] + self.dead_time + self.conn.set_detector_config(**kwargs) + def get_image(self, exposure: Optional[float] = None, **kwargs) -> np.ndarray: """Image acquisition interface. If the exposure is not given, the default value is read from the config file. Binning is ignored. @@ -48,147 +117,156 @@ def get_image(self, exposure: Optional[float] = None, **kwargs) -> np.ndarray: exposure: `float` or `None` Exposure time in seconds. """ - return self._get_images(n_frames=Ignore, exposure=exposure, **kwargs) - - def _get_images( - self, - n_frames: Union[int, Ignore], - exposure: Optional[float] = None, - **kwargs, - ) -> Union[np.ndarray, List[np.ndarray]]: - """General media acquisition dispatcher for other protected methods.""" - n: int = 1 if n_frames is Ignore else n_frames e: float = self.default_exposure if exposure is None else exposure - if n_frames == 0: # single image is communicated via n_frames = Ignore - return [] - - elif e < self.MIN_EXPOSURE: - logger.warning('%s: %d', self.BAD_EXPOSURE_MSG, e) - if n_frames is Ignore: - return self._get_image_null(exposure=e, **kwargs) - return [self._get_image_null(exposure=e, **kwargs) for _ in range(n)] + if e < self.MIN_EXPOSURE: + logger.warning('%s: %g', self.BAD_EXPOSURE_MSG, e) + return self.null_image elif e > self.MAX_EXPOSURE: - logger.warning('%s: %d', self.BAD_EXPOSURE_MSG, e) - n1 = math.ceil(e / self.MAX_EXPOSURE) - e = (e + self.dead_time) / n1 - self.dead_time - images = self._get_image_stack(n_frames=n * n1, exposure=e, **kwargs) - if n_frames is Ignore: - return self._spliced_sum(images, exposure=e) - return [self._spliced_sum(i, exposure=e) for i in batched(images, n1)] - - else: # if exposure is within limits - if n_frames is Ignore: - return self._get_image_single(exposure=e, **kwargs) - return self._get_image_stack(n_frames=n, exposure=e, **kwargs) + logger.warning('%s: %g', self.BAD_EXPOSURE_MSG, e) + n = math.ceil(e / self.MAX_EXPOSURE) + e = (e + self.dead_time) / n - self.dead_time + images = list(self.get_movie(n_frames=n, exposure=e)) + return self._spliced_sum(images, exposure=e) - def _spliced_sum(self, arrays: Sequence[np.ndarray], exposure: float) -> np.ndarray: - """Sum a series of arrays while applying a dead time correction.""" - array_sum = sum(arrays, np.zeros_like(arrays[0])) - total_exposure = len(arrays) * exposure + (len(arrays) - 1) * self.dead_time - live_fraction = len(arrays) * exposure / total_exposure - return (array_sum / live_fraction).astype(arrays[0].dtype) - - def _get_image_null(self, **_) -> np.ndarray: - logger.debug('Creating a synthetic image with zero counts') - return np.zeros(shape=self.get_image_dimensions(), dtype=np.int32) - - def _get_image_single(self, exposure: float, **_) -> np.ndarray: - """Request a single frame in the mode in a trigger collection mode.""" logger.debug(f'Collecting a single image with exposure {exposure} s') - # Upload exposure settings (Note: will do nothing if no change in settings) - self.conn.set_detector_config( - ExposureTime=exposure, - TriggerPeriod=exposure + self.dead_time, - ) - - # Check if measurement is running. If not: start + self.conn.set_detector_config(ExposureTime=e, TriggerPeriod=e + self.dead_time) db = self.conn.dashboard if db['Measurement'] is None or db['Measurement']['Status'] != 'DA_RECORDING': self.conn.measurement_start() - - # Start the acquisition self.conn.trigger_start() - # Request a frame. Will be streamed *after* the exposure finishes response = self.conn.get_request('/measurement/image') return tifffile.imread(BytesIO(response.content)) - def _get_image_stack(self, n_frames: int, exposure: float, **_) -> list[np.ndarray]: - """Get a series of images in a mode with minimal dead time.""" - return list(self.get_movie(n_frames=n_frames, exposure=exposure)) + def _spliced_sum(self, arrays: Sequence[np.ndarray], exposure: float) -> np.ndarray: + """Sum a series of arrays while applying a dead time correction.""" + array_sum = sum(arrays, np.zeros_like(arrays[0])) + total_exposure = len(arrays) * exposure + (len(arrays) - 1) * self.dead_time + live_fraction = len(arrays) * exposure / total_exposure + return (array_sum / live_fraction).astype(arrays[0].dtype) - def get_movie( - self, n_frames: int, exposure: Optional[float] = None, **kwargs - ) -> Generator[np.ndarray, None, None]: - """A generator yielding images using a mode with minimal dead time. If - the exposure is not given, the default value is read from the config - file. Binning is ignored. + def get_movie(self, n_frames: int, exposure: Optional[float] = None, **_) -> Movie: + """Yield `n_frames` images received via HTTP (convenient) or TCP + (fast); If the exposure is None, the default is read from the config. + Binning is ignored. Setup is eager, start is delayed until iteration. n_frames: `int` Number of frames to collect exposure: `float` or `None` Exposure time in seconds. """ - logger.debug(f'Collecting {n_frames}-frame movie with exposure {exposure} s') - mode = 'AUTOTRIGSTART_TIMERSTOP' if self.dead_time else 'CONTINUOUS' + logger.debug(f'Collecting {n_frames}-frame movie with {exposure=} s') + e: float = self.default_exposure if exposure is None else exposure self.conn.measurement_stop() - previous_config = self.conn.detector_config + get_movie = self._get_movie_tcp if self.tcp_listener else self._get_movie_http + return get_movie(n_frames=n_frames, exposure=e) + + def _get_movie_http(self, n_frames: int, exposure: float) -> Movie: + """Convenient method: poll frames via HTTP using the client socket.""" + self.previous_config = self.conn.detector_config + self.set_detector_config(ExposureTime=exposure, nTriggers=n_frames) + return self._get_movie_inner(n_frames=n_frames) + + def _get_movie_tcp(self, n_frames: int, exposure: float) -> Movie: + """Fast: stream frames directly via TCP if STREAM_MOVIES_VIA_TCP.""" + self.previous_config = self.conn.detector_config + self.previous_destination = self.conn.destination + self.conn.destination = {'Image': [self.tcp_dest]} + self.set_detector_config(ExposureTime=exposure, nTriggers=n_frames) + return self._get_movie_inner(n_frames=n_frames) + + def _get_movie_inner(self, n_frames: int) -> Movie: + """Frame generator; Separate method because `yield` keyword makes this + code execution delayed; In other words, setup runs when `get_movie` is + called, but this method only when returned iterator is iterated.""" try: - self.conn.set_detector_config( - TriggerMode=mode, - ExposureTime=exposure, - TriggerPeriod=exposure + self.dead_time, - nTriggers=n_frames, - ) - self.conn.measurement_start() - for i in range(n_frames): - response = self.conn.get_request('/measurement/image') - yield tifffile.imread(BytesIO(response.content)) + if self.tcp_listener: + Thread(target=self.conn.measurement_start, daemon=True).start() + self.tcp_listener.settimeout(5.0) + sock, _ = self.tcp_listener.accept() + sock.settimeout(1.1 * self.MAX_EXPOSURE) + with sock: + yield from ServalMovieDeserializer(sock, n_frames, self.movie_bufsize) + else: + self.conn.measurement_start() + for _ in range(n_frames): + response = self.conn.get_request('/measurement/image') + yield tifffile.imread(BytesIO(response.content)) finally: - self.conn.measurement_stop() - self.conn.set_detector_config(**previous_config) + try: + self.conn.measurement_stop() + except Exception as ex: + logger.error(f'Error stopping measurement: {ex}') + try: + self.conn.set_detector_config(**self.previous_config) + if self.tcp_listener: + self.conn.destination = self.previous_destination + except Exception as ex: + logger.error(f'Error restoring config and destination: {ex}') def get_image_dimensions(self) -> Tuple[int, int]: """Get the binned dimensions reported by the camera.""" binning = self.get_binning() dim_x, dim_y = self.get_camera_dimensions() - - dim_x = int(dim_x / binning) - dim_y = int(dim_y / binning) - - return dim_x, dim_y - - def establish_connection(self) -> None: - """Establish connection to the camera.""" - self.conn = ServalCamera() - self.conn.connect(self.url) - self.conn.set_chip_config_files( - bpc_file_path=self.bpc_file_path, dacs_file_path=self.dacs_file_path - ) - self.conn.set_detector_config(**self.detector_config) - - self.conn.destination = { - 'Image': [ - { - # Where to place the preview files (HTTP end-point: GET localhost:8080/measurement/image) - 'Base': 'http://localhost', - # What (image) format to provide the files in. - 'Format': 'tiff', - # What data to build a frame from - 'Mode': 'count', - # 'QueueSize': 2, - } - ], - } - - def release_connection(self) -> None: - """Release the connection to the camera.""" - self.conn.measurement_stop() - name = self.get_name() - msg = f"Connection to camera '{name}' released" - logger.info(msg) + return int(dim_x / binning), int(dim_y / binning) + + +class ServalMovieDeserializer(Movie): + """Deserializes Serval camera TCP byte stream from socket into images.""" + + def __init__(self, sock: socket.socket, n_frames: int, bufsize: int) -> None: + self.sock: socket.socket = sock + self.buffer = bytearray(bufsize) + self.view = memoryview(self.buffer) + self.used: int = 0 + self.i_frame: int = 0 + self.n_frames: int = n_frames + self.shape: tuple[int, int] = (0, 0) + self.size: int = 0 + self.dtype: np.dtype = np.dtype(np.uint32) + + def _receive_more(self) -> None: + """Attempt to receive bytes from the socket into free buffer space.""" + recv_len = self.sock.recv_into(self.view[self.used :]) + if not recv_len: + raise EOFError + self.used += recv_len + + def _receive_until(self, token: bytes) -> int: + """Recv data until `token` is found, return index after the token.""" + while True: + token_idx = self.buffer.find(token, 0, self.used) + if token_idx >= 0: + return token_idx + len(token) + self._receive_more() + + def _parse_header(self, header_size: int) -> None: + """Read shape, size, dtype of all images from the 1st frame header.""" + header_str = self.buffer[:header_size].decode('utf-8') + header_dict = json.loads(header_str) + bit_depth = header_dict['bitDepth'] + self.shape = (header_dict['height'], header_dict['width']) + self.dtype = np.dtype(f'uint{bit_depth}').newbyteorder('>') + self.size = header_dict.get('dataSize', prod(self.shape) * self.dtype.itemsize) + + def __next__(self) -> np.ndarray: + """Recv as much data as needed, return next frame from TCP stream.""" + if self.i_frame >= self.n_frames: + raise StopIteration + header_end = self._receive_until(b'}') + 1 # json image never nests "}" + if self.i_frame == 0: + self._parse_header(header_end) + while self.used < header_end + self.size: + self._receive_more() + i, j = header_end, header_end + self.size + frame = np.frombuffer(self.buffer[i:j], dtype=self.dtype).reshape(self.shape).copy() + + self.buffer[: self.used - j] = self.buffer[j : self.used] + self.used -= j + self.i_frame += 1 + return frame if __name__ == '__main__': diff --git a/src/instamatic/camera/videostream.py b/src/instamatic/camera/videostream.py index 43c97b6e..4c120851 100644 --- a/src/instamatic/camera/videostream.py +++ b/src/instamatic/camera/videostream.py @@ -65,6 +65,7 @@ def __init__(self, cam: CameraBase, callback, frametime: float = 0.05): self.stopEvent = threading.Event() self.acquireInitiateEvent = threading.Event() + self.acquireInitiateEvent2 = threading.Event() self.continuousCollectionEvent = threading.Event() def run(self): @@ -79,7 +80,9 @@ def run(self): self.callback(media, request=r) else: # isinstance(r, MovieRequest): n = r.n_frames if r.n_frames else 1 - for media in self.cam.get_movie(n_frames=n, exposure=e, binsize=b): + m = self.cam.get_movie(n_frames=n, exposure=e, binsize=b) + self.acquireInitiateEvent2.wait() + for media in m: self.callback(media, request=r) time.sleep(0) # yields thread priority to VideoStream @@ -206,16 +209,26 @@ def get_image(self, exposure=None, binsize=None) -> np.ndarray: def get_movie( self, n_frames: int, exposure=None, binsize=None ) -> Generator[np.ndarray, None, None]: + self.blocked().__enter__() # Stop the passive collection during request acquisition try: - with self.blocked(): # Stop the passive collection during request acquisition - self.grabber.request = MovieRequest(n_frames, exposure, binsize) - self.grabber.acquireInitiateEvent.set() + self.grabber.request = MovieRequest(n_frames, exposure, binsize) + self.grabber.acquireInitiateEvent.set() + except Exception: + self.blocked().__exit__(None, None, None) + + def _movie_generator() -> Generator[np.ndarray, None, None]: + try: + self.grabber.acquireInitiateEvent2.set() for _ in range(n_frames): while not self.requested: time.sleep(0) # yields thread priority to MediaGrabber yield self.requested.popleft() - finally: - self.grabber.request = None + finally: + self.grabber.request = None + self.grabber.acquireInitiateEvent2.clear() + self.blocked().__exit__(None, None, None) + + return _movie_generator() def update_frametime(self, frametime): self.frametime = frametime diff --git a/src/instamatic/config/camera/serval.yaml b/src/instamatic/config/camera/serval.yaml index 13b573c8..5cd92273 100644 --- a/src/instamatic/config/camera/serval.yaml +++ b/src/instamatic/config/camera/serval.yaml @@ -37,3 +37,4 @@ detector_config: bpc_file_path: '/home/asi/Desktop/Factory_settings/SPM-HGM/config.bpc' dacs_file_path: '/home/asi/Desktop/Factory_settings/SPM-HGM/config.dacs' url: 'http://localhost:8080' +stream_movies_via_tcp: false diff --git a/src/instamatic/config/scripts/benchmark_movie_rates.py b/src/instamatic/config/scripts/benchmark_movie_rates.py new file mode 100644 index 00000000..a3043951 --- /dev/null +++ b/src/instamatic/config/scripts/benchmark_movie_rates.py @@ -0,0 +1,437 @@ +#!/usr/bin/env python3 +"""benchmark_movie_rates.py. + +Benchmark achievable frame rates using TEMController.get_movie. + +Key behaviour: +- Uses `with ctrl.cam.blocked():` to avoid interference during timed acquisition. +- Optional warm-up cycle (single tiny-frame) before timed measurement. +- For each exposure requested, runs N rounds. Each round collects frames + until at least `--min-duration` seconds elapse (or until a safe max frames limit). +- Does not save image data; only timestamps and basic header-derived times are recorded. +- Optional light processing per frame (`--process`) to emulate CPU work (np.mean). +- Prints summary after measurements and optionally writes raw CSV. + +Use: + python benchmark_movie_rates.py --exposures 0.01,0.02 --variable-headers BeamShift --rounds 3 +""" + +from __future__ import annotations + +import argparse +import csv +import dataclasses +import math +import time +from pathlib import Path +from statistics import mean, stdev +from typing import Generator, Optional + +import matplotlib.pyplot as plt + +from instamatic.controller import TEMController, _ctrl, initialize +from instamatic.utils.iterating import pairwise + +if not _ctrl: + _ctrl: TEMController = initialize() + + +@dataclasses.dataclass +class FrameStamp: + """Timestamps and minimal metadata for a single yielded frame.""" + + frame_index: int + t0: float # perf_counter right before calling next(gen) + t1: float # perf_counter right after next(gen) returned + h0: Optional[float] = None # header-reported 'ImageGetTimeStart' if present + h1: Optional[float] = None # header-reported 'ImageGetTimeEnd' if present + + +@dataclasses.dataclass +class RoundResult: + """Results for a single timed round at one exposure and header + configuration.""" + + index: int + exposure: float + header_keys: tuple + header_keys_common: tuple + frames: list[FrameStamp] + t_start: float + t_end: float + + @property + def n_frames(self) -> int: + return len(self.frames) + + @property + def duration(self) -> float: + return self.t_end - self.t_start + + @property + def fps(self) -> float: + return self.n_frames / self.duration if self.duration > 0 else float('nan') + + def inter_frame_intervals(self) -> list[float]: + """Return list of intervals between successive t1 frame yield times.""" + return [g.t1 - f.t1 for f, g in pairwise(self.frames)] + + def mean_inter_frame_interval(self) -> Optional[float]: + ints = self.inter_frame_intervals() + return mean(ints) if ints else None + + def mean_header_duration(self) -> Optional[float]: + durations = [f.h1 - f.h0 for f in self.frames if f.h0 and f.h1] + return mean(durations) if durations else None + + +class MovieBench: + """Benchmark runner using a TEMController instance.""" + + def __init__( + self, + n_rounds: int = 3, + n_frames: int = 1000, + ) -> None: + self.rounds = int(n_rounds) + self.n_frames = int(n_frames) + + if _ctrl is None: + raise RuntimeError('No TEMController instance available.') + + if not getattr(_ctrl, 'cam', None): + raise RuntimeError("Controller has no 'cam' attribute.") + if not hasattr(_ctrl.cam, 'blocked'): + raise RuntimeError("Camera does not provide 'blocked()' context manager.") + + def _warm_up(self, exposure: float, header_keys: tuple, header_keys_common: tuple) -> None: + """Optional tiny dummy call warming camera / generator start-up.""" + + try: + hk, hkc = header_keys, header_keys_common + gen = _ctrl.get_movie(1, exposure, header_keys=hk, header_keys_common=hkc) + next(gen) + gen.close() + except (StopIteration, RuntimeError): + pass + + def run_round( + self, + exposure: float, + header_keys: tuple, + header_keys_common: tuple, + ) -> RoundResult: + frames: list[FrameStamp] = [] + hk, hkc = header_keys, header_keys_common + gen = _ctrl.get_movie(self.n_frames, exposure, header_keys=hk, header_keys_common=hkc) + t_start = time.perf_counter() + try: + for i in range(self.n_frames): + t0 = time.perf_counter() + try: + img, header = next(gen) + except StopIteration: + break + t1 = time.perf_counter() + h0 = header.get('ImageGetTimeStart') + h1 = header.get('ImageGetTimeEnd') + frames.append(FrameStamp(frame_index=i + 1, t0=t0, t1=t1, h0=h0, h1=h1)) + # _ = float(np.mean(img)) # if benchmarking later? + finally: + try: + gen.close() + except Exception: + pass + t_end = time.perf_counter() + return RoundResult( + exposure=exposure, + header_keys=header_keys, + header_keys_common=header_keys_common, + index=0, + frames=frames, + t_start=t_start, + t_end=t_end, + ) + + def run( + self, + exposures: list[float], + header_keys: tuple = (), + header_keys_common: tuple = (), + warmup: bool = True, + ) -> Generator[RoundResult]: + """Run benchmark across provided exposure times and header + configurations.""" + hk, hkc = header_keys, header_keys_common + for e in exposures: + for r in range(self.rounds): + if warmup and r == 0: + self._warm_up(exposure=e, header_keys=hk, header_keys_common=hkc) + with _ctrl.cam.blocked(): + res = self.run_round(exposure=e, header_keys=hk, header_keys_common=hkc) + res.index = r + 1 + yield res + + +# ------------------------- +# Reporting utilities +# ------------------------- +def summarize_result(result: RoundResult) -> dict: + """Return a dict with human-friendly summarized numbers for a run.""" + inter_intervals = result.inter_frame_intervals() + mean_interval = mean(inter_intervals) if inter_intervals else None + std_interval = stdev(inter_intervals) if len(inter_intervals) >= 2 else None + header_mean = result.mean_header_duration() + header_time_ratio = None + if header_mean is not None and mean_interval is not None: + header_time_ratio = header_mean / mean_interval if mean_interval > 0 else None + + dead_time_est = ( + (mean_interval - result.exposure) if (mean_interval and result.exposure) else None + ) + + return { + 'exposure_seconds': result.exposure, + 'round_index': result.index, + 'n_frames': result.n_frames, + 'duration_s': result.duration, + 'fps_measured': result.fps, + 'init_time_est_s': (result.frames[0].t1 - result.t_start) if result.frames else None, + 'mean_interframe_s': mean_interval, + 'std_interframe_s': std_interval, + 'min_interframe_s': min(inter_intervals) if inter_intervals else None, + 'max_interframe_s': max(inter_intervals) if inter_intervals else None, + 'dead_time_est_s': dead_time_est, + 'header_mean_s': header_mean, + 'header_time_ratio': header_time_ratio, + } + + +def print_run_summary(summ: dict) -> None: + print( + f'Exposure {summ["exposure_seconds"]:.6f}s | Round {summ["round_index"]} | frames={summ["n_frames"]} | duration={summ["duration_s"]:.4f}s' + ) + print(f' fps={summ["fps_measured"]:.3f} | init_time≈{summ["init_time_est_s"]:.6f}s') + print( + f' interframe mean={summ["mean_interframe_s"]:.6f}s ±{summ["std_interframe_s"] or 0:.6f}s | min={summ["min_interframe_s"]:.6f}s max={summ["max_interframe_s"]:.6f}s' + ) + if summ['dead_time_est_s'] is not None: + print(f' dead_time_est = mean_interframe - exposure = {summ["dead_time_est_s"]:.6f}s') + if summ['header_mean_s'] is not None: + print( + f' header_mean = {summ["header_mean_s"]:.6f}s | header_time_ratio = {summ["header_time_ratio"]:.3f}' + ) + + +def print_aggregated_table(results: list[RoundResult]) -> None: + """Print a CSV-like aggregated summary grouped by exposure.""" + grouped: dict[float, list[RoundResult]] = {} + for r in results: + grouped.setdefault(r.exposure, []).append(r) + + header = [ + 'exposure_s', + 'rounds', + 'frames_avg', + 'fps_mean', + 'fps_std', + 'dead_time_mean_s', + 'header_mean_s', + 'header_time_ratio', + ] + print('\nAggregated summary (CSV):') + print(','.join(header)) + + for exposure in sorted(grouped.keys()): + runs = grouped[exposure] + fps_vals = [rr.fps for rr in runs if not math.isnan(rr.fps)] + frames_avg = mean([rr.n_frames for rr in runs]) if runs else 0 + fps_mean = mean(fps_vals) if fps_vals else float('nan') + fps_std = stdev(fps_vals) if len(fps_vals) >= 2 else 0.0 + + dead_times = [] + header_means = [] + for rr in runs: + mi = rr.mean_inter_frame_interval() + if mi is not None: + dead_times.append(mi - rr.exposure) + hm = rr.mean_header_duration() + if hm is not None: + header_means.append(hm) + + dead_mean = mean(dead_times) if dead_times else float('nan') + header_mean = mean(header_means) if header_means else float('nan') + header_ratio = ( + (header_mean / (dead_mean + exposure)) + if (not math.isnan(header_mean) and not math.isnan(dead_mean)) + else float('nan') + ) + + row = [ + exposure, + len(runs), + frames_avg, + fps_mean, + fps_std, + dead_mean, + header_mean, + header_ratio, + ] + print(','.join([f'{v:.6g}' if isinstance(v, float) else str(v) for v in row])) + + +def save_raw_csv(results: list[RoundResult]) -> None: + filename = Path.cwd() / f'benchmark_movie_raw_{int(time.time())}.csv' + with filename.open('w', newline='') as fh: + writer = csv.writer(fh) + writer.writerow(['exposure_s', 'round', 'frame_index', 't0', 't1', 'h0', 'h1']) + for r in results: + for f in r.frames: + writer.writerow([r.exposure, r.index, f.frame_index, f.t0, f.t1, f.h0, f.h1]) + print(f'Raw timestamps saved to: {filename}') + + +# ------------------------- +# CLI helpers +# ------------------------- +def parse_exposures(arg: Optional[str]) -> list[float]: + """Parse comma-separated exposures or default to 1/10..1/100 s.""" + if not arg: + defaults_fps = [10, 20, 30, 40, 50, 60, 70, 80, 90, 100] + return [1.0 / f for f in defaults_fps] + parts = [p.strip() for p in arg.split(',') if p.strip()] + exposures: list[float] = [] + for p in parts: + try: + exposures.append(float(p)) + except Exception: + raise argparse.ArgumentTypeError(f'Invalid exposure value: {p}') + return exposures + + +def parse_header_keys(arg: Optional[str]) -> tuple: + if (arg is None) or (arg == ''): + return () + return tuple(x.strip() for x in arg.split(',') if x.strip()) + + +def generate_plots(results: list[RoundResult]) -> None: + """Generate basic plots: exposure vs fps, exposure vs dead-time.""" + + # Aggregate by exposure + grouped: dict[float, list[RoundResult]] = {} + for r in results: + grouped.setdefault(r.exposure, []).append(r) + + exposures = sorted(grouped.keys()) + fps_means = [] + dead_time_means = [] + + for exp in exposures: + runs = grouped[exp] + # fps mean + fps_vals = [rr.fps for rr in runs if not math.isnan(rr.fps)] + fps_means.append(mean(fps_vals) if fps_vals else float('nan')) + dead_times = [] + for rr in runs: + mi = rr.mean_inter_frame_interval() + if mi is not None: + dead_times.append(mi - rr.exposure) + dead_time_means.append(mean(dead_times) if dead_times else float('nan')) + + # ---- PLOT 1: Exposure vs FPS ---- + plt.figure() + plt.plot(exposures, fps_means, marker='o') + plt.xlabel('Exposure (s)') + plt.ylabel('FPS (measured)') + plt.title('Exposure vs Measured FPS') + plt.grid(True) + plt.show() + + # ---- PLOT 2: Exposure vs Dead Time ---- + plt.figure() + plt.plot(exposures, dead_time_means, marker='o') + plt.xlabel('Exposure (s)') + plt.ylabel('Dead time (s)') + plt.title('Exposure vs Dead Time') + plt.grid(True) + plt.show() + + +def main(argv: Optional[list[str]] = None) -> None: + parser = argparse.ArgumentParser( + description='Benchmark TEMController.get_movie to measure achievable frame rates.' + ) + parser.add_argument( + '--exposures', + type=str, + default=None, + help="Comma-separated exposures in seconds, e.g. '0.01,0.02'. Default: 1/10..1/100.", + ) + parser.add_argument( + '--variable-headers', + type=str, + default=None, + help='Comma-separated variable header keys to collect per frame. Use empty string to disable.', + ) + parser.add_argument( + '--common-headers', + type=str, + default=None, + help='Comma-separated common header keys to collect once before movie. Use empty string to disable.', + ) + parser.add_argument( + '--n_rounds', type=int, default=3, help='Rounds per exposure. Default: 3' + ) + parser.add_argument( + '--n_frames', type=int, default=10, help='Number of frames to be collected. Default: 10' + ) + parser.add_argument('--no-warmup', action='store_true', help='Disable warmup dummy frame.') + parser.add_argument( + '--plot', + action='store_true', + help='Generate simple matplotlib plots after the benchmark.', + ) + args = parser.parse_args(argv) + + exposures = parse_exposures(args.exposures) + variable_headers = parse_header_keys(args.variable_headers) + common_headers = parse_header_keys(args.common_headers) + + bench = MovieBench( + n_rounds=args.n_rounds, + n_frames=args.n_frames, + ) + + print('=== benchmark_movie_rates: starting benchmark ===') + print(f'Exposures (s): {exposures}') + print(f'Variable header keys: {variable_headers or "(none)"}') + print(f'Common header keys: {common_headers or "(none)"}') + print(f'Rounds per exposure: {args.n_rounds}') + + results: list[RoundResult] = [] + results_generator = bench.run( + exposures=exposures, + header_keys=variable_headers, + header_keys_common=common_headers, + warmup=(not args.no_warmup), + ) + + # Print per-run summaries + for res in results_generator: + summ = summarize_result(res) + print() + print_run_summary(summ) + results.append(res) + print_aggregated_table(results) + try: + save_raw_csv(results) + except PermissionError: + pass + generate_plots(results) + + print('\n=== benchmark_movie_rates: finished ===') + + +if __name__ == '__main__': + main() diff --git a/src/instamatic/controller.py b/src/instamatic/controller.py index dea63ff2..36022631 100644 --- a/src/instamatic/controller.py +++ b/src/instamatic/controller.py @@ -709,7 +709,7 @@ def get_movie( comment: str = '', header_keys: Tuple[str] = MOVIE_HEADER_KEYS_VARIABLE, header_keys_common: Tuple[str] = MOVIE_HEADER_KEYS_COMMON, - ) -> Generator[np.ndarray, None, None]: + ) -> Generator[tuple[np.ndarray, dict], None, None]: """Generate (image, header) pairs using camera's movie mode. If the exposure and binsize are not given, the default values are read from the config file. Common header info is collected before the generator @@ -732,8 +732,8 @@ def get_movie( Yields ------- - image_header: Generator[(np.ndarray, collections.ChainMap), None, None] - Generator of (numpy arrays with image data, ChainMap with + image_header: Generator[(np.ndarray, dict), None, None] + Generator of (numpy arrays with image data, dict with all the tem parameters and image attributes) pairs. Usage: diff --git a/src/instamatic/gui/gui.py b/src/instamatic/gui/gui.py index 40d19489..fde8f447 100644 --- a/src/instamatic/gui/gui.py +++ b/src/instamatic/gui/gui.py @@ -98,7 +98,6 @@ def load(self, modules, master): for location in self.locations: selected_modules = [module for module in modules if module.location == location] - is_group = len(selected_modules) > 1 if is_group: diff --git a/tests/test_serval_movie.py b/tests/test_serval_movie.py new file mode 100644 index 00000000..716ec3c6 --- /dev/null +++ b/tests/test_serval_movie.py @@ -0,0 +1,59 @@ +from __future__ import annotations + +import numpy as np + +from instamatic.camera.camera_serval import ServalMovieDeserializer + +rng = np.random.default_rng(1337) +json_image = rng.integers(low=0, high=256, size=(512, 512), dtype=np.uint16) +json_header = b"""{ + "timeAtFrame": 1655990130.181, + "frameNumber": 14, + "measurementID": "None", + "dataSize": 524288, + "bitDepth": 16, + "isPreviewSampled": true, + "thresholdID": 0, + "pixelEventNumber": 0, + "tdc1EventNumber": 0, + "tdc2EventNumber": 0, + "integrationSize": 0, + "integrationMode": "None", + "width": 512, + "height": 512, + "corrections": [] +}""" +json_bytes = json_header + bytearray(json_image.data) +movie_bufsize = 2 * 4 * 512 * 512 + + +class MockSocket: + """A simple socket-like object that serves predefined bytes.""" + + def __init__(self, data: bytes): + self._buf = memoryview(data) + self._offset = 0 + + def recv_into(self, b: memoryview) -> int: + n = min(len(b), len(self._buf) - self._offset) + if n == 0: + return 0 + b[:n] = self._buf[self._offset : self._offset + n] + self._offset += n + return n + + +def test_serval_movie_deserializer1() -> None: + """Check that ServalMovieDeserializer can yield one image correctly.""" + sock = MockSocket(json_bytes) + smd = ServalMovieDeserializer(sock, n_frames=1, bufsize=movie_bufsize) # type: ignore + image = next(smd) + np.testing.assert_array_equal(image, json_image) + + +def test_serval_movie_deserializer100() -> None: + """Check that ServalMovieDeserializer can yield 100 images correctly.""" + sock = MockSocket(json_bytes * 100) + smd = ServalMovieDeserializer(sock, n_frames=100, bufsize=movie_bufsize) # type: ignore + for image in smd: + np.testing.assert_array_equal(image, json_image)