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

Formatted with Ruff

Browse source
This commit is contained in:
Richard Bowman 2024-12-03 06:46:08 +00:00
parent 27aec769a2
commit 9fd8fc37f1
19 changed files with 680 additions and 392 deletions
Download patch file Download diff file Expand all files Collapse all files

37
src/openflexure_microscope_server/things/camera/__init__.py
View file

@ -5,6 +5,7 @@ should enabe the server to work.
See repository root for licensing information.
"""
from __future__ import annotations
import logging
from typing import Literal, Protocol, runtime_checkable
@ -23,15 +24,14 @@ from labthings_fastapi.types.numpy import NDArray
class JPEGBlob(Blob):
media_type: str = "image/jpeg"
@runtime_checkable
class CameraProtocol(Protocol):
"""A Thing representing a camera"""
def __enter__(self) -> None:
...
def __exit__(self, _exc_type, _exc_value, _traceback) -> None:
...
def __enter__(self) -> None: ...
def __exit__(self, _exc_type, _exc_value, _traceback) -> None: ...
@property
def stream_active(self) -> bool:
@ -45,9 +45,8 @@ class CameraProtocol(Protocol):
def capture_array(
self,
resolution: Literal["lores", "main", "full"] = "main",
) -> NDArray:
...
) -> NDArray: ...
def capture_jpeg(
self,
metadata_getter: GetThingStates,
@ -78,9 +77,10 @@ class CameraProtocol(Protocol):
"""Acquire one image from the preview stream and return its size"""
...
class BaseCamera(Thing):
"""A Thing representing a camera
This is a concrete base class for `Thing`s implementing the `CameraProtocol`.
It provides the stream descriptors and actions to grab from the stream.
"""
@ -110,12 +110,16 @@ class BaseCamera(Thing):
stream (either "main" for the preview stream, or "lores" for the low
resolution preview). No metadata is returned.
"""
logging.info(f"StreamingPiCamera2.grab_jpeg(stream_name={stream_name}) starting")
logging.info(
f"StreamingPiCamera2.grab_jpeg(stream_name={stream_name}) starting"
)
stream = (
self.lores_mjpeg_stream if stream_name == "lores" else self.mjpeg_stream
)
frame = portal.call(stream.grab_frame)
logging.info(f"StreamingPiCamera2.grab_jpeg(stream_name={stream_name}) got frame")
logging.info(
f"StreamingPiCamera2.grab_jpeg(stream_name={stream_name}) got frame"
)
return JPEGBlob.from_bytes(frame)
@thing_action
@ -129,12 +133,12 @@ class BaseCamera(Thing):
self.lores_mjpeg_stream if stream_name == "lores" else self.mjpeg_stream
)
return portal.call(stream.next_frame_size)
class CameraStub(BaseCamera):
"""A stub for a camera, to allow dependencies
As of LabThings-FastAPI 0.0.7, we can't make a thing client dependency
based on a protocol, because the protocol is not a Thing, and its
methods/properties are not decorated as Affordances. This stub class
@ -142,9 +146,10 @@ class CameraStub(BaseCamera):
This stub class should be used for dependencies on the CameraProtocol.
"""
def __enter__(self) -> None:
raise NotImplementedError("Cameras must not inherit from CameraStub")
def __exit__(self, _exc_type, _exc_value, _traceback) -> None:
raise NotImplementedError("Cameras must not inherit from CameraStub")
@ -164,7 +169,7 @@ class CameraStub(BaseCamera):
resolution: Literal["lores", "main", "full"] = "main",
) -> NDArray:
raise NotImplementedError("Cameras must not inherit from CameraStub")
@thing_action
def capture_jpeg(
self,

27
src/openflexure_microscope_server/things/camera/opencv.py
View file

@ -1,10 +1,11 @@
"""OpenFlexure Microscope OpenCV Camera
This module defines a camera Thing that uses OpenCV's
This module defines a camera Thing that uses OpenCV's
`VideoCapture`.
See repository root for licensing information.
"""
from __future__ import annotations
import io
import json
@ -26,7 +27,8 @@ from . import BaseCamera, JPEGBlob
class OpenCVCamera(BaseCamera):
"""A Thing representing an OpenCV camera"""
def __init__(self, camera_index: int=0):
def __init__(self, camera_index: int = 0):
self.camera_index = camera_index
self._capture_thread: Optional[Thread] = None
self._capture_enabled = False
@ -37,7 +39,7 @@ class OpenCVCamera(BaseCamera):
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
@ -50,6 +52,7 @@ class OpenCVCamera(BaseCamera):
if self._capture_enabled and self._capture_thread:
return self._capture_thread.is_alive()
return False
mjpeg_stream = MJPEGStreamDescriptor()
lores_mjpeg_stream = MJPEGStreamDescriptor()
@ -58,11 +61,15 @@ class OpenCVCamera(BaseCamera):
while self._capture_enabled:
ret, frame = self.cap.read()
if not ret:
logging.error(f"Failed to capture frame from camera {self.camera_index}")
logging.error(
f"Failed to capture frame from camera {self.camera_index}"
)
break
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()
jpeg_lores = cv2.imencode(".jpg", cv2.resize(frame, (320, 240)))[
1
].tobytes()
self.lores_mjpeg_stream.add_frame(jpeg_lores, portal)
@thing_action
@ -78,9 +85,11 @@ class OpenCVCamera(BaseCamera):
"""
ret, frame = self.cap.read()
if not ret:
raise RuntimeError(f"Failed to capture frame from camera {self.camera_index}")
raise RuntimeError(
f"Failed to capture frame from camera {self.camera_index}"
)
return frame
@thing_action
def capture_jpeg(
self,
@ -95,7 +104,9 @@ class OpenCVCamera(BaseCamera):
jpeg = cv2.imencode(".jpg", frame)[1].tobytes()
exif_dict = {
"Exif": {
piexif.ExifIFD.UserComment: json.dumps(metadata_getter()).encode("utf-8")
piexif.ExifIFD.UserComment: json.dumps(metadata_getter()).encode(
"utf-8"
)
},
"GPS": {},
"Interop": {},

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

@ -5,6 +5,7 @@ camera together to perform an autofocus routine.
See repository root for licensing information.
"""
from __future__ import annotations
import io
import json
@ -28,22 +29,26 @@ from pydantic import RootModel
from . import BaseCamera, JPEGBlob
from ..stage import StageProtocol as Stage
class ArrayModel(RootModel):
"""A model for an array"""
root: NDArray
class SimulatedCamera(BaseCamera):
"""A Thing representing an OpenCV camera"""
_stage: Optional[Stage] = None
_server: Optional[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: 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
@ -53,14 +58,14 @@ class SimulatedCamera(BaseCamera):
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)
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
@ -68,15 +73,15 @@ class SimulatedCamera(BaseCamera):
def generate_blobs(self, N: 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, 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)
self.blobs[:, 1] = rng.uniform(w/2, self.canvas_shape[1]-w/2, N)
self.blobs[:, 0] = rng.uniform(w / 2, self.canvas_shape[0] - w / 2, N)
self.blobs[:, 1] = rng.uniform(w / 2, self.canvas_shape[1] - w / 2, N)
self.blobs[:, 2] = rng.choice(len(self.sprites), N)
def generate_canvas(self):
@ -86,31 +91,34 @@ class SimulatedCamera(BaseCamera):
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,
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]):
"""Generate an image with blobs based on supplied coordinates"""
cw, ch, _ = self.canvas_shape
w, h, _ = self.shape
tl = (int(pos[0]) - w//2 - cw//2, int(pos[1]) - h//2 - ch//2)
tl = (int(pos[0]) - w // 2 - cw // 2, int(pos[1]) - h // 2 - ch // 2)
image = self.canvas[
tuple(slice(tl[i],tl[i] + self.shape[i]) for i in range(2)) + (slice(None),)
tuple(slice(tl[i], tl[i] + self.shape[i]) for i in range(2))
+ (slice(None),)
]
if image.shape != self.shape:
raise ValueError(f"Image shape {image.shape} does not match intended 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: ThingServer, path: str):
self._server = server
return super().attach_to_server(server, 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:
@ -118,14 +126,14 @@ class SimulatedCamera(BaseCamera):
except Exception as e:
print(f"Failed to get stage position: {e}")
pos = {"x": 0, "y": 0}
return self.generate_image((pos["x"]/10, pos["y"]/10))
return self.generate_image((pos["x"] / 10, pos["y"] / 10))
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
@ -137,6 +145,7 @@ class SimulatedCamera(BaseCamera):
if self._capture_enabled and self._capture_thread:
return self._capture_thread.is_alive()
return False
mjpeg_stream = MJPEGStreamDescriptor()
lores_mjpeg_stream = MJPEGStreamDescriptor()
@ -148,7 +157,9 @@ class SimulatedCamera(BaseCamera):
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()
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...")
@ -165,7 +176,7 @@ class SimulatedCamera(BaseCamera):
binary image formats will be added in due course.
"""
return self.generate_frame()
@thing_action
def capture_jpeg(
self,
@ -180,7 +191,9 @@ class SimulatedCamera(BaseCamera):
jpeg = cv2.imencode(".jpg", frame)[1].tobytes()
exif_dict = {
"Exif": {
piexif.ExifIFD.UserComment: json.dumps(metadata_getter()).encode("utf-8")
piexif.ExifIFD.UserComment: json.dumps(metadata_getter()).encode(
"utf-8"
)
},
"GPS": {},
"Interop": {},