rahix/openflexure-microscope-server
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Merge branch 'fast-sim' into 'v3'

Browse source 
Simulation Camera Improvements

Closes #638

See merge request openflexure/openflexure-microscope-server!458
This commit is contained in:
Joe Knapper 2026-01-13 14:15:35 +00:00
commit 9df3fd6360
4 changed files with 406 additions and 104 deletions
Download patch file Download diff file Expand all files Collapse all files

281
src/openflexure_microscope_server/things/camera/simulation.py
View file

@ -10,10 +10,11 @@ from __future__ import annotations
import io
import logging
import re
import time
from threading import Thread
from types import TracebackType
from typing import Literal, Optional, Self
from typing import Literal, Optional, Self, overload
import numpy as np
from PIL import Image, ImageFilter
@ -43,6 +44,68 @@ BG_COLOR = [220, 215, 217]
# Random Number Generator
RNG = np.random.default_rng()
DOWNSAMPLE = 2
# Upsample for sprites and then downsample to create sharp edges for each sprite
# as these are small and calculated once there is almost no performance penalty
# for a nice gain in quality.
SPRITE_UPSAMPLE = 4
# A list of 6 digit hex colour codes separated by ;. Allow a trailing ;
# For example, OpenFlexure pink would be #C5247F;
COLOUR_LIST_REGEX = re.compile(
r"^\s*(#[0-9a-fA-F]{6})\s*(?:;\s*(#[0-9a-fA-F]{6})\s*)*;?\s*$"
)
# regex to separate R, G and B from a 6 digit hex code with preceding #
COLOUR_REGEX = re.compile(r"^#([0-9a-f]{2})([0-9a-f]{2})([0-9a-f]{2})$")
@overload
def _downsample_shape(shape: tuple[int, int]) -> tuple[int, int]: ...
@overload
def _downsample_shape(shape: tuple[int, int, int]) -> tuple[int, int, int]: ...
def _downsample_shape(
shape: tuple[int, int] | tuple[int, int, int],
) -> tuple[int, int] | tuple[int, int, int]:
if len(shape) == 2:
return (shape[0] // DOWNSAMPLE, shape[1] // DOWNSAMPLE)
if len(shape) == 3:
return (shape[0] // DOWNSAMPLE, shape[1] // DOWNSAMPLE, shape[2])
raise ValueError("Shape should be a 2 or 3 element tuple.")
def colour_str_to_colour(colour_str: str) -> tuple[int, int, int]:
"""Convert a colour string into RGB colour values.
:param colour_str: Should be a hex colour such as #33aa33 or a list of hex
colours separated by semicolons (with optional spaces).
:return: The colour as a tuple of 3 integers from 0 to 255 in value
:raises ValueError: If the hex string is not valid. This should never happen if the
user enters a bad colour string as the colour property setter checks the
whole string regex.
"""
if ";" in colour_str:
colours = colour_str.split(";")
if len(colours) > 1 and colours[-1].strip() == "":
colours.pop(-1)
single_colour_str = colours[RNG.integers(0, len(colours))]
else:
single_colour_str = colour_str
single_colour_str = single_colour_str.lower().strip()
colour_match = COLOUR_REGEX.match(single_colour_str)
if colour_match is None:
raise ValueError(
f"{colour_str} is not a valid colour. Please use HTML hex notation."
)
r = int("0x" + colour_match.group(1), 16)
g = int("0x" + colour_match.group(2), 16)
b = int("0x" + colour_match.group(3), 16)
return r, g, b
class SimulatedCamera(BaseCamera):
"""A Thing that simulates a camera for testing."""
@ -55,62 +118,79 @@ class SimulatedCamera(BaseCamera):
self,
thing_server_interface: lt.ThingServerInterface,
shape: tuple[int, int, int] = (616, 820, 3),
glyph_shape: tuple[int, int, int] = (121, 121, 3),
canvas_shape: tuple[int, int, int] = (3000, 4000, 3),
sample_limits: Optional[tuple[int, int]] = (1000, 1500),
canvas_shape: tuple[int, int, int] = (1500, 2000, 3),
frame_interval: float = 0.1,
) -> None:
"""Initialise the simulated with settings for how images are generated.
:param shape: The shape (size) of the generated image.
:param glyph_shape: The size randomly positioned glyphs.
:param canvas_shape: The shape (size) of the canvas generated on initialisation
that images are cropped from. If this is too large the it uses resources,
but its size limits the range of motion of the simulation.
:param sample_limits: The shape of the sample. Outside this range, the
camera won't generate any blobs, preventing scanning from running
indefinitely and better demonstrating background detect.
:param frame_interval: Nominally the time between frames on the MJPEG stream,
however the rate may be slower due to calculation time for focus.
"""
super().__init__(thing_server_interface)
self.shape = shape
self.glyph_shape = glyph_shape
self.canvas_shape = canvas_shape
self.sample_limits = (
canvas_shape[:2] if sample_limits is None else sample_limits
)
self.ds_shape = _downsample_shape(shape)
self.glyph_size = 105 // DOWNSAMPLE
self.canvas_shape = _downsample_shape(canvas_shape)
self.frame_interval = frame_interval
self._capture_thread: Optional[Thread] = None
self._capture_enabled = False
self.validate_inputs()
self.generate_sprites()
self.generate_blobs()
self.generate_canvas()
repeating: bool = lt.property(default=False)
_blob_density: int = 400
@lt.property
def blob_density(self) -> int:
"""The number of blobs per million pixels."""
return self._blob_density
@blob_density.setter
def _set_blob_density(self, value: int) -> None:
self._blob_density = value
if self._capture_enabled:
self.generate_canvas()
_colour: str = "#b937b9"
@lt.property
def colour(self) -> str:
"""The colour of the blobs as a HTML hex string.
The string can either be a single colour (e.g. "#c5247f") or a list of
colours separated by semicolons (e.g. "#c5247f; #b937b9"). Additional
spaces are allowed between colours.
"""
return self._colour
@colour.setter
def _set_colour(self, colour_value: str) -> None:
if COLOUR_LIST_REGEX.match(colour_value) is None:
self.logger.warning(f"{colour_value} is not a valid colour string.")
return
self._colour = colour_value
if self._capture_enabled:
self.generate_canvas()
@lt.property
def calibration_required(self) -> bool:
"""Whether the camera needs calibrating."""
return not self.background_detector_status.ready
def validate_inputs(self) -> None:
"""Validate the inputs passed to the simulation, and raises an error if invalid.
Currently only tests that the sample size is not greater than the canvas size in any dimension.
"""
# Iterate through elements in both tuples. As strict is False, will use the shorter of the two tuples
for a, b in zip(self.canvas_shape, self.sample_limits, strict=False):
if a < b:
raise ValueError(
"Canvas size must be bigger than or equal to canvas size"
)
def generate_sprites(self) -> None:
"""Generate sprites to populate the image."""
sprite_sizes = [10, 21, 36, 40, 50]
sprite_sizes = [s * SPRITE_UPSAMPLE for s in sprite_sizes]
self.sprites = []
channel_block = np.zeros(self.glyph_shape[0:2])
block_size = self.glyph_size * DOWNSAMPLE * SPRITE_UPSAMPLE
channel_block = np.zeros((block_size, block_size))
x = np.arange(channel_block.shape[0])
y = np.arange(channel_block.shape[1])
# 2D grid of radii
@ -121,22 +201,29 @@ class SimulatedCamera(BaseCamera):
for sprite_size in sprite_sizes:
# Mask of where this sprite is
sprite_mask = r_coord < sprite_size
# Calculate a sharp edged circle with value varying from 0 in centre to 1
# Calculate a sharp edged circle with value varying from 0 in centre to 255
# at the edge
sprite_px = r_coord[sprite_mask]
sprite_px -= np.min(sprite_px)
sprite_px /= np.max(sprite_px)
# Create each channel. Note these will be subtracted from the white value.
sprite_r = channel_block.copy()
sprite_r[sprite_mask] = 70 * sprite_px
sprite_g = channel_block.copy()
sprite_g[sprite_mask] = 200 * sprite_px
sprite_b = channel_block.copy()
sprite_b[sprite_mask] = 70 * sprite_px
# Stack into a negative image of the sprite
sprite = np.stack([sprite_r, sprite_g, sprite_b], axis=2)
# Convert to uint8 and append to the list
self.sprites.append(sprite.astype(np.uint8))
sprite = channel_block.copy()
sprite[sprite_mask] = 255 * sprite_px
# Convert to uint8
sprite = sprite.astype(np.uint8)
# Convert to PIL (and back) to resize then append to list of sprites
sprite_pil = Image.fromarray(sprite)
sprite_pil = sprite_pil.resize(
(self.glyph_size, self.glyph_size), Image.Resampling.BILINEAR
)
# Convert back and ensure all edges are zero as these are repeated at sample
# edge
sprite = np.array(sprite_pil)
sprite[0, :] = 0
sprite[-1, :] = 0
sprite[:, 0] = 0
sprite[:, -1] = 0
self.sprites.append(sprite)
def generate_blobs(self, n_blobs: int = 1000) -> None:
"""Generate coordinates of blobs and their sizes, centered around (0,0).
@ -148,10 +235,10 @@ class SimulatedCamera(BaseCamera):
:param n_blobs: The number of blobs to generate.
"""
self.blobs = np.zeros((n_blobs, 3))
w = np.max(self.glyph_shape)
w = self.glyph_size
self.blobs[:, 0] = RNG.uniform(w // 2, self.sample_limits[1] - w // 2, n_blobs)
self.blobs[:, 1] = RNG.uniform(w // 2, self.sample_limits[0] - w // 2, n_blobs)
self.blobs[:, 0] = RNG.uniform(w // 2, self.canvas_shape[1] - w // 2, n_blobs)
self.blobs[:, 1] = RNG.uniform(w // 2, self.canvas_shape[0] - w // 2, n_blobs)
self.blobs[:, 2] = RNG.choice(len(self.sprites), n_blobs)
def generate_canvas(self) -> None:
@ -160,22 +247,22 @@ class SimulatedCamera(BaseCamera):
Canvas is int16 so that random noise can be added to simulation image before
changing to unit8 to stop wrapping.
"""
n_pixels = self.canvas_shape[0] * self.canvas_shape[1] * DOWNSAMPLE**2
self.generate_blobs(int(self.blob_density * 1e-6 * n_pixels))
self.blank_canvas = np.ones(self.canvas_shape, dtype=np.int16)
self.blank_canvas[:, :, 0] *= BG_COLOR[0]
self.blank_canvas[:, :, 1] *= BG_COLOR[1]
self.blank_canvas[:, :, 2] *= BG_COLOR[2]
self.canvas = self.blank_canvas.copy()
new_canvas = self.blank_canvas.copy()
for blob_x, blob_y, sprite_index in self.blobs:
self.draw_sprite_on_canvas(
self.sprites[int(sprite_index)], int(blob_y), int(blob_x)
new_canvas, self.sprites[int(sprite_index)], int(blob_y), int(blob_x)
)
self.canvas[self.canvas < 0] = 0
self.canvas[self.canvas > 255] = 255
self.canvas = np.clip(new_canvas, 0, 255)
def draw_sprite_on_canvas(
self, sprite: np.ndarray, centre_y: int, centre_x: int
self, canvas: np.ndarray, sprite: np.ndarray, centre_y: int, centre_x: int
) -> None:
"""Place one sprite on canvas at given centre coordinates.
@ -185,8 +272,15 @@ class SimulatedCamera(BaseCamera):
:param centre_y: The y coordinate to place the centre of the sprite.
:param centre_x: The x coordinate to place the centre of the sprite.
"""
canvas_h, canvas_w, _ = self.canvas.shape
sprite_h, sprite_w, _ = sprite.shape
canvas_h, canvas_w, _ = canvas.shape
sprite_h, sprite_w = sprite.shape
sprite_f = sprite.astype(float) / 255
r, g, b = colour_str_to_colour(self.colour)
sprite_r = (255 - r) * sprite_f
sprite_g = (255 - g) * sprite_f
sprite_b = (255 - b) * sprite_f
sprite_rgb = np.stack([sprite_r, sprite_g, sprite_b], axis=2)
# Canvas region containing the sprite
top = max(centre_y - sprite_h // 2, 0)
@ -194,7 +288,7 @@ class SimulatedCamera(BaseCamera):
bottom = min(centre_y + (sprite_h - sprite_h // 2), canvas_h)
right = min(centre_x + (sprite_w - sprite_w // 2), canvas_w)
self.canvas[top:bottom, left:right] -= sprite
canvas[top:bottom, left:right] -= sprite_rgb.astype("int16")
def generate_image(self, pos: tuple[int, int, int]) -> Image.Image:
"""Generate an image with blobs based on supplied coordinates.
@ -202,46 +296,55 @@ class SimulatedCamera(BaseCamera):
:param pos: a 3-item tuple containing the x,y,z coordinates of the 'stage'
"""
canvas_width, canvas_height, _ = self.canvas_shape
image_width, image_height, _ = self.shape
# Scale position by RATIO to get position in base image.
im_pos = tuple(x * ratio for x, ratio in zip(pos, RATIO, strict=True))
top_left = (
int(im_pos[0]) - image_width // 2 + self.sample_limits[0] // 2,
int(im_pos[1]) - image_height // 2 + self.sample_limits[1] // 2,
image_width, image_height, _ = self.ds_shape
im_pos = (
pos[0] * RATIO[0] / DOWNSAMPLE,
pos[1] * RATIO[1] / DOWNSAMPLE,
pos[2] * RATIO[2],
)
# Create index list with modulo rather than slicing to handle wrapping at the
# canvas edge.
x_indices = (np.arange(top_left[0], top_left[0] + image_width)) % canvas_width
y_indices = (np.arange(top_left[1], top_left[1] + image_height)) % canvas_height
z_indices = np.arange(self.shape[2])
top_left = (
int(im_pos[0]) - image_width // 2 + self.canvas_shape[0] // 2,
int(im_pos[1]) - image_height // 2 + self.canvas_shape[1] // 2,
)
x_indices = np.arange(top_left[0], top_left[0] + image_width)
y_indices = np.arange(top_left[1], top_left[1] + image_height)
if self.repeating:
# Create index list with modulo rather than slicing to handle wrapping at the
# canvas edge.
x_indices = x_indices % canvas_width
y_indices = y_indices % canvas_height
else:
# Rather than use a modulo for the index list, as above when wrapping,
# this uses np.clip to coerce all out of bound indices to repeat the
# first or last pixel in the canvas. This works because no sprite touches
# the very edge of the canvas (to prevent partial sprites).
x_indices = np.clip(x_indices, 0, canvas_width - 1)
y_indices = np.clip(y_indices, 0, canvas_height - 1)
z_indices = np.arange(self.ds_shape[2])
canvas = self.canvas if self._show_sample else self.blank_canvas
# Use npx to make each 1d index list 3D
focused_image = canvas[np.ix_(x_indices, y_indices, z_indices)]
focused_np_img = canvas[np.ix_(x_indices, y_indices, z_indices)]
image = fast_pil_blur(focused_image, sigma=np.abs(im_pos[2]) / 5)
if image.shape != self.shape:
raise ValueError(
f"Image shape {image.shape} does not match intended shape {self.shape}"
)
np_img = fast_pil_blur(focused_np_img, sigma=np.abs(im_pos[2]) / 5)
# Add noise and convert to uint8
image += RNG.normal(scale=self.noise_level, size=self.shape).astype("int16")
image[image < 0] = 0
image[image > 255] = 255
return Image.fromarray(image.astype("uint8"))
np_img += RNG.normal(scale=self.noise_level, size=self.ds_shape).astype("int16")
np.clip(np_img, 0, 255, out=np_img)
pl_img = Image.fromarray(np_img.astype("uint8"))
return pl_img.resize((self.shape[1], self.shape[0]), Image.Resampling.BILINEAR)
def generate_frame(self) -> Image.Image:
"""Generate a frame with blobs based on the stage coordinates."""
try:
pos = self._stage.instantaneous_position
except Exception as e:
LOGGER.debug(f"Failed to get stage position: {e}")
pos = {"x": 0, "y": 0, "z": 0}
pos = self._stage.instantaneous_position
return self.generate_image((pos["y"], pos["x"], pos["z"]))
def __enter__(self) -> Self:
"""Start the capture thread when the Thing context manager is opened."""
self.generate_canvas()
self.start_streaming()
return self
@ -298,14 +401,11 @@ class SimulatedCamera(BaseCamera):
if wait_time > 0:
time.sleep(wait_time)
last_frame_t = time.time()
try:
frame = self.generate_frame()
self.mjpeg_stream.add_frame(_frame2bytes(frame))
ds_frame = frame.resize((320, 240), resample=Image.Resampling.NEAREST)
self.lores_mjpeg_stream.add_frame(_frame2bytes(ds_frame))
except Exception as e:
LOGGER.exception(f"Failed to capture frame: {e}, retrying...")
frame = self.generate_frame()
self.mjpeg_stream.add_frame(_frame2bytes(frame))
ds_frame = frame.resize((320, 240), resample=Image.Resampling.NEAREST)
self.lores_mjpeg_stream.add_frame(_frame2bytes(ds_frame))
@lt.action
def discard_frames(self) -> None:
@ -401,7 +501,12 @@ class SimulatedCamera(BaseCamera):
@lt.property
def manual_camera_settings(self) -> list[PropertyControl]:
"""The camera settings to expose as property controls in the settings panel."""
return [property_control_for(self, "noise_level", label="Noise Level")]
return [
property_control_for(self, "repeating", label="Infinite Sample"),
property_control_for(self, "blob_density", label="Sample Density"),
property_control_for(self, "colour", label="Sample Colour"),
property_control_for(self, "noise_level", label="Noise Level"),
]
def _frame2bytes(frame: Image.Image) -> bytes: