"""OpenFlexure Microscope OpenCV Camera This module defines a Thing that is responsible for using the stage and camera together to perform an autofocus routine. See repository root for licensing information. """ from __future__ import annotations import io import json import logging from typing import Literal, Optional from threading import Thread import time import cv2 import numpy as np import piexif from scipy.ndimage import gaussian_filter import labthings_fastapi as lt from . import BaseCamera, JPEGBlob, ArrayModel from ..stage import BaseStage # The ratio between "motor" steps and pixels # higher related to a faster movement RATIO = 0.2 class SimulatedCamera(BaseCamera): """A Thing representing an OpenCV camera""" _stage: Optional[BaseStage] = None _server: Optional[lt.ThingServer] = None def __init__( self, shape: tuple[int, int, int] = (600, 800, 3), glyph_shape: tuple[int, int, int] = (51, 51, 3), canvas_shape: tuple[int, int, int] = (3000, 4000, 3), frame_interval: float = 0.1, ): self.shape = shape self.glyph_shape = glyph_shape self.canvas_shape = canvas_shape self.frame_interval = frame_interval self._capture_thread: Optional[Thread] = None self._capture_enabled = False self.generate_sprites() self.generate_blobs() self.generate_canvas() def generate_sprites(self): """Generate sprites to populate the image""" self.sprites = [] black = np.zeros(self.glyph_shape, dtype=np.uint8) x = np.arange(black.shape[0]) y = np.arange(black.shape[1]) rr = np.sqrt((x[:, None] - np.mean(x)) ** 2 + (y[None, :] - np.mean(y)) ** 2) for i in [5, 7, 9, 11, 13, 15]: sprite = black.copy() sprite[rr < i] = 255 self.sprites.append(sprite) def generate_blobs(self, n_blobs: int = 1000): """Generate coordinates of blobs Blobs are characterised by X, Y, sprite We also generate a KD tree to rapidly find blobs in an image """ self.blobs = np.zeros((n_blobs, 3)) rng = np.random.default_rng() w = np.max(self.glyph_shape) self.blobs[:, 0] = rng.uniform(w / 2, self.canvas_shape[0] - w / 2, n_blobs) self.blobs[:, 1] = rng.uniform(w / 2, self.canvas_shape[1] - w / 2, n_blobs) self.blobs[:, 2] = rng.choice(len(self.sprites), n_blobs) def generate_canvas(self): """Generate a blank canvas""" self.canvas = np.zeros(self.canvas_shape, dtype=np.uint8) self.canvas[...] = 255 w, h, _ = self.glyph_shape for x, y, sprite in self.blobs: self.canvas[ int(x) - w // 2 : int(x) - w // 2 + w, int(y) - h // 2 : int(y) - h // 2 + h, ] -= self.sprites[int(sprite)] def generate_image(self, pos: tuple[int, int, int]): """Generate an image with blobs based on supplied coordinates""" canvas_width, canvas_height, _ = self.canvas_shape image_width, image_height, _ = self.shape pos = tuple(x * RATIO for x in pos) top_left = ( int(pos[0]) - image_width // 2 - canvas_width // 2, int(pos[1]) - image_height // 2 - canvas_height // 2, ) x_slice = slice(top_left[0], top_left[0] + self.shape[0]) y_slice = slice(top_left[1], top_left[1] + self.shape[1]) z_slice = slice(None) focused_image = self.canvas[(x_slice, y_slice, z_slice)] image = gaussian_filter( focused_image, sigma=np.abs(pos[2]) / 5, axes=(0, 1), ) if image.shape != self.shape: raise ValueError( f"Image shape {image.shape} does not match intended shape {self.shape}" ) return image def attach_to_server( self, server: lt.ThingServer, path: str, setting_storage_path: str ): self._server = server return super().attach_to_server(server, path, setting_storage_path) def get_stage_position(self): if not self._stage and self._server: self._stage = self._server.things["/stage/"] return self._stage.instantaneous_position def generate_frame(self): """Generate a frame with blobs based on the stage coordinates""" try: pos = self.get_stage_position() except Exception as e: print(f"Failed to get stage position: {e}") pos = {"x": 0, "y": 0, "z": 0} return self.generate_image((pos["y"], pos["x"], pos["z"])) def __enter__(self): self._capture_enabled = True self._capture_thread = Thread(target=self._capture_frames) self._capture_thread.start() return self def __exit__(self, _exc_type, _exc_value, _traceback): if self.stream_active: self._capture_enabled = False self._capture_thread.join() @lt.thing_property def stream_active(self) -> bool: """Whether the MJPEG stream is active""" if self._capture_enabled and self._capture_thread: return self._capture_thread.is_alive() return False def _capture_frames(self): portal = lt.get_blocking_portal(self) while self._capture_enabled: time.sleep(self.frame_interval) try: frame = self.generate_frame() jpeg = cv2.imencode(".jpg", frame)[1].tobytes() self.mjpeg_stream.add_frame(jpeg, portal) jpeg_lores = cv2.imencode(".jpg", cv2.resize(frame, (320, 240)))[ 1 ].tobytes() self.lores_mjpeg_stream.add_frame(jpeg_lores, portal) except Exception as e: logging.error(f"Failed to capture frame: {e}, retrying...") @lt.thing_action def capture_array( self, resolution: Literal["main", "full"] = "full", ) -> ArrayModel: """Acquire one image from the camera and return as an array This function will produce a nested list containing an uncompressed RGB image. It's likely to be highly inefficient - raw and/or uncompressed captures using binary image formats will be added in due course. """ logging.warning(f"Simulation camera doen't respect {resolution} setting") return self.generate_frame() @lt.thing_action def capture_jpeg( self, metadata_getter: lt.deps.GetThingStates, resolution: Literal["main", "full"] = "main", ) -> JPEGBlob: """Acquire one image from the camera and return as a JPEG blob This function will produce a JPEG image. """ logging.warning(f"Simulation camera doen't respect {resolution} setting") frame = self.capture_array() jpeg = cv2.imencode(".jpg", frame)[1].tobytes() exif_dict = { "Exif": { piexif.ExifIFD.UserComment: json.dumps(metadata_getter()).encode( "utf-8" ) }, "GPS": {}, "Interop": {}, "1st": {}, "thumbnail": None, } output = io.BytesIO() piexif.insert(piexif.dump(exif_dict), jpeg, output) return JPEGBlob.from_bytes(output.getvalue())