Merge branch 'v3-camera-protocol' into 'v3'
Formal Protocol for camera and stage interfaces See merge request openflexure/openflexure-microscope-server!192
This commit is contained in:
Richard Bowman
commit
4d4ca831b3
20 changed files with 215 additions and 1794 deletions
|
|
@ -4,13 +4,10 @@ import logging
|
|||
from labthings_fastapi.thing import Thing
|
||||
from labthings_fastapi.dependencies.thing import direct_thing_client_dependency
|
||||
from labthings_fastapi.decorators import thing_action
|
||||
from .stage import Stage
|
||||
from .camera import Camera
|
||||
from .stage import StageDependency as StageDep
|
||||
from openflexure_microscope_server.things.autofocus import AutofocusThing
|
||||
from openflexure_microscope_server.things.camera_stage_mapping import CameraStageMapper
|
||||
|
||||
StageDep = direct_thing_client_dependency(Stage, "/stage/")
|
||||
CamDep = direct_thing_client_dependency(Camera, "/camera/")
|
||||
CSMDep = direct_thing_client_dependency(CameraStageMapper, "/camera_stage_mapping/")
|
||||
AutofocusDep = direct_thing_client_dependency(AutofocusThing, "/autofocus/")
|
||||
|
||||
|
|
|
|||
|
|
@ -19,14 +19,12 @@ from labthings_fastapi.dependencies.thing import direct_thing_client_dependency
|
|||
from labthings_fastapi.dependencies.blocking_portal import BlockingPortal
|
||||
from labthings_fastapi.decorators import thing_action
|
||||
from labthings_fastapi.types.numpy import NDArray
|
||||
from .camera import Camera as CameraThing
|
||||
from .stage import Stage as StageThing
|
||||
from .camera import RawCameraDependency as Camera
|
||||
from .camera import CameraDependency as WrappedCamera
|
||||
from .stage import StageDependency as Stage
|
||||
import numpy as np
|
||||
from pydantic import BaseModel
|
||||
|
||||
Stage = direct_thing_client_dependency(StageThing, "/stage/")
|
||||
Camera = raw_thing_dependency(CameraThing)
|
||||
WrappedCamera = direct_thing_client_dependency(CameraThing, "/camera/")
|
||||
|
||||
### Autofocus utilities
|
||||
|
||||
|
|
|
|||
|
|
@ -7,54 +7,84 @@ See repository root for licensing information.
|
|||
"""
|
||||
from __future__ import annotations
|
||||
import logging
|
||||
from typing import Literal
|
||||
from typing import Literal, Protocol, runtime_checkable
|
||||
|
||||
from labthings_fastapi.thing import Thing
|
||||
from labthings_fastapi.decorators import thing_action, thing_property
|
||||
from labthings_fastapi.dependencies.metadata import GetThingStates
|
||||
from labthings_fastapi.dependencies.blocking_portal import BlockingPortal
|
||||
from labthings_fastapi.dependencies.thing import direct_thing_client_dependency
|
||||
from labthings_fastapi.dependencies.raw_thing import raw_thing_dependency
|
||||
from labthings_fastapi.outputs.mjpeg_stream import MJPEGStreamDescriptor
|
||||
from labthings_fastapi.outputs.blob import blob_type
|
||||
from labthings_fastapi.outputs.blob import Blob
|
||||
from labthings_fastapi.types.numpy import NDArray
|
||||
|
||||
|
||||
JPEGBlob = blob_type("image/jpeg")
|
||||
class JPEGBlob(Blob):
|
||||
media_type: str = "image/jpeg"
|
||||
|
||||
class CameraMeta:
|
||||
def __subclasscheck__(cls, C):
|
||||
"""Override Python's default behaviour and use duck typing"""
|
||||
print(f"Checking if {C} is a camera...")
|
||||
for action in ["snap_image", "capture_array", "capture_jpeg", "grab_jpeg", "grab_jpeg_size"]:
|
||||
a = getattr(C, action, None)
|
||||
if not callable(a):
|
||||
return False
|
||||
for prop in ["stream_active"]:
|
||||
if not hasattr(C, prop):
|
||||
return False
|
||||
for k in ["mjpeg_stream", "lores_mjpeg_stream"]:
|
||||
stream = getattr(C, k)
|
||||
if not isinstance(stream, MJPEGStreamDescriptor):
|
||||
return False
|
||||
return True
|
||||
|
||||
def __instancecheck__(cls, I):
|
||||
return cls.__subclasscheck__(I)
|
||||
|
||||
|
||||
class Camera(Thing):
|
||||
__metaclass__ = CameraMeta
|
||||
@runtime_checkable
|
||||
class CameraProtocol(Protocol):
|
||||
"""A Thing representing a camera"""
|
||||
|
||||
def __enter__(self):
|
||||
raise NotImplementedError("Subclasses must implement __enter__")
|
||||
def __enter__(self) -> None:
|
||||
...
|
||||
|
||||
def __exit__(self, _exc_type, _exc_value, _traceback):
|
||||
raise NotImplementedError("Subclasses must implement __exit__")
|
||||
def __exit__(self, _exc_type, _exc_value, _traceback) -> None:
|
||||
...
|
||||
|
||||
@thing_property
|
||||
@property
|
||||
def stream_active(self) -> bool:
|
||||
"Whether the MJPEG stream is active"
|
||||
raise NotImplementedError("Subclasses must implement stream_active")
|
||||
...
|
||||
|
||||
def snap_image(self) -> NDArray:
|
||||
"""Acquire one image from the camera."""
|
||||
...
|
||||
|
||||
def capture_array(
|
||||
self,
|
||||
resolution: Literal["lores", "main", "full"] = "main",
|
||||
) -> NDArray:
|
||||
...
|
||||
|
||||
def capture_jpeg(
|
||||
self,
|
||||
metadata_getter: GetThingStates,
|
||||
resolution: Literal["lores", "main", "full"] = "main",
|
||||
) -> JPEGBlob:
|
||||
"""Acquire one image from the camera and return as a JPEG blob"""
|
||||
...
|
||||
|
||||
def grab_jpeg(
|
||||
self,
|
||||
portal: BlockingPortal,
|
||||
stream_name: Literal["main", "lores"] = "main",
|
||||
) -> JPEGBlob:
|
||||
"""Acquire one image from the preview stream and return as an array
|
||||
|
||||
This differs from `capture_jpeg` in that it does not pause the MJPEG
|
||||
preview stream. Instead, we simply return the next frame from that
|
||||
stream (either "main" for the preview stream, or "lores" for the low
|
||||
resolution preview). No metadata is returned.
|
||||
"""
|
||||
...
|
||||
|
||||
def grab_jpeg_size(
|
||||
self,
|
||||
portal: BlockingPortal,
|
||||
stream_name: Literal["main", "lores"] = "main",
|
||||
) -> int:
|
||||
"""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.
|
||||
"""
|
||||
|
||||
mjpeg_stream = MJPEGStreamDescriptor()
|
||||
lores_mjpeg_stream = MJPEGStreamDescriptor()
|
||||
|
||||
|
|
@ -67,31 +97,6 @@ class Camera(Thing):
|
|||
"""
|
||||
return self.capture_array()
|
||||
|
||||
@thing_action
|
||||
def capture_array(
|
||||
self,
|
||||
resolution: Literal["lores", "main", "full"] = "main",
|
||||
) -> NDArray:
|
||||
"""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.
|
||||
"""
|
||||
raise NotImplementedError("Subclasses must implement capture_array")
|
||||
|
||||
@thing_action
|
||||
def capture_jpeg(
|
||||
self,
|
||||
metadata_getter: GetThingStates,
|
||||
resolution: Literal["lores", "main", "full"] = "main",
|
||||
) -> JPEGBlob:
|
||||
"""Acquire one image from the camera and return as a JPEG blob
|
||||
|
||||
This function will produce a JPEG image.
|
||||
"""
|
||||
raise NotImplementedError("Subclasses must implement capture_jpeg")
|
||||
|
||||
@thing_action
|
||||
def grab_jpeg(
|
||||
self,
|
||||
|
|
@ -124,4 +129,51 @@ class Camera(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
|
||||
is a workaround for that limitation, and should not be used directly.
|
||||
|
||||
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")
|
||||
|
||||
@thing_property
|
||||
def stream_active(self) -> bool:
|
||||
"Whether the MJPEG stream is active"
|
||||
raise NotImplementedError("Cameras must not inherit from CameraStub")
|
||||
|
||||
@thing_action
|
||||
def snap_image(self) -> NDArray:
|
||||
"""Acquire one image from the camera."""
|
||||
raise NotImplementedError("Cameras must not inherit from CameraStub")
|
||||
|
||||
@thing_action
|
||||
def capture_array(
|
||||
self,
|
||||
resolution: Literal["lores", "main", "full"] = "main",
|
||||
) -> NDArray:
|
||||
raise NotImplementedError("Cameras must not inherit from CameraStub")
|
||||
|
||||
@thing_action
|
||||
def capture_jpeg(
|
||||
self,
|
||||
metadata_getter: GetThingStates,
|
||||
resolution: Literal["lores", "main", "full"] = "main",
|
||||
) -> JPEGBlob:
|
||||
"""Acquire one image from the camera and return as a JPEG blob"""
|
||||
raise NotImplementedError("Cameras must not inherit from CameraStub")
|
||||
|
||||
|
||||
CameraDependency = direct_thing_client_dependency(CameraStub, "/camera/")
|
||||
RawCameraDependency = raw_thing_dependency(CameraProtocol)
|
||||
|
|
|
|||
|
|
@ -21,10 +21,10 @@ from labthings_fastapi.dependencies.metadata import GetThingStates
|
|||
from labthings_fastapi.outputs.mjpeg_stream import MJPEGStreamDescriptor
|
||||
from labthings_fastapi.types.numpy import NDArray
|
||||
|
||||
from . import Camera, JPEGBlob
|
||||
from . import BaseCamera, JPEGBlob
|
||||
|
||||
|
||||
class OpenCVCamera(Camera):
|
||||
class OpenCVCamera(BaseCamera):
|
||||
"""A Thing representing an OpenCV camera"""
|
||||
def __init__(self, camera_index: int=0):
|
||||
self.camera_index = camera_index
|
||||
|
|
|
|||
|
|
@ -24,11 +24,11 @@ from labthings_fastapi.outputs.mjpeg_stream import MJPEGStreamDescriptor
|
|||
from labthings_fastapi.types.numpy import NDArray
|
||||
from labthings_fastapi.server import ThingServer
|
||||
|
||||
from . import Camera, JPEGBlob
|
||||
from ..stage import Stage
|
||||
from . import BaseCamera, JPEGBlob
|
||||
from ..stage import StageProtocol as Stage
|
||||
|
||||
|
||||
class SimulatedCamera(Camera):
|
||||
class SimulatedCamera(BaseCamera):
|
||||
"""A Thing representing an OpenCV camera"""
|
||||
shape = (600, 800, 3)
|
||||
glyph_shape = (51, 51, 3)
|
||||
|
|
|
|||
|
|
@ -21,18 +21,13 @@ from camera_stage_mapping.camera_stage_calibration_1d import (
|
|||
calibrate_backlash_1d,
|
||||
image_to_stage_displacement_from_1d,
|
||||
)
|
||||
from camera_stage_mapping.camera_stage_tracker import Tracker
|
||||
from .camera import Camera as CameraThing
|
||||
from .stage import Stage as StageThing
|
||||
|
||||
from labthings_fastapi.dependencies.thing import direct_thing_client_dependency
|
||||
from labthings_fastapi.dependencies.invocation import InvocationCancelledError, InvocationLogger
|
||||
from labthings_fastapi.types.numpy import NDArray, denumpify, DenumpifyingDict
|
||||
from labthings_fastapi.decorators import thing_action, thing_property
|
||||
from labthings_fastapi.thing import Thing
|
||||
|
||||
Camera = direct_thing_client_dependency(CameraThing, "/camera/")
|
||||
Stage = direct_thing_client_dependency(StageThing, "/stage/")
|
||||
from camera_stage_mapping.camera_stage_tracker import Tracker
|
||||
from .camera import CameraDependency as Camera
|
||||
from .stage import StageDependency as Stage
|
||||
|
||||
CoordinateType = Tuple[float, float, float]
|
||||
XYCoordinateType = Tuple[float, float]
|
||||
|
|
|
|||
|
|
@ -28,15 +28,13 @@ from labthings_fastapi.dependencies.thing import direct_thing_client_dependency
|
|||
from labthings_fastapi.dependencies.invocation import CancelHook, InvocationLogger, InvocationCancelledError
|
||||
from labthings_fastapi.decorators import thing_action, thing_property, fastapi_endpoint
|
||||
from labthings_fastapi.outputs.blob import blob_type
|
||||
from .camera import Camera
|
||||
from .stage import Stage
|
||||
from .camera import CameraDependency as CamDep
|
||||
from .stage import StageDependency as StageDep
|
||||
from openflexure_microscope_server.things.autofocus import AutofocusThing
|
||||
from openflexure_microscope_server.things.camera_stage_mapping import CameraStageMapper
|
||||
from openflexure_microscope_server.things.auto_recentre_stage import RecentringThing
|
||||
|
||||
|
||||
CamDep = direct_thing_client_dependency(Camera, "/camera/")
|
||||
StageDep = direct_thing_client_dependency(Stage, "/stage/")
|
||||
CSMDep = direct_thing_client_dependency(CameraStageMapper, "/camera_stage_mapping/")
|
||||
AutofocusDep = direct_thing_client_dependency(AutofocusThing, "/autofocus/")
|
||||
RecentreStage = direct_thing_client_dependency(RecentringThing, "/auto_recentre_stage/")
|
||||
|
|
|
|||
|
|
@ -1,16 +1,63 @@
|
|||
from __future__ import annotations
|
||||
from typing import Any, Protocol, TypeAlias, runtime_checkable
|
||||
from labthings_fastapi.descriptors.property import PropertyDescriptor
|
||||
from labthings_fastapi.thing import Thing
|
||||
from labthings_fastapi.decorators import thing_action, thing_property
|
||||
from labthings_fastapi.dependencies.invocation import CancelHook
|
||||
from labthings_fastapi.dependencies.thing import direct_thing_client_dependency
|
||||
from collections.abc import Sequence, Mapping
|
||||
|
||||
@runtime_checkable
|
||||
class StageProtocol(Protocol):
|
||||
"""A protocol for the OpenFlexure translation stage
|
||||
"""
|
||||
_axis_names: Sequence[str]
|
||||
|
||||
class Stage(Thing):
|
||||
"""A dummy stage for testing purposes
|
||||
@property
|
||||
def axis_names(self) -> Sequence[str]:
|
||||
"""The names of the stage's axes, in order."""
|
||||
...
|
||||
|
||||
@property
|
||||
def position(self) -> Mapping[str, int]:
|
||||
"""Current position of the stage"""
|
||||
...
|
||||
|
||||
@property
|
||||
def moving() -> bool:
|
||||
"Whether the stage is in motion"
|
||||
...
|
||||
|
||||
@property
|
||||
def thing_state(self) -> Mapping[str, Any]:
|
||||
"""Summary metadata describing the current state of the stage"""
|
||||
...
|
||||
|
||||
This stage should work similarly to a Sangaboard stage, but without any
|
||||
hardware attached.
|
||||
def move_relative(self, cancel: CancelHook, block_cancellation: bool=False, **kwargs: Mapping[str, int]):
|
||||
"""Make a relative move. Keyword arguments should be axis names."""
|
||||
...
|
||||
|
||||
def move_absolute(self, cancel: CancelHook, block_cancellation: bool=False, **kwargs: Mapping[str, int]):
|
||||
"""Make an absolute move. Keyword arguments should be axis names."""
|
||||
...
|
||||
|
||||
def set_zero_position(self):
|
||||
"""Make the current position zero in all axes
|
||||
|
||||
This action does not move the stage, but resets the position to zero.
|
||||
It is intended for use after manually or automatically recentring the
|
||||
stage.
|
||||
"""
|
||||
...
|
||||
|
||||
|
||||
class BaseStage(Thing):
|
||||
"""A base stage class for OpenFlexure translation stages
|
||||
|
||||
This can't be used directly but should reduce boilerplate code when
|
||||
implementing new stages. A minimal working stage must implement
|
||||
`move_relative` and `move_absolute` actions, which update the
|
||||
`position` property on completion, and provide `set_zero_position`.
|
||||
"""
|
||||
_axis_names = ("x", "y", "z")
|
||||
|
||||
|
|
@ -41,17 +88,26 @@ class Stage(Thing):
|
|||
return {
|
||||
"position": self.position
|
||||
}
|
||||
|
||||
|
||||
|
||||
class StageStub(BaseStage):
|
||||
"""A Stub Thing for the OpenFlexure translation stage
|
||||
|
||||
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
|
||||
is a workaround for that limitation, and should not be used directly.
|
||||
"""
|
||||
@thing_action
|
||||
def move_relative(self, cancel: CancelHook, block_cancellation: bool=False, **kwargs: Mapping[str, int]):
|
||||
"""Make a relative move. Keyword arguments should be axis names."""
|
||||
raise NotImplementedError("Subclasses should implement this method")
|
||||
raise NotImplementedError("StageStub should not be used directly")
|
||||
|
||||
@thing_action
|
||||
def move_absolute(self, cancel: CancelHook, block_cancellation: bool=False, **kwargs: Mapping[str, int]):
|
||||
"""Make an absolute move. Keyword arguments should be axis names."""
|
||||
raise NotImplementedError("Subclasses should implement this method")
|
||||
|
||||
raise NotImplementedError("StageStub should not be used directly")
|
||||
|
||||
@thing_action
|
||||
def set_zero_position(self):
|
||||
"""Make the current position zero in all axes
|
||||
|
|
@ -60,4 +116,7 @@ class Stage(Thing):
|
|||
It is intended for use after manually or automatically recentring the
|
||||
stage.
|
||||
"""
|
||||
raise NotImplementedError("Subclasses should implement this method")
|
||||
raise NotImplementedError("StageStub should not be used directly")
|
||||
|
||||
|
||||
StageDependency: TypeAlias = direct_thing_client_dependency(StageStub, "/stage/")
|
||||
|
|
|
|||
|
|
@ -4,10 +4,10 @@ from labthings_fastapi.dependencies.invocation import CancelHook, InvocationCanc
|
|||
from collections.abc import Mapping
|
||||
import time
|
||||
|
||||
from . import Stage
|
||||
from . import BaseStage
|
||||
|
||||
|
||||
class DummyStage(Stage):
|
||||
class DummyStage(BaseStage):
|
||||
"""A dummy stage for testing purposes
|
||||
|
||||
This stage should work similarly to a Sangaboard stage, but without any
|
||||
|
|
|
|||
Binary file not shown.
|
Before Width: | Height: | Size: 906 B |
|
|
@ -1,19 +0,0 @@
|
|||
import pytest
|
||||
|
||||
from openflexure_microscope.api import utilities
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"test_input,expected",
|
||||
[
|
||||
("true", True),
|
||||
("True", True),
|
||||
("1", True),
|
||||
(True, True),
|
||||
("false", False),
|
||||
("somestring", False),
|
||||
(False, False),
|
||||
],
|
||||
)
|
||||
def test_get_bool(test_input, expected):
|
||||
assert utilities.get_bool(test_input) == expected
|
||||
|
|
@ -1,43 +0,0 @@
|
|||
from apispec.utils import validate_spec
|
||||
import json
|
||||
import jsonschema
|
||||
import os
|
||||
|
||||
from openflexure_microscope.api.app import api_microscope, app, labthing
|
||||
from openflexure_microscope.camera.base import BaseCamera
|
||||
from openflexure_microscope.microscope import Microscope
|
||||
from openflexure_microscope.stage.base import BaseStage
|
||||
from labthings.json import encode_json
|
||||
|
||||
|
||||
def test_app_creation():
|
||||
assert labthing.app is app
|
||||
|
||||
|
||||
def test_microscope_creation():
|
||||
assert isinstance(api_microscope, Microscope)
|
||||
assert isinstance(api_microscope.camera, BaseCamera)
|
||||
assert isinstance(api_microscope.stage, BaseStage)
|
||||
|
||||
|
||||
def test_openapi_valid():
|
||||
assert validate_spec(labthing.spec)
|
||||
|
||||
|
||||
def test_thing_description_valid():
|
||||
# TODO: it would be nice to put this into LabThings
|
||||
# First load the schema from file and check it validates
|
||||
schema_fname = os.path.join(os.path.dirname(__file__), "w3c_td_schema.json")
|
||||
schema = json.load(open(schema_fname, "r"), encoding="utf-8")
|
||||
jsonschema.Draft7Validator.check_schema(schema)
|
||||
|
||||
# Build a TD dictionary
|
||||
with labthing.app.test_request_context():
|
||||
td_dict = labthing.thing_description.to_dict()
|
||||
|
||||
# Allow our LabThingsJSONEncoder to encode the RD
|
||||
td_json = encode_json(td_dict)
|
||||
# Decode the JSON back into a primitive dictionary
|
||||
td_json_dict = json.loads(td_json)
|
||||
# Validate
|
||||
jsonschema.validate(instance=td_json_dict, schema=schema)
|
||||
|
|
@ -1,74 +0,0 @@
|
|||
import os
|
||||
import threading
|
||||
import time
|
||||
|
||||
from openflexure_microscope.camera.base import FrameStream
|
||||
|
||||
|
||||
def test_framestream_write():
|
||||
d = b"openflexure"
|
||||
|
||||
with FrameStream() as f:
|
||||
f.write(d)
|
||||
assert f.getvalue() == d
|
||||
|
||||
|
||||
def test_framestream_overwrite():
|
||||
d1 = b"openflexure"
|
||||
d2 = b"microscope"
|
||||
|
||||
with FrameStream() as f:
|
||||
f.write(d1)
|
||||
assert f.getvalue() == d1
|
||||
f.write(d2)
|
||||
assert f.getvalue() == d2
|
||||
# d1 should be removed, so f.getbuffer() should only contain d2
|
||||
assert f.getbuffer().nbytes == len(d2)
|
||||
|
||||
|
||||
def test_tracker():
|
||||
sizes = range(1, 100)
|
||||
|
||||
with FrameStream() as f:
|
||||
for length in sizes:
|
||||
time.sleep(0.01)
|
||||
d = b"\x00" + os.urandom(length) + b"\x00"
|
||||
f.write(d)
|
||||
assert f.getbuffer().nbytes == len(d)
|
||||
|
||||
# Make sure we have one TrackerFrame for each f.write() call
|
||||
assert len(f.frames) == len(sizes)
|
||||
# For each tested write size
|
||||
for i, frame in enumerate(f.frames):
|
||||
# Make sure the recorded size is what we expect
|
||||
# Should be size +2 for our padding bytes \x00
|
||||
assert frame.size == sizes[i] + 2
|
||||
# Make sure we have a valid time value
|
||||
assert isinstance(frame.time, float)
|
||||
# Make sure the recorded time increases for each frame
|
||||
if i > 0:
|
||||
assert f.frames[i].time > f.frames[i - 1].time
|
||||
|
||||
# Test FrameStream.last
|
||||
assert f.last.size == sizes[-1] + 2
|
||||
|
||||
|
||||
def test_new_frame_event():
|
||||
d = b"openflexure"
|
||||
|
||||
def target(framestream):
|
||||
time.sleep(0.01)
|
||||
framestream.write(d)
|
||||
|
||||
with FrameStream() as f:
|
||||
# Start a thread to write some data
|
||||
thread = threading.Thread(target=target, args=(f,))
|
||||
thread.daemon = True
|
||||
thread.start()
|
||||
# Make sure the stream is empty before we wait for a frame
|
||||
assert not f.getvalue()
|
||||
# Wait for a frame to be written
|
||||
frame = f.getframe()
|
||||
assert frame == d
|
||||
# Make sure getframe() cleared the event
|
||||
assert not f.new_frame.events[threading.get_ident()][0].is_set()
|
||||
|
|
@ -1,150 +0,0 @@
|
|||
import os
|
||||
|
||||
import pytest
|
||||
from freezegun import freeze_time
|
||||
|
||||
from openflexure_microscope.captures.capture_manager import (
|
||||
BASE_CAPTURE_PATH,
|
||||
TEMP_CAPTURE_PATH,
|
||||
CaptureManager,
|
||||
)
|
||||
|
||||
|
||||
def test_new_manager():
|
||||
m = CaptureManager()
|
||||
assert BASE_CAPTURE_PATH
|
||||
assert TEMP_CAPTURE_PATH
|
||||
assert m.paths["default"] == BASE_CAPTURE_PATH
|
||||
assert m.paths["temp"] == TEMP_CAPTURE_PATH
|
||||
|
||||
|
||||
def test_update_settings():
|
||||
m = CaptureManager()
|
||||
assert m.paths["default"] == BASE_CAPTURE_PATH
|
||||
assert m.paths["temp"] == TEMP_CAPTURE_PATH
|
||||
|
||||
new_settings = {
|
||||
"paths": {"default": "BASE_CAPTURE_PATH", "temp": "TEMP_CAPTURE_PATH"}
|
||||
}
|
||||
m.update_settings(new_settings)
|
||||
assert m.paths["default"] == "BASE_CAPTURE_PATH"
|
||||
assert m.paths["temp"] == "TEMP_CAPTURE_PATH"
|
||||
assert m.read_settings() == new_settings
|
||||
|
||||
|
||||
def test__new_output():
|
||||
m = CaptureManager()
|
||||
|
||||
out = m._new_output(True, "filename", "subfolder", "fmt")
|
||||
assert out.file == os.path.join(TEMP_CAPTURE_PATH, "subfolder", "filename.fmt")
|
||||
|
||||
|
||||
def test__new_output_generate_filename():
|
||||
m = CaptureManager()
|
||||
|
||||
out = m._new_output(True, None, "subfolder", "fmt")
|
||||
# Assert a filename was actually generated
|
||||
assert out.name != ".fmt"
|
||||
|
||||
|
||||
def test_new_image():
|
||||
m = CaptureManager()
|
||||
out = m.new_image()
|
||||
capture_key = str(out.id)
|
||||
assert capture_key in m.images.keys()
|
||||
assert m.images[capture_key] is out
|
||||
|
||||
|
||||
def test_new_video():
|
||||
m = CaptureManager()
|
||||
out = m.new_video()
|
||||
capture_key = str(out.id)
|
||||
assert capture_key in m.videos.keys()
|
||||
assert m.videos[capture_key] is out
|
||||
|
||||
|
||||
def test_image_from_id_str():
|
||||
m = CaptureManager()
|
||||
out = m.new_image()
|
||||
capture_key = str(out.id)
|
||||
|
||||
assert m.image_from_id(capture_key) is out
|
||||
|
||||
|
||||
def test_image_from_id_int():
|
||||
m = CaptureManager()
|
||||
out = m.new_image()
|
||||
capture_key = int(out.id)
|
||||
|
||||
assert m.image_from_id(capture_key) is out
|
||||
|
||||
|
||||
def test_image_from_id_uuid():
|
||||
m = CaptureManager()
|
||||
out = m.new_image()
|
||||
capture_key = out.id
|
||||
|
||||
assert m.image_from_id(capture_key) is out
|
||||
|
||||
|
||||
def test_image_from_id_invalid():
|
||||
m = CaptureManager()
|
||||
out = m.new_image()
|
||||
capture_key = object()
|
||||
|
||||
with pytest.raises(TypeError):
|
||||
m.image_from_id(capture_key) is out
|
||||
|
||||
|
||||
def test_video_from_id_str():
|
||||
m = CaptureManager()
|
||||
out = m.new_video()
|
||||
capture_key = str(out.id)
|
||||
|
||||
assert m.video_from_id(capture_key) is out
|
||||
|
||||
|
||||
def test_video_from_id_int():
|
||||
m = CaptureManager()
|
||||
out = m.new_video()
|
||||
capture_key = int(out.id)
|
||||
|
||||
assert m.video_from_id(capture_key) is out
|
||||
|
||||
|
||||
def test_video_from_id_uuid():
|
||||
m = CaptureManager()
|
||||
out = m.new_video()
|
||||
capture_key = out.id
|
||||
|
||||
assert m.video_from_id(capture_key) is out
|
||||
|
||||
|
||||
def test_video_from_id_invalid():
|
||||
m = CaptureManager()
|
||||
out = m.new_video()
|
||||
capture_key = object()
|
||||
|
||||
with pytest.raises(TypeError):
|
||||
m.video_from_id(capture_key) is out
|
||||
|
||||
|
||||
@freeze_time("2020-11-27 12:00:01")
|
||||
def test_generate_numbered_basename():
|
||||
m = CaptureManager()
|
||||
for _ in range(10):
|
||||
m.new_image()
|
||||
generated_filenames = [obj.basename for obj in m.images.values()]
|
||||
# Assert enough names were generated
|
||||
assert len(generated_filenames) == 10
|
||||
# Assert all names are unique
|
||||
len(generated_filenames) > len(set(generated_filenames))
|
||||
# Assert all filenames are in the expected form
|
||||
for fname in generated_filenames:
|
||||
assert fname.startswith("2020-11-27_12-00-01")
|
||||
|
||||
|
||||
def test_context_manager():
|
||||
with CaptureManager() as m:
|
||||
m.new_image()
|
||||
m.new_video()
|
||||
|
|
@ -1,232 +0,0 @@
|
|||
import datetime
|
||||
import io
|
||||
import os
|
||||
import uuid
|
||||
from collections import OrderedDict
|
||||
|
||||
import piexif
|
||||
from PIL import Image, ImageChops
|
||||
|
||||
from openflexure_microscope.captures import THUMBNAIL_SIZE, CaptureObject
|
||||
from openflexure_microscope.captures.capture import (
|
||||
build_captures_from_exif,
|
||||
capture_from_path,
|
||||
)
|
||||
|
||||
IN_DIR = "tests/images/"
|
||||
OUT_DIR = "tests/images/out/"
|
||||
|
||||
|
||||
def generate_small_image():
|
||||
# Create a dummy image to serve in the stream
|
||||
image = Image.new("RGB", (20, 20), color=(0, 0, 0))
|
||||
|
||||
return image
|
||||
|
||||
|
||||
def generate_small_thumbnail():
|
||||
# Create a dummy image to serve in the stream
|
||||
image = Image.new("RGB", (20, 20), color=(0, 0, 0))
|
||||
image.thumbnail(THUMBNAIL_SIZE)
|
||||
|
||||
return image
|
||||
|
||||
|
||||
def generate_small_capture(filename: str):
|
||||
out = os.path.join(OUT_DIR, filename)
|
||||
obj = CaptureObject(out)
|
||||
generate_small_image().save(obj, format="JPEG")
|
||||
|
||||
return obj
|
||||
|
||||
|
||||
def check_valid_capture(obj: CaptureObject):
|
||||
assert isinstance(obj.time, datetime.datetime)
|
||||
assert isinstance(obj.id, uuid.UUID)
|
||||
|
||||
assert isinstance(obj.annotations, dict)
|
||||
assert isinstance(obj.tags, list)
|
||||
|
||||
assert obj.file
|
||||
assert obj.format
|
||||
assert obj.basename
|
||||
assert obj.filefolder
|
||||
assert obj.name
|
||||
|
||||
|
||||
def check_valid_openflexure_metadata(d: dict):
|
||||
assert "image" in d
|
||||
assert d["image"].get("id")
|
||||
assert d["image"].get("name")
|
||||
assert d["image"].get("time")
|
||||
assert d["image"].get("format")
|
||||
|
||||
assert isinstance(d["image"].get("tags"), list)
|
||||
assert isinstance(d["image"].get("annotations"), dict)
|
||||
|
||||
|
||||
def test_new_capture_object():
|
||||
obj = generate_small_capture("capture0.jpg")
|
||||
check_valid_capture(obj)
|
||||
|
||||
|
||||
def test_initial_metadata():
|
||||
obj = generate_small_capture("capture1.jpg")
|
||||
|
||||
data = obj.read_full_metadata()
|
||||
check_valid_openflexure_metadata(data)
|
||||
|
||||
|
||||
def test_tags():
|
||||
obj = generate_small_capture("capture2.jpg")
|
||||
|
||||
obj.put_tags(["foo", "bar"])
|
||||
assert obj.read_full_metadata()["image"]["tags"] == ["foo", "bar"]
|
||||
|
||||
obj.delete_tag("foo")
|
||||
assert obj.read_full_metadata()["image"]["tags"] == ["bar"]
|
||||
|
||||
# Delete nonexistent tag, should change nothing
|
||||
obj.delete_tag("foo")
|
||||
assert obj.read_full_metadata()["image"]["tags"] == ["bar"]
|
||||
|
||||
|
||||
def test_annotations():
|
||||
obj = generate_small_capture("capture3.jpg")
|
||||
|
||||
obj.put_annotations({"foo": "zoo", "bar": "zar"})
|
||||
assert obj.read_full_metadata()["image"]["annotations"] == {
|
||||
"foo": "zoo",
|
||||
"bar": "zar",
|
||||
}
|
||||
|
||||
obj.delete_annotation("foo")
|
||||
assert obj.read_full_metadata()["image"]["annotations"] == {"bar": "zar"}
|
||||
|
||||
# Delete nonexistent annotation, should change nothing
|
||||
obj.delete_annotation("foo")
|
||||
assert obj.read_full_metadata()["image"]["annotations"] == {"bar": "zar"}
|
||||
|
||||
|
||||
def test_put_metadata():
|
||||
obj = generate_small_capture("capture4.jpg")
|
||||
|
||||
obj.put_metadata({"foo": {"bar": "zar"}})
|
||||
assert obj.read_full_metadata()["foo"] == {"bar": "zar"}
|
||||
|
||||
|
||||
def test_put_and_save():
|
||||
obj = generate_small_capture("capture_all.jpg")
|
||||
obj.put_and_save(
|
||||
tags=["foo", "bar"],
|
||||
annotations={"foo": "zoo", "bar": "zar"},
|
||||
metadata={"foo": {"bar": "zar"}},
|
||||
)
|
||||
|
||||
full_metadata = obj.read_full_metadata()
|
||||
assert full_metadata["image"]["tags"] == ["foo", "bar"]
|
||||
assert full_metadata["image"]["annotations"] == {"foo": "zoo", "bar": "zar"}
|
||||
assert full_metadata["foo"] == {"bar": "zar"}
|
||||
|
||||
|
||||
def test_capture_from_path_this_version():
|
||||
"""Tests reloading a capture object from a file created in the working server version
|
||||
"""
|
||||
|
||||
def _check_metadata(data: dict):
|
||||
assert data["image"]["tags"] == ["foo", "bar"]
|
||||
assert data["image"]["annotations"] == {"foo": "zoo", "bar": "zar"}
|
||||
assert data["foo"] == {"bar": "zar"}
|
||||
|
||||
# Create the capture file
|
||||
obj_in = generate_small_capture("capture_reload.jpg")
|
||||
obj_in.put_and_save(
|
||||
tags=["foo", "bar"],
|
||||
annotations={"foo": "zoo", "bar": "zar"},
|
||||
metadata={"foo": {"bar": "zar"}},
|
||||
)
|
||||
|
||||
full_metadata_in = obj_in.read_full_metadata()
|
||||
check_valid_openflexure_metadata(full_metadata_in)
|
||||
_check_metadata(full_metadata_in)
|
||||
|
||||
# Reload the capture file
|
||||
obj = capture_from_path(os.path.join(OUT_DIR, "capture_reload.jpg"))
|
||||
check_valid_capture(obj)
|
||||
full_metadata = obj.read_full_metadata()
|
||||
check_valid_openflexure_metadata(full_metadata)
|
||||
_check_metadata(full_metadata)
|
||||
|
||||
|
||||
def test_capture_from_path_v280():
|
||||
"""Tests reloading a capture object from a file created in server v2.8.0
|
||||
"""
|
||||
obj = capture_from_path(os.path.join(IN_DIR, "capture_v280.jpg"))
|
||||
check_valid_capture(obj)
|
||||
full_metadata = obj.read_full_metadata()
|
||||
check_valid_openflexure_metadata(full_metadata)
|
||||
assert full_metadata["image"]["tags"] == ["foo", "bar"]
|
||||
assert full_metadata["image"]["annotations"] == {"foo": "zoo", "bar": "zar"}
|
||||
assert full_metadata["foo"] == {"bar": "zar"}
|
||||
|
||||
|
||||
def test_build_captures_from_exif():
|
||||
# Load the whole tests captures directory
|
||||
objs = build_captures_from_exif(IN_DIR)
|
||||
# Make sure we have a valid, populated OrderedDict
|
||||
assert isinstance(objs, OrderedDict)
|
||||
assert len(objs) > 0
|
||||
# Check each reloaded capture
|
||||
for obj in objs.values():
|
||||
check_valid_capture(obj)
|
||||
|
||||
|
||||
def test_data():
|
||||
# Get a reference PIL image
|
||||
ref_data = generate_small_image()
|
||||
# Generate a capture with the same image data as our reference
|
||||
obj = generate_small_capture("capture5.jpg")
|
||||
# Pull the capture data out
|
||||
obj_data = Image.open(obj.data)
|
||||
# Compare the images
|
||||
diff = ImageChops.difference(obj_data, ref_data)
|
||||
assert not diff.getbbox()
|
||||
|
||||
|
||||
def test_binary():
|
||||
# Get a reference PIL image
|
||||
ref_data = generate_small_image()
|
||||
# Generate a capture with the same image data as our reference
|
||||
obj = generate_small_capture("capture6.jpg")
|
||||
# Pull the capture data out
|
||||
obj_data = Image.open(io.BytesIO(obj.binary))
|
||||
# Compare the images
|
||||
diff = ImageChops.difference(obj_data, ref_data)
|
||||
assert not diff.getbbox()
|
||||
|
||||
|
||||
def test_thumbnail():
|
||||
# Get a reference PIL image
|
||||
ref_thumb = generate_small_thumbnail()
|
||||
# Generate a capture with the same image data as our reference
|
||||
obj = generate_small_capture("capture7.jpg")
|
||||
# Pull the generated thumbnail out
|
||||
obj_thumb = Image.open(obj.thumbnail)
|
||||
# Compare the images
|
||||
diff = ImageChops.difference(obj_thumb, ref_thumb)
|
||||
assert not diff.getbbox()
|
||||
|
||||
# Assert thumbnail was saved to capture file EXIF data
|
||||
exif_dict = piexif.load(obj.file)
|
||||
thumbnail = exif_dict.pop("thumbnail")
|
||||
assert thumbnail
|
||||
# Compare the images
|
||||
diff = ImageChops.difference(Image.open(io.BytesIO(thumbnail)), ref_thumb)
|
||||
|
||||
|
||||
def test_delete():
|
||||
# Generate a capture with the same image data as our reference
|
||||
obj = generate_small_capture("capture_del.jpg")
|
||||
assert os.path.isfile(obj.file)
|
||||
obj.delete()
|
||||
assert not os.path.isfile(obj.file)
|
||||
22
tests/test_dummy_server.py
Normal file
22
tests/test_dummy_server.py
Normal file
|
|
@ -0,0 +1,22 @@
|
|||
import tempfile
|
||||
|
||||
from fastapi import Depends, FastAPI
|
||||
from fastapi.testclient import TestClient
|
||||
|
||||
from labthings_fastapi.client import ThingClient
|
||||
|
||||
from openflexure_microscope_server.server import ThingServer
|
||||
from openflexure_microscope_server.things.camera.simulation import SimulatedCamera
|
||||
from openflexure_microscope_server.things.stage.dummy import DummyStage
|
||||
from openflexure_microscope_server.things.autofocus import AutofocusThing
|
||||
|
||||
temp_folder = tempfile.TemporaryDirectory()
|
||||
server = ThingServer(temp_folder.name)
|
||||
server.add_thing(SimulatedCamera(), "/camera/")
|
||||
server.add_thing(DummyStage(), "/stage/")
|
||||
server.add_thing(AutofocusThing(), "/autofocus/")
|
||||
|
||||
def test_autofocus():
|
||||
with TestClient(server.app) as client:
|
||||
autofocus = ThingClient.from_url("/autofocus/", client)
|
||||
_ = autofocus.fast_autofocus()
|
||||
|
|
@ -1,57 +0,0 @@
|
|||
from openflexure_microscope.api.default_extensions import scan
|
||||
|
||||
|
||||
def test_construct_grid_raster():
|
||||
grid = scan.construct_grid((0, 0), (100, 100), (3, 3), style="raster")
|
||||
assert grid == [
|
||||
[(0, 0), (0, 100), (0, 200)],
|
||||
[(100, 0), (100, 100), (100, 200)],
|
||||
[(200, 0), (200, 100), (200, 200)],
|
||||
]
|
||||
|
||||
|
||||
def test_construct_grid_snake():
|
||||
grid = scan.construct_grid((0, 0), (100, 100), (3, 3), style="snake")
|
||||
assert grid == [
|
||||
[(0, 0), (0, 100), (0, 200)],
|
||||
[(100, 200), (100, 100), (100, 0)],
|
||||
[(200, 0), (200, 100), (200, 200)],
|
||||
]
|
||||
|
||||
|
||||
def test_construct_grid_spiral():
|
||||
grid = scan.construct_grid((0, 0), (100, 100), (3, 0), style="spiral")
|
||||
assert grid == [
|
||||
[(0, 0)],
|
||||
[
|
||||
(0, 100),
|
||||
(100, 100),
|
||||
(100, 0),
|
||||
(100, -100),
|
||||
(0, -100),
|
||||
(-100, -100),
|
||||
(-100, 0),
|
||||
(-100, 100),
|
||||
],
|
||||
[
|
||||
(-100, 200),
|
||||
(0, 200),
|
||||
(100, 200),
|
||||
(200, 200),
|
||||
(200, 100),
|
||||
(200, 0),
|
||||
(200, -100),
|
||||
(200, -200),
|
||||
(100, -200),
|
||||
(0, -200),
|
||||
(-100, -200),
|
||||
(-200, -200),
|
||||
(-200, -100),
|
||||
(-200, 0),
|
||||
(-200, 100),
|
||||
(-200, 200),
|
||||
],
|
||||
]
|
||||
|
||||
# Ensure second n_steps value makes no difference
|
||||
assert grid == scan.construct_grid((0, 0), (100, 100), (3, 3), style="spiral")
|
||||
|
|
@ -1,59 +0,0 @@
|
|||
import json
|
||||
import uuid
|
||||
from fractions import Fraction
|
||||
|
||||
import numpy as np
|
||||
import pytest
|
||||
|
||||
from openflexure_microscope.json import JSONEncoder
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"test_input,expected",
|
||||
[
|
||||
(uuid.uuid1(), 1),
|
||||
(uuid.uuid3(uuid.NAMESPACE_DNS, "openflexure.org"), 3),
|
||||
(uuid.uuid4(), 4),
|
||||
(uuid.uuid5(uuid.NAMESPACE_DNS, "openflexure.org"), 5),
|
||||
],
|
||||
)
|
||||
def test_encode_uuid(test_input, expected):
|
||||
encoded = json.dumps(test_input, cls=JSONEncoder)
|
||||
decoded = json.loads(encoded)
|
||||
assert uuid.UUID(decoded).version == expected
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"test_input",
|
||||
[
|
||||
Fraction(1, 2),
|
||||
Fraction(1, 3),
|
||||
Fraction(-27, 4),
|
||||
Fraction(27, -4),
|
||||
Fraction(10),
|
||||
Fraction(18, 63),
|
||||
Fraction(1, 1000000),
|
||||
],
|
||||
)
|
||||
def test_encode_fraction(test_input):
|
||||
encoded = json.dumps(test_input, cls=JSONEncoder)
|
||||
decoded = json.loads(encoded)
|
||||
assert Fraction(decoded).limit_denominator() == test_input
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"test_input", [np.random.randint(100, dtype=np.int64) for _ in range(10)]
|
||||
)
|
||||
def test_encode_np_int(test_input):
|
||||
encoded = json.dumps(test_input, cls=JSONEncoder)
|
||||
decoded = json.loads(encoded)
|
||||
assert np.int64(decoded) == test_input
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"test_input", [np.float64(np.random.random()) for _ in range(10)]
|
||||
)
|
||||
def test_encode_np_float(test_input):
|
||||
encoded = json.dumps(test_input, cls=JSONEncoder)
|
||||
decoded = json.loads(encoded)
|
||||
assert np.float64(decoded) == test_input
|
||||
|
|
@ -1,31 +0,0 @@
|
|||
import numpy as np
|
||||
|
||||
from openflexure_microscope import utilities
|
||||
|
||||
|
||||
def test_serialise_array_b64():
|
||||
shape_in = (3, 2)
|
||||
arr_in = np.random.randint(100, size=shape_in, dtype=np.int32)
|
||||
|
||||
b64_string, dtype, shape = utilities.serialise_array_b64(arr_in)
|
||||
assert b64_string
|
||||
assert dtype == "int32"
|
||||
assert shape == shape_in
|
||||
|
||||
arr_out = utilities.deserialise_array_b64(b64_string, dtype, shape)
|
||||
assert np.array_equal(arr_out, arr_in)
|
||||
|
||||
|
||||
def test_ndarray_to_json():
|
||||
shape_in = (3, 2)
|
||||
arr_in = np.random.randint(100, size=shape_in, dtype=np.int32)
|
||||
|
||||
json_out = utilities.ndarray_to_json(arr_in)
|
||||
|
||||
arr_out = utilities.json_to_ndarray(json_out)
|
||||
assert np.array_equal(arr_out, arr_in)
|
||||
|
||||
|
||||
def test_axes_to_array():
|
||||
dict_in = {"x": 1, "y": 2, "z": 3}
|
||||
assert utilities.axes_to_array(dict_in) == [1, 2, 3]
|
||||
File diff suppressed because it is too large
Load diff
Loading…
Add table
Add a link
Reference in a new issue