Merge branch 'ANN-rules' into 'v3'
Enable ruff checks for type annotations See merge request openflexure/openflexure-microscope-server!382
This commit is contained in:
Julian Stirling
commit
cb62bc3cc6
26 changed files with 252 additions and 156 deletions
|
|
@ -12,7 +12,7 @@ from gitlab.exceptions import GitlabGetError
|
|||
PROJECT_ID = 9238334
|
||||
|
||||
|
||||
def main(tag, branch: str = "v3"):
|
||||
def main(tag: str, branch: str = "v3") -> None:
|
||||
"""Create a list of the MRs into a given branch after a tag.
|
||||
|
||||
:param tag: The tag the mrs should be since.
|
||||
|
|
|
|||
Binary file not shown.
|
|
@ -43,7 +43,7 @@ def extract_zip(zip_data: bytes) -> None:
|
|||
print(f"Extracted zip to: {THIS_DIR}")
|
||||
|
||||
|
||||
def main(branch: str = "v3"):
|
||||
def main(branch: str = "v3") -> None:
|
||||
"""Fetch build artifacts from GitLab for a given branch and unzip.
|
||||
|
||||
Clear any existing static directory before extract the new files.
|
||||
|
|
|
|||
|
|
@ -124,8 +124,9 @@ select = [
|
|||
"C90", # McCabe complexity!
|
||||
"NPY", # Numpy linting
|
||||
"N", # PEP8 naming
|
||||
"D", # Docstring checks these may need to be added gradually
|
||||
"D", # Docstring checks
|
||||
"ERA001", # Commented out code!
|
||||
"ANN",
|
||||
]
|
||||
|
||||
ignore = [
|
||||
|
|
@ -133,13 +134,19 @@ ignore = [
|
|||
"D203", # incompatible with D204
|
||||
"D213", # incompatible with D212
|
||||
"D400", # A stricter version of #415 that doesn't allow !
|
||||
# The checkers below should be turned on as they complain about missing docstrings.
|
||||
"ANN401", # Disalows Any, Any is needed at times, Once MyPy is running Any will be
|
||||
# handled appropriately
|
||||
]
|
||||
|
||||
[tool.ruff.lint.per-file-ignores]
|
||||
# Tests are currently not fully docstring-ed, we'll ignore this for now.
|
||||
"tests/*" = [
|
||||
"B018", # Complaining about useless attribute access in tests, but we need them to check errors are raised
|
||||
"ANN", # Tests are not typehinted for fixtures etc
|
||||
]
|
||||
"hardware-specific-tests/*" = [
|
||||
"B018", # Complaining about useless attribute access in tests, but we need them to check errors are raised
|
||||
"ANN", # Tests are not typehinted for fixtures etc
|
||||
]
|
||||
|
||||
[tool.ruff.lint.pydocstyle]
|
||||
|
|
|
|||
|
|
@ -55,7 +55,7 @@ class BackgroundDetectAlgorithm:
|
|||
settings_data_model: BaseModel = BaseModel
|
||||
"""The data model of algorithm settings. This must be set by child classes"""
|
||||
|
||||
def __init__(self):
|
||||
def __init__(self) -> None:
|
||||
"""Initialise the algorithm settings."""
|
||||
try:
|
||||
self._settings: BaseModel = self.settings_data_model()
|
||||
|
|
|
|||
|
|
@ -21,7 +21,7 @@ from fastapi import HTTPException
|
|||
OFM_LOG_FILE = None
|
||||
|
||||
|
||||
def configure_logging(log_folder):
|
||||
def configure_logging(log_folder: str) -> None:
|
||||
"""Configure logging for the server while it is running.
|
||||
|
||||
This modifies the root logger to have a rotating file handler and
|
||||
|
|
@ -103,7 +103,7 @@ def retrieve_log_from_file() -> PlainTextResponse:
|
|||
class OFMLogFileFormatter(logging.Formatter):
|
||||
"""The formatter used for the OpenFlexure Microscope Server log file."""
|
||||
|
||||
def format(self, record: logging.LogRecord):
|
||||
def format(self, record: logging.LogRecord) -> str:
|
||||
"""Adjust the logging formatting for uvicorn logs.
|
||||
|
||||
uvicorn has two loggers. Each API access is ``uvicorn.access`` which we filter
|
||||
|
|
@ -120,7 +120,7 @@ class OFMLogFileFormatter(logging.Formatter):
|
|||
class OFMHandler(logging.Handler):
|
||||
"""A logging.Handler that stores the most recent logs for access by the server."""
|
||||
|
||||
def __init__(self, level=logging.INFO, max_logs=250):
|
||||
def __init__(self, level: int = logging.INFO, max_logs: int = 250) -> None:
|
||||
"""Initialise the handler with a set logging level and message buffer size.
|
||||
|
||||
:param level: The level of logs captured. As standard logs of INFO and above
|
||||
|
|
@ -132,7 +132,7 @@ class OFMHandler(logging.Handler):
|
|||
self._log = []
|
||||
self._max_logs = max_logs
|
||||
|
||||
def append_record(self, record: logging.LogRecord):
|
||||
def append_record(self, record: logging.LogRecord) -> None:
|
||||
"""Format message and append it to a list of records.
|
||||
|
||||
The built in formatter is used to format the record.
|
||||
|
|
@ -143,7 +143,7 @@ class OFMHandler(logging.Handler):
|
|||
while len(self._log) > self._max_logs:
|
||||
self._log.pop(0)
|
||||
|
||||
def emit(self, record: logging.LogRecord):
|
||||
def emit(self, record: logging.LogRecord) -> None:
|
||||
"""Emit will save the logged record to the log."""
|
||||
try:
|
||||
if record.levelno >= self.level:
|
||||
|
|
@ -156,7 +156,7 @@ class OFMHandler(logging.Handler):
|
|||
self.handleError(record)
|
||||
|
||||
@property
|
||||
def log_history(self):
|
||||
def log_history(self) -> str:
|
||||
"""Return the log history up to the maximum number of logs."""
|
||||
return "\n".join(self._log)
|
||||
|
||||
|
|
@ -164,7 +164,7 @@ class OFMHandler(logging.Handler):
|
|||
class UvicornAccessFilter(logging.Filter):
|
||||
"""A logging filter to filter out "uvicorn.access" messages."""
|
||||
|
||||
def filter(self, record: logging.LogRecord):
|
||||
def filter(self, record: logging.LogRecord) -> bool:
|
||||
"""Return False if record is from "uvicorn.access"."""
|
||||
return not record.name.startswith("uvicorn.access")
|
||||
|
||||
|
|
|
|||
|
|
@ -105,7 +105,7 @@ class ScanData(BaseModel):
|
|||
This should be set with ``set_final_data()`` to ensure duration is set.
|
||||
"""
|
||||
|
||||
def set_final_data(self, result: str):
|
||||
def set_final_data(self, result: str) -> None:
|
||||
"""Set the final data for the scan, scan duration is automatically calculated.
|
||||
|
||||
:param result: A string describing the result.
|
||||
|
|
@ -167,7 +167,7 @@ class ScanDirectoryManager:
|
|||
|
||||
_base_scan_dir: str
|
||||
|
||||
def __init__(self, base_scan_dir: str):
|
||||
def __init__(self, base_scan_dir: str) -> None:
|
||||
"""Initialise the scan directory manager.
|
||||
|
||||
:param base_scan_dir: Path of the directory that holds all scans.
|
||||
|
|
@ -374,7 +374,7 @@ class ScanDirectory:
|
|||
_name: str
|
||||
_base_scan_dir: str
|
||||
|
||||
def __init__(self, name: str, base_scan_dir: str):
|
||||
def __init__(self, name: str, base_scan_dir: str) -> None:
|
||||
"""Initialise the scan directory.
|
||||
|
||||
:param name: the name of the scan (the scan directory basename).
|
||||
|
|
@ -423,13 +423,13 @@ class ScanDirectory:
|
|||
"""The time the directory was created on disk."""
|
||||
return os.path.getctime(self.dir_path)
|
||||
|
||||
def get_scan_files(self):
|
||||
def get_scan_files(self) -> list[str]:
|
||||
"""Return a list of the files in the images dir."""
|
||||
if self.images_dir is None:
|
||||
return []
|
||||
return os.listdir(self.images_dir)
|
||||
|
||||
def _extract_scan_images(self, file_list: list[str]):
|
||||
def _extract_scan_images(self, file_list: list[str]) -> list[str]:
|
||||
"""Extract files which match the naming convention for scan images.
|
||||
|
||||
:param file_list: The list of files to search. Normally this would be
|
||||
|
|
@ -439,7 +439,7 @@ class ScanDirectory:
|
|||
"""
|
||||
return [i for i in file_list if IMAGE_REGEX.search(i)]
|
||||
|
||||
def _extract_final_stitches(self, file_list: list[str]):
|
||||
def _extract_final_stitches(self, file_list: list[str]) -> list[str]:
|
||||
"""Extract files which match the naming convention for final stitches.
|
||||
|
||||
:param file_list: The list of files to search.
|
||||
|
|
@ -448,7 +448,7 @@ class ScanDirectory:
|
|||
"""
|
||||
return [i for i in file_list if STITCH_REGEX.search(i)]
|
||||
|
||||
def _extract_dzi_files(self, file_list: list[str]):
|
||||
def _extract_dzi_files(self, file_list: list[str]) -> list[str]:
|
||||
"""Extract files which match the naming convention for dzi_files.
|
||||
|
||||
:param file_list: The list of files to search.
|
||||
|
|
@ -509,7 +509,7 @@ class ScanDirectory:
|
|||
files.append(os.path.relpath(full_path, self.dir_path))
|
||||
return files
|
||||
|
||||
def save_scan_data(self, scan_data: ScanData):
|
||||
def save_scan_data(self, scan_data: ScanData) -> None:
|
||||
"""Save the scan data for this scan to disk."""
|
||||
if self.scan_data_path is None:
|
||||
raise FileNotFoundError(
|
||||
|
|
@ -551,7 +551,7 @@ class ScanDirectory:
|
|||
return zip_fname
|
||||
|
||||
|
||||
def get_files_in_zip(zip_path):
|
||||
def get_files_in_zip(zip_path: str) -> list[str]:
|
||||
"""List the relative paths of all files and folders in the zip folder specified."""
|
||||
scan_zip = zipfile.ZipFile(zip_path)
|
||||
return [os.path.normpath(i) for i in scan_zip.namelist()]
|
||||
|
|
|
|||
|
|
@ -80,7 +80,7 @@ class ScanPlanner:
|
|||
|
||||
def __init__(
|
||||
self, initial_position: XYPos, planner_settings: Optional[dict] = None
|
||||
):
|
||||
) -> None:
|
||||
"""Set up lists for the path planning, and scan history."""
|
||||
self._initial_position = enforce_xy_tuple(initial_position)
|
||||
self._parse(planner_settings)
|
||||
|
|
@ -332,7 +332,7 @@ class SmartSpiral(ScanPlanner):
|
|||
"""Sort the remaining positions based on the current location."""
|
||||
|
||||
# Defined rather than use a lambda for readability
|
||||
def sort_key(pos):
|
||||
def sort_key(pos: XYPos) -> tuple[float, float, float]:
|
||||
return (
|
||||
self.moves_between(current_pos, pos),
|
||||
self.moves_between(self._initial_position, pos),
|
||||
|
|
|
|||
|
|
@ -2,20 +2,20 @@
|
|||
|
||||
from __future__ import annotations
|
||||
|
||||
from typing import Optional, Callable
|
||||
from typing import Optional, Callable, Any
|
||||
from functools import wraps
|
||||
from copy import copy
|
||||
import logging
|
||||
|
||||
import labthings_fastapi as lt
|
||||
import uvicorn
|
||||
import logging
|
||||
from uvicorn.main import Server
|
||||
from .serve_static_files import add_static_files
|
||||
from .legacy_api import add_v2_endpoints
|
||||
from ..logging import configure_logging, retrieve_log, retrieve_log_from_file
|
||||
|
||||
|
||||
def set_shutdown_function(shutdown_function: Callable[[], None]):
|
||||
def set_shutdown_function(shutdown_function: Callable[[], None]) -> None:
|
||||
"""Ensure a function is called before the shutdown.
|
||||
|
||||
This monkey patches the Uvicorn Server's handle_exit. This is needed because
|
||||
|
|
@ -32,7 +32,7 @@ def set_shutdown_function(shutdown_function: Callable[[], None]):
|
|||
original_handler = Server.handle_exit
|
||||
|
||||
@wraps(Server.handle_exit)
|
||||
def handle_exit(*args, **kwargs):
|
||||
def handle_exit(*args: Any, **kwargs: Any) -> None:
|
||||
shutdown_function()
|
||||
original_handler(*args, **kwargs)
|
||||
|
||||
|
|
@ -41,7 +41,7 @@ def set_shutdown_function(shutdown_function: Callable[[], None]):
|
|||
|
||||
def customise_server(
|
||||
server: lt.ThingServer, log_folder: str, scans_folder: Optional[str]
|
||||
):
|
||||
) -> None:
|
||||
"""Customise the server with additional endpoints, etc."""
|
||||
configure_logging(log_folder)
|
||||
add_v2_endpoints(server)
|
||||
|
|
@ -66,7 +66,7 @@ def _get_scans_dir(config: dict) -> Optional[str]:
|
|||
return None
|
||||
|
||||
|
||||
def serve_from_cli(argv: Optional[list[str]] = None):
|
||||
def serve_from_cli(argv: Optional[list[str]] = None) -> None:
|
||||
"""Start the server from the command line."""
|
||||
args = lt.cli.parse_args(argv)
|
||||
|
||||
|
|
@ -85,7 +85,7 @@ def serve_from_cli(argv: Optional[list[str]] = None):
|
|||
server = lt.cli.server_from_config(config)
|
||||
customise_server(server, log_folder, scans_folder)
|
||||
|
||||
def shutdown_call():
|
||||
def shutdown_call() -> None:
|
||||
try:
|
||||
# Kill any mjpeg streams so that StreamingResponses close.
|
||||
server.things["/camera/"].kill_mjpeg_streams()
|
||||
|
|
|
|||
|
|
@ -50,7 +50,7 @@ def add_static_files(app: FastAPI, scans_folder: Optional[str]) -> None:
|
|||
check_static_dir()
|
||||
|
||||
@app.get("/", response_class=RedirectResponse)
|
||||
async def redirect_fastapi():
|
||||
async def redirect_fastapi() -> str:
|
||||
return "/index.html"
|
||||
|
||||
# Mounting the webapp at / file by file to allow other endpoints to be created
|
||||
|
|
@ -77,7 +77,7 @@ def add_static_files(app: FastAPI, scans_folder: Optional[str]) -> None:
|
|||
)
|
||||
|
||||
|
||||
def check_static_dir():
|
||||
def check_static_dir() -> None:
|
||||
"""Check that the static dir exists and contains expected files and dirs."""
|
||||
if not os.path.isdir(STATIC_PATH):
|
||||
raise FileNotFoundError(
|
||||
|
|
|
|||
|
|
@ -34,7 +34,7 @@ class StitcherValidationError(RuntimeError):
|
|||
"""The stitcher received values that it deems unsafe to create a command from."""
|
||||
|
||||
|
||||
def validate_command(cmd: list[str]):
|
||||
def validate_command(cmd: list[str]) -> None:
|
||||
"""Validate that the command only characters that are allowed in a path.
|
||||
|
||||
The values in the commands should be numbers, commandline flags, paths, and
|
||||
|
|
@ -59,7 +59,9 @@ class BaseStitcher:
|
|||
complete.
|
||||
"""
|
||||
|
||||
def __init__(self, images_dir: str, *, overlap: float, correlation_resize: float):
|
||||
def __init__(
|
||||
self, images_dir: str, *, overlap: float, correlation_resize: float
|
||||
) -> None:
|
||||
"""Initialise a stitcher.
|
||||
|
||||
All args except images_dir are positional only.
|
||||
|
|
@ -128,7 +130,9 @@ class PreviewStitcher(BaseStitcher):
|
|||
one preview must finish before another can be started.
|
||||
"""
|
||||
|
||||
def __init__(self, images_dir: str, *, overlap: float, correlation_resize: float):
|
||||
def __init__(
|
||||
self, images_dir: str, *, overlap: float, correlation_resize: float
|
||||
) -> None:
|
||||
"""Initialise a preview stitcher.
|
||||
|
||||
All args except images_dir are positional only.
|
||||
|
|
@ -185,14 +189,14 @@ class FinalStitcher(BaseStitcher):
|
|||
|
||||
def __init__(
|
||||
self,
|
||||
images_dir,
|
||||
images_dir: str,
|
||||
*,
|
||||
logger: lt.deps.InvocationLogger,
|
||||
overlap: Optional[float] = None,
|
||||
correlation_resize: Optional[float] = None,
|
||||
stitch_tiff: bool = False,
|
||||
scan_data_dict: Optional[dict[str, Any]] = None,
|
||||
):
|
||||
) -> None:
|
||||
"""Initialise a final stitcher, this has more args than the base class.
|
||||
|
||||
All args except images_dir are positional only.
|
||||
|
|
@ -340,7 +344,7 @@ class FinalStitcher(BaseStitcher):
|
|||
# Print everything in the buffer when program finishes
|
||||
self.log_buffer(process.stdout)
|
||||
|
||||
def log_buffer(self, buffer: TextIOWrapper):
|
||||
def log_buffer(self, buffer: TextIOWrapper) -> None:
|
||||
"""Log everything in the buffer at INFO level."""
|
||||
while line := buffer.readline():
|
||||
self.logger.info(line)
|
||||
|
|
|
|||
|
|
@ -127,7 +127,7 @@ class StackParams:
|
|||
"""
|
||||
return self.min_images_to_test + 15
|
||||
|
||||
def slice_to_save(self, sharpest_index):
|
||||
def slice_to_save(self, sharpest_index: int) -> slice:
|
||||
"""Return the slice of images to save given the index of the sharpest image."""
|
||||
images_each_side = (self.images_to_save - 1) // 2
|
||||
return slice(
|
||||
|
|
@ -215,7 +215,9 @@ class JPEGSharpnessMonitor:
|
|||
SharpnessMonitorDep as an argument is called.
|
||||
"""
|
||||
|
||||
def __init__(self, stage: Stage, camera: RawCamera, portal: lt.deps.BlockingPortal):
|
||||
def __init__(
|
||||
self, stage: Stage, camera: RawCamera, portal: lt.deps.BlockingPortal
|
||||
) -> None:
|
||||
"""Initialise a new JPEGSharpnessMonitor. The args are injected automatically.
|
||||
|
||||
:param stage: A direct_thing_client dependency for the the microscope stage.
|
||||
|
|
@ -234,7 +236,7 @@ class JPEGSharpnessMonitor:
|
|||
|
||||
running = False
|
||||
|
||||
async def monitor_sharpness(self):
|
||||
async def monitor_sharpness(self) -> None:
|
||||
"""Start monitoring the frame sizes."""
|
||||
self.running = True
|
||||
async for frame in self.camera.lores_mjpeg_stream.frame_async_generator():
|
||||
|
|
@ -244,7 +246,7 @@ class JPEGSharpnessMonitor:
|
|||
break
|
||||
|
||||
@contextmanager
|
||||
def run(self):
|
||||
def run(self) -> None:
|
||||
"""Context manager, during which we will monitor sharpness from the camera."""
|
||||
self.portal.start_task_soon(self.monitor_sharpness)
|
||||
try:
|
||||
|
|
@ -252,7 +254,7 @@ class JPEGSharpnessMonitor:
|
|||
finally:
|
||||
self.running = False
|
||||
|
||||
def focus_rel(self, dz: int, **kwargs) -> tuple[int, int]:
|
||||
def focus_rel(self, dz: int, block_cancellation: int = False) -> tuple[int, int]:
|
||||
"""Move the stage by dz, monitoring the position over time.
|
||||
|
||||
This performs exactly one move. Multiple calls of this method
|
||||
|
|
@ -268,7 +270,7 @@ class JPEGSharpnessMonitor:
|
|||
self.stage_positions.append(self.stage.position)
|
||||
|
||||
# Main move
|
||||
self.stage.move_relative(z=dz, **kwargs)
|
||||
self.stage.move_relative(z=dz, block_cancellation=block_cancellation)
|
||||
|
||||
# Store the end time and position
|
||||
self.stage_times.append(time.time())
|
||||
|
|
@ -401,9 +403,9 @@ class AutofocusThing(lt.Thing):
|
|||
self,
|
||||
stage: Stage,
|
||||
sharpness_monitor: SharpnessMonitorDep,
|
||||
dz=2000,
|
||||
dz: int = 2000,
|
||||
start: Literal["centre", "base"] = "centre",
|
||||
):
|
||||
) -> tuple[list[float], list[float]]:
|
||||
"""Repeatedly autofocus the stage until it looks focused.
|
||||
|
||||
This action will run the ``fast_autofocus`` action until it settles on a point
|
||||
|
|
|
|||
|
|
@ -8,6 +8,7 @@ See repository root for licensing information.
|
|||
|
||||
from __future__ import annotations
|
||||
from typing import Literal, Optional, Tuple, Any
|
||||
from types import TracebackType
|
||||
import json
|
||||
import io
|
||||
import time
|
||||
|
|
@ -68,7 +69,7 @@ class CameraMemoryBuffer:
|
|||
|
||||
_storage: dict[int, tuple[Any, Optional[dict]]]
|
||||
|
||||
def __init__(self):
|
||||
def __init__(self) -> None:
|
||||
"""Create the buffer instance."""
|
||||
# This dictionary is the main store for data. Dictionaries are ordered since
|
||||
# Python 3.6, so the order in the dictionary is the capture order
|
||||
|
|
@ -132,7 +133,7 @@ class CameraMemoryBuffer:
|
|||
"No image with matching id in memory to retrieve."
|
||||
) from e
|
||||
|
||||
def clear(self):
|
||||
def clear(self) -> None:
|
||||
"""Clear all images from memory."""
|
||||
self._storage.clear()
|
||||
|
||||
|
|
@ -169,7 +170,7 @@ class BaseCamera(lt.Thing):
|
|||
lores_mjpeg_stream = lt.outputs.MJPEGStreamDescriptor()
|
||||
_memory_buffer = CameraMemoryBuffer()
|
||||
|
||||
def __init__(self):
|
||||
def __init__(self) -> None:
|
||||
"""Initialise the base camera, this creates the background detectors.
|
||||
|
||||
This must be run by all child camera classes.
|
||||
|
|
@ -185,7 +186,12 @@ class BaseCamera(lt.Thing):
|
|||
"""Open hardware connection when the Thing context manager is opened."""
|
||||
raise NotImplementedError("CameraThings must define their own __enter__ method")
|
||||
|
||||
def __exit__(self, _exc_type, _exc_value, _traceback) -> None:
|
||||
def __exit__(
|
||||
self,
|
||||
_exc_type: type[BaseException],
|
||||
_exc_value: Optional[BaseException],
|
||||
_traceback: Optional[TracebackType],
|
||||
) -> None:
|
||||
"""Close hardware connection when the Thing context manager is closed."""
|
||||
raise NotImplementedError("CameraThings must define their own __exit__ method")
|
||||
|
||||
|
|
@ -202,7 +208,7 @@ class BaseCamera(lt.Thing):
|
|||
"CameraThings must define their own start_streaming method"
|
||||
)
|
||||
|
||||
def kill_mjpeg_streams(self):
|
||||
def kill_mjpeg_streams(self) -> None:
|
||||
"""Kill the streams now as the server is shutting down.
|
||||
|
||||
This is called when uvicorn gets the a shutdown signal. As this is called from
|
||||
|
|
@ -575,7 +581,7 @@ class BaseCamera(lt.Thing):
|
|||
return self.active_detector.status
|
||||
|
||||
@lt.thing_action
|
||||
def update_detector_settings(self, data) -> None:
|
||||
def update_detector_settings(self, data: dict[str, Any]) -> None:
|
||||
"""Update the settings of the current detector.
|
||||
|
||||
This is an action not a setting/property as the data model depends on the
|
||||
|
|
|
|||
|
|
@ -10,6 +10,7 @@ from __future__ import annotations
|
|||
|
||||
import logging
|
||||
from typing import Literal, Optional
|
||||
from types import TracebackType
|
||||
from threading import Thread
|
||||
|
||||
import cv2
|
||||
|
|
@ -24,7 +25,7 @@ from . import BaseCamera
|
|||
class OpenCVCamera(BaseCamera):
|
||||
"""A Thing that provides and interface to an OpenCV Camera."""
|
||||
|
||||
def __init__(self, camera_index: int = 0):
|
||||
def __init__(self, camera_index: int = 0) -> None:
|
||||
"""Iniatilise the thing storing the index of the camera to use.
|
||||
|
||||
:param camera_index: The index of the camera to use for the microscope.
|
||||
|
|
@ -34,7 +35,7 @@ class OpenCVCamera(BaseCamera):
|
|||
self._capture_thread: Optional[Thread] = None
|
||||
self._capture_enabled = False
|
||||
|
||||
def __enter__(self):
|
||||
def __enter__(self) -> None:
|
||||
"""Start the capture thread when the Thing context manager is opened."""
|
||||
self.cap = cv2.VideoCapture(self.camera_index)
|
||||
self._capture_enabled = True
|
||||
|
|
@ -42,7 +43,12 @@ class OpenCVCamera(BaseCamera):
|
|||
self._capture_thread.start()
|
||||
return self
|
||||
|
||||
def __exit__(self, _exc_type, _exc_value, _traceback):
|
||||
def __exit__(
|
||||
self,
|
||||
_exc_type: type[BaseException],
|
||||
_exc_value: Optional[BaseException],
|
||||
_traceback: Optional[TracebackType],
|
||||
) -> None:
|
||||
"""Release the camera when the Thing context manager is closed.
|
||||
|
||||
Before releasing the camera the capture thread is closed.
|
||||
|
|
@ -59,7 +65,7 @@ class OpenCVCamera(BaseCamera):
|
|||
return self._capture_thread.is_alive()
|
||||
return False
|
||||
|
||||
def _capture_frames(self):
|
||||
def _capture_frames(self) -> None:
|
||||
portal = lt.get_blocking_portal(self)
|
||||
while self._capture_enabled:
|
||||
ret, frame = self.cap.read()
|
||||
|
|
|
|||
|
|
@ -15,7 +15,8 @@ https://datasheets.raspberrypi.com/camera/raspberry-pi-camera-guide.pdf
|
|||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
from typing import Annotated, Iterator, Literal, Mapping, Optional, overload
|
||||
from typing import Annotated, Iterator, Literal, Mapping, Optional, overload, Any
|
||||
from types import TracebackType
|
||||
import json
|
||||
import logging
|
||||
import os
|
||||
|
|
@ -52,7 +53,9 @@ class MissingCalibrationError(RuntimeError):
|
|||
class PicameraStreamOutput(Output):
|
||||
"""An Output class that sends frames to a stream."""
|
||||
|
||||
def __init__(self, stream: lt.outputs.MJPEGStream, portal: lt.deps.BlockingPortal):
|
||||
def __init__(
|
||||
self, stream: lt.outputs.MJPEGStream, portal: lt.deps.BlockingPortal
|
||||
) -> None:
|
||||
"""Create an output that puts frames in an MJPEGStream.
|
||||
|
||||
We need to pass the stream object, and also the blocking portal, because
|
||||
|
|
@ -65,8 +68,13 @@ class PicameraStreamOutput(Output):
|
|||
self.portal = portal
|
||||
|
||||
def outputframe(
|
||||
self, frame, _keyframe=True, _timestamp=None, _packet=None, _audio=False
|
||||
):
|
||||
self,
|
||||
frame: bytes,
|
||||
_keyframe: Optional[bool] = True,
|
||||
_timestamp: Optional[int] = None,
|
||||
_packet: Any = None,
|
||||
_audio: bool = False,
|
||||
) -> None:
|
||||
"""Add a frame to the stream's ringbuffer."""
|
||||
self.stream.add_frame(frame, self.portal)
|
||||
|
||||
|
|
@ -120,7 +128,7 @@ class StreamingPiCamera2(BaseCamera):
|
|||
generalisation.
|
||||
"""
|
||||
|
||||
def __init__(self, camera_num: int = 0):
|
||||
def __init__(self, camera_num: int = 0) -> None:
|
||||
"""Initialise the camera with the given camera number.
|
||||
|
||||
This makes no connection to the camera (except to get the default tuning file).
|
||||
|
|
@ -167,7 +175,7 @@ class StreamingPiCamera2(BaseCamera):
|
|||
)
|
||||
"""Whether the MJPEG stream is active."""
|
||||
|
||||
def save_settings(self):
|
||||
def save_settings(self) -> None:
|
||||
"""Override save_settings to ensure that camera properties don't recurse.
|
||||
|
||||
This method is run by any Thing when a ThingSetting is saved. However, the
|
||||
|
|
@ -198,7 +206,7 @@ class StreamingPiCamera2(BaseCamera):
|
|||
return self._analogue_gain
|
||||
|
||||
@analogue_gain.setter
|
||||
def analogue_gain(self, value: float):
|
||||
def analogue_gain(self, value: float) -> None:
|
||||
self._analogue_gain = value
|
||||
if self.streaming:
|
||||
with self._streaming_picamera() as cam:
|
||||
|
|
@ -218,7 +226,7 @@ class StreamingPiCamera2(BaseCamera):
|
|||
return self._colour_gains
|
||||
|
||||
@colour_gains.setter
|
||||
def colour_gains(self, value: tuple[float, float]):
|
||||
def colour_gains(self, value: tuple[float, float]) -> None:
|
||||
self._colour_gains = value
|
||||
if self.streaming:
|
||||
with self._streaming_picamera() as cam:
|
||||
|
|
@ -242,7 +250,7 @@ class StreamingPiCamera2(BaseCamera):
|
|||
return self._exposure_time
|
||||
|
||||
@exposure_time.setter
|
||||
def exposure_time(self, value: int):
|
||||
def exposure_time(self, value: int) -> None:
|
||||
self._exposure_time = value
|
||||
if self.streaming:
|
||||
with self._streaming_picamera() as cam:
|
||||
|
|
@ -286,7 +294,7 @@ class StreamingPiCamera2(BaseCamera):
|
|||
return SensorModeSelector(**self._sensor_mode)
|
||||
|
||||
@sensor_mode.setter
|
||||
def sensor_mode(self, new_mode: Optional[SensorModeSelector | dict]):
|
||||
def sensor_mode(self, new_mode: Optional[SensorModeSelector | dict]) -> None:
|
||||
"""Change the sensor mode used."""
|
||||
if new_mode is None:
|
||||
self._sensor_mode = None
|
||||
|
|
@ -346,7 +354,7 @@ class StreamingPiCamera2(BaseCamera):
|
|||
) from e
|
||||
return None
|
||||
|
||||
def _initialise_picamera(self):
|
||||
def _initialise_picamera(self) -> None:
|
||||
"""Acquire the picamera device and store it as ``self._picamera``.
|
||||
|
||||
This duplicates logic in ``Picamera2.__init__`` to provide a tuning file that
|
||||
|
|
@ -378,7 +386,7 @@ class StreamingPiCamera2(BaseCamera):
|
|||
)
|
||||
self._picamera_lock = RLock()
|
||||
|
||||
def __enter__(self):
|
||||
def __enter__(self) -> None:
|
||||
"""Start streaming when the Thing context manager is opened.
|
||||
|
||||
This opens the picamera connection, initialises the camera, sets the
|
||||
|
|
@ -396,7 +404,7 @@ class StreamingPiCamera2(BaseCamera):
|
|||
return self._picamera is not None and self._picamera.started
|
||||
|
||||
@contextmanager
|
||||
def _streaming_picamera(self, pause_stream=False) -> Iterator[Picamera2]:
|
||||
def _streaming_picamera(self, pause_stream: bool = False) -> Iterator[Picamera2]:
|
||||
"""Lock access to picamera and return the underlying ``Picamera2`` instance.
|
||||
|
||||
Optionally the stream can be paused to allow updating the camera settings.
|
||||
|
|
@ -419,7 +427,12 @@ class StreamingPiCamera2(BaseCamera):
|
|||
if pause_stream and already_streaming:
|
||||
self.start_streaming()
|
||||
|
||||
def __exit__(self, exc_type, exc_value, traceback):
|
||||
def __exit__(
|
||||
self,
|
||||
_exc_type: type[BaseException],
|
||||
_exc_value: Optional[BaseException],
|
||||
_traceback: Optional[TracebackType],
|
||||
) -> None:
|
||||
"""Close the picamera connection when the Thing context manager is closed."""
|
||||
self.stop_streaming()
|
||||
with self._streaming_picamera() as cam:
|
||||
|
|
@ -635,7 +648,7 @@ class StreamingPiCamera2(BaseCamera):
|
|||
self,
|
||||
target_white_level: int = 700,
|
||||
percentile: float = 99.9,
|
||||
):
|
||||
) -> None:
|
||||
"""Adjust exposure until a the target white level is reached.
|
||||
|
||||
Starting from the minimum exposure, gradually increase exposure until
|
||||
|
|
@ -661,7 +674,7 @@ class StreamingPiCamera2(BaseCamera):
|
|||
self,
|
||||
method: Literal["percentile", "centre"] = "centre",
|
||||
luminance_power: float = 1.0,
|
||||
):
|
||||
) -> None:
|
||||
"""Correct the white balance of the image.
|
||||
|
||||
This calibration requires a neutral image, such that the 99th centile
|
||||
|
|
@ -725,7 +738,10 @@ class StreamingPiCamera2(BaseCamera):
|
|||
return tuple(recalibrate_utils.get_static_ccm(self.tuning)[0]["ccm"])
|
||||
|
||||
@colour_correction_matrix.setter # type: ignore
|
||||
def colour_correction_matrix(self, value) -> None:
|
||||
def colour_correction_matrix(
|
||||
self,
|
||||
value: tuple[float, float, float, float, float, float, float, float, float],
|
||||
) -> None:
|
||||
recalibrate_utils.set_static_ccm(self.tuning, value)
|
||||
|
||||
if self._picamera is not None:
|
||||
|
|
@ -733,7 +749,7 @@ class StreamingPiCamera2(BaseCamera):
|
|||
self._initialise_picamera()
|
||||
|
||||
@lt.thing_action
|
||||
def reset_ccm(self):
|
||||
def reset_ccm(self) -> None:
|
||||
"""Overwrite the colour correction matrix in camera tuning with default values.
|
||||
|
||||
These values are from the Raspberry Pi Camera Algorithm and Tuning Guide, page
|
||||
|
|
|
|||
|
|
@ -353,7 +353,9 @@ def upsample_channels(grids: np.ndarray, shape: tuple[int]) -> np.ndarray:
|
|||
return zoom(grids, zoom_factors, order=1)[:, : shape[0], : shape[1]]
|
||||
|
||||
|
||||
def downsampled_channels(channels: np.ndarray, blacklevel=64) -> list[np.ndarray]:
|
||||
def downsampled_channels(
|
||||
channels: np.ndarray, blacklevel: int = 64
|
||||
) -> list[np.ndarray]:
|
||||
"""Generate a downsampled, un-normalised image from which to calculate the LST.
|
||||
|
||||
TODO: blacklevel probably ought to be determined from the camera...
|
||||
|
|
|
|||
|
|
@ -8,7 +8,8 @@ See repository root for licensing information.
|
|||
|
||||
from __future__ import annotations
|
||||
import logging
|
||||
from typing import Literal, Optional
|
||||
from typing import Literal, Optional, Mapping
|
||||
from types import TracebackType
|
||||
from threading import Thread
|
||||
import time
|
||||
|
||||
|
|
@ -53,7 +54,7 @@ class SimulatedCamera(BaseCamera):
|
|||
glyph_shape: tuple[int, int, int] = (91, 91, 3),
|
||||
canvas_shape: tuple[int, int, int] = (3000, 4000, 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.
|
||||
|
|
@ -75,7 +76,7 @@ class SimulatedCamera(BaseCamera):
|
|||
self.generate_blobs()
|
||||
self.generate_canvas()
|
||||
|
||||
def generate_sprites(self):
|
||||
def generate_sprites(self) -> None:
|
||||
"""Generate sprites to populate the image."""
|
||||
sprite_sizes = [5, 7, 10, 21, 36, 40]
|
||||
self.sprites = []
|
||||
|
|
@ -108,7 +109,7 @@ class SimulatedCamera(BaseCamera):
|
|||
# Convert to uint8 and append to the list
|
||||
self.sprites.append(sprite.astype(np.uint8))
|
||||
|
||||
def generate_blobs(self, n_blobs: int = 1000):
|
||||
def generate_blobs(self, n_blobs: int = 1000) -> None:
|
||||
"""Generate coordinates of blobs and their sizes.
|
||||
|
||||
A 1000x3 array is returned. Each row represents (x,y) coordinate
|
||||
|
|
@ -124,7 +125,7 @@ class SimulatedCamera(BaseCamera):
|
|||
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):
|
||||
def generate_canvas(self) -> None:
|
||||
"""Generate a canvas.
|
||||
|
||||
Canvas is int16 so that random noise can be added to simulation image before
|
||||
|
|
@ -144,7 +145,7 @@ class SimulatedCamera(BaseCamera):
|
|||
self.canvas[self.canvas < 0] = 0
|
||||
self.canvas[self.canvas > 255] = 255
|
||||
|
||||
def generate_image(self, pos: tuple[int, int, int]):
|
||||
def generate_image(self, pos: tuple[int, int, int]) -> np.ndarray:
|
||||
"""Generate an image with blobs based on supplied coordinates."""
|
||||
canvas_width, canvas_height, _ = self.canvas_shape
|
||||
image_width, image_height, _ = self.shape
|
||||
|
|
@ -179,16 +180,16 @@ class SimulatedCamera(BaseCamera):
|
|||
|
||||
def attach_to_server(
|
||||
self, server: lt.ThingServer, path: str, setting_storage_path: str
|
||||
):
|
||||
) -> None:
|
||||
"""Wrap the attach_to_server method so the server instance can be stored.
|
||||
|
||||
Direct access to the server instance is needed to get the stage position while
|
||||
maintaining the same public API as a real camera that doesn't need this access.
|
||||
"""
|
||||
self._server = server
|
||||
return super().attach_to_server(server, path, setting_storage_path)
|
||||
super().attach_to_server(server, path, setting_storage_path)
|
||||
|
||||
def get_stage_position(self):
|
||||
def get_stage_position(self) -> Mapping[str, int]:
|
||||
"""Return the stage position.
|
||||
|
||||
The simulation camera has access to the stage position so it can generate a
|
||||
|
|
@ -198,7 +199,7 @@ class SimulatedCamera(BaseCamera):
|
|||
self._stage = self._server.things["/stage/"]
|
||||
return self._stage.instantaneous_position
|
||||
|
||||
def generate_frame(self):
|
||||
def generate_frame(self) -> np.ndarray:
|
||||
"""Generate a frame with blobs based on the stage coordinates."""
|
||||
try:
|
||||
pos = self.get_stage_position()
|
||||
|
|
@ -207,12 +208,17 @@ class SimulatedCamera(BaseCamera):
|
|||
pos = {"x": 0, "y": 0, "z": 0}
|
||||
return self.generate_image((pos["y"], pos["x"], pos["z"]))
|
||||
|
||||
def __enter__(self):
|
||||
def __enter__(self) -> None:
|
||||
"""Start the capture thread when the Thing context manager is opened."""
|
||||
self.start_streaming()
|
||||
return self
|
||||
|
||||
def __exit__(self, _exc_type, _exc_value, _traceback):
|
||||
def __exit__(
|
||||
self,
|
||||
_exc_type: type[BaseException],
|
||||
_exc_value: Optional[BaseException],
|
||||
_traceback: Optional[TracebackType],
|
||||
) -> None:
|
||||
"""Close the capture thread when the Thing context manager is closed."""
|
||||
if self.stream_active:
|
||||
self._capture_enabled = False
|
||||
|
|
@ -250,7 +256,7 @@ class SimulatedCamera(BaseCamera):
|
|||
|
||||
noise_level = lt.ThingProperty(float, 2.0)
|
||||
|
||||
def _capture_frames(self):
|
||||
def _capture_frames(self) -> None:
|
||||
portal = lt.get_blocking_portal(self)
|
||||
while self._capture_enabled:
|
||||
time.sleep(self.frame_interval)
|
||||
|
|
@ -306,14 +312,14 @@ class SimulatedCamera(BaseCamera):
|
|||
return Image.fromarray(self.generate_frame())
|
||||
|
||||
@lt.thing_action
|
||||
def remove_sample(self):
|
||||
def remove_sample(self) -> None:
|
||||
"""Show the simulated background with no sample."""
|
||||
if not self._show_sample:
|
||||
raise RuntimeError("Sample is already removed.")
|
||||
self._show_sample = False
|
||||
|
||||
@lt.thing_action
|
||||
def load_sample(self):
|
||||
def load_sample(self) -> None:
|
||||
"""Show the simulated sample."""
|
||||
if self._show_sample:
|
||||
raise RuntimeError("Sample is already in place.")
|
||||
|
|
|
|||
|
|
@ -18,6 +18,7 @@ from typing import (
|
|||
NamedTuple,
|
||||
Optional,
|
||||
Tuple,
|
||||
Mapping,
|
||||
)
|
||||
from fastapi import HTTPException
|
||||
|
||||
|
|
@ -61,7 +62,7 @@ class RecordedMove:
|
|||
for calibrating the stage as it allows measuring how long moves take.
|
||||
"""
|
||||
|
||||
def __init__(self, stage: Stage):
|
||||
def __init__(self, stage: Stage) -> None:
|
||||
"""Set the stage client used for for movement.
|
||||
|
||||
:param stage: the stage client to be used. ``stage.move_to_xyz_position`` will
|
||||
|
|
@ -71,7 +72,7 @@ class RecordedMove:
|
|||
self._current_position: Optional[CoordinateType] = None
|
||||
self._history: List[Tuple[float, Optional[CoordinateType]]] = []
|
||||
|
||||
def __call__(self, new_position: CoordinateType):
|
||||
def __call__(self, new_position: CoordinateType) -> None:
|
||||
"""Move to a new position, and record it."""
|
||||
self._history.append((time.time(), self._current_position))
|
||||
self._stage.move_to_xyz_position(xyz_pos=new_position)
|
||||
|
|
@ -85,7 +86,7 @@ class RecordedMove:
|
|||
positions: List[CoordinateType] = [p for t, p in self._history if p is not None]
|
||||
return MoveHistory(times, positions)
|
||||
|
||||
def clear_history(self):
|
||||
def clear_history(self) -> None:
|
||||
"""Reset our history to be an empty list."""
|
||||
self._history = []
|
||||
|
||||
|
|
@ -98,7 +99,7 @@ class CSMUncalibratedError(HTTPException):
|
|||
live preview to move, or when performing a scan.
|
||||
"""
|
||||
|
||||
def __init__(self):
|
||||
def __init__(self) -> None:
|
||||
"""Customise the default error code and message of HTTPException."""
|
||||
HTTPException.__init__(
|
||||
self,
|
||||
|
|
@ -241,7 +242,7 @@ class CameraStageMapper(lt.Thing):
|
|||
return None
|
||||
return self.last_calibration["image_resolution"]
|
||||
|
||||
def assert_calibrated(self):
|
||||
def assert_calibrated(self) -> None:
|
||||
"""Raise an exception if the image_to_stage_displacement matrix is not set."""
|
||||
if self.image_to_stage_displacement_matrix is None:
|
||||
# Disable check of no message in raised exception as the message is explicitly
|
||||
|
|
@ -254,7 +255,7 @@ class CameraStageMapper(lt.Thing):
|
|||
stage: Stage,
|
||||
x: float,
|
||||
y: float,
|
||||
):
|
||||
) -> None:
|
||||
"""Move by a given number of pixels on the camera.
|
||||
|
||||
NB x and y here refer to what is usually understood to be the horizontal and
|
||||
|
|
@ -273,7 +274,7 @@ class CameraStageMapper(lt.Thing):
|
|||
stage.move_relative(x=relative_move[0], y=relative_move[1])
|
||||
|
||||
@lt.thing_property
|
||||
def thing_state(self) -> dict[str, Any]:
|
||||
def thing_state(self) -> Mapping[str, Any]:
|
||||
"""Summary metadata describing the current state of the Thing."""
|
||||
return {
|
||||
k: getattr(self, k)
|
||||
|
|
|
|||
|
|
@ -7,7 +7,7 @@ It also controls external processes for live stitching composite images, and
|
|||
the creation of the final stitched images.
|
||||
"""
|
||||
|
||||
from typing import Optional
|
||||
from typing import Optional, Self, TypeVar, ParamSpec, Callable, Concatenate
|
||||
import threading
|
||||
import os
|
||||
import time
|
||||
|
|
@ -30,20 +30,34 @@ from .camera_stage_mapping import CameraStageMapper
|
|||
from .camera import CameraDependency as CameraClient
|
||||
from .stage import StageDependency as StageDep
|
||||
|
||||
T = TypeVar("T")
|
||||
P = ParamSpec("P")
|
||||
|
||||
CSMDep = lt.deps.direct_thing_client_dependency(
|
||||
CameraStageMapper, "/camera_stage_mapping/"
|
||||
)
|
||||
AutofocusDep = lt.deps.direct_thing_client_dependency(AutofocusThing, "/autofocus/")
|
||||
|
||||
JPEGBlob = lt.blob.blob_type("image/jpeg")
|
||||
ZipBlob = lt.blob.blob_type("application/zip")
|
||||
|
||||
class JPEGBlob(lt.blob.Blob):
|
||||
"""A class representing a JPEG image as a LabThings FastAPI Blob."""
|
||||
|
||||
media_type: str = "image/jpeg"
|
||||
|
||||
|
||||
class ZipBlob(lt.blob.Blob):
|
||||
"""A class representing a Zip file as a LabThings FastAPI Blob."""
|
||||
|
||||
media_type: str = "application/zip"
|
||||
|
||||
|
||||
class ScanNotRunningError(RuntimeError):
|
||||
"""Exception called when scan not running that requires a scan to be running."""
|
||||
|
||||
|
||||
def _scan_running(method):
|
||||
def _scan_running(
|
||||
method: Callable[Concatenate[Self, P], T],
|
||||
) -> Callable[Concatenate[Self, P], T]:
|
||||
"""Decorate a method so that it will error if a scan is not running.
|
||||
|
||||
This decorator is used by all methods in SmartScanThing that are using
|
||||
|
|
@ -52,7 +66,7 @@ def _scan_running(method):
|
|||
the same time and released with the lock
|
||||
"""
|
||||
|
||||
def scan_running_wrapper(self, *args, **kwargs):
|
||||
def scan_running_wrapper(self: Self, *args: P.args, **kwargs: P.kwargs) -> T:
|
||||
# Only start the method is the scan logger is set
|
||||
if self._scan_logger is not None:
|
||||
return method(self, *args, **kwargs)
|
||||
|
|
@ -71,7 +85,7 @@ class SmartScanThing(lt.Thing):
|
|||
past scans.
|
||||
"""
|
||||
|
||||
def __init__(self, scans_folder: str):
|
||||
def __init__(self, scans_folder: str) -> None:
|
||||
"""Initialise a SmartScanThing saving to and loading from the input directory.
|
||||
|
||||
:param scans_folder: This is the path to the directory where all scans will be
|
||||
|
|
@ -111,7 +125,7 @@ class SmartScanThing(lt.Thing):
|
|||
cam: CameraClient,
|
||||
csm: CSMDep,
|
||||
scan_name: str = "",
|
||||
):
|
||||
) -> None:
|
||||
"""Move the stage to cover an area, taking images that can be tiled together.
|
||||
|
||||
The stage will move in a pattern that grows outwards from the starting point,
|
||||
|
|
@ -166,7 +180,7 @@ class SmartScanThing(lt.Thing):
|
|||
self._preview_stitcher = None
|
||||
|
||||
@_scan_running
|
||||
def _check_background_and_csm_set(self):
|
||||
def _check_background_and_csm_set(self) -> None:
|
||||
"""Before starting a scan, check that background and camera-stage-mapping are set.
|
||||
|
||||
Raise error if:
|
||||
|
|
@ -306,7 +320,7 @@ class SmartScanThing(lt.Thing):
|
|||
)
|
||||
|
||||
@_scan_running
|
||||
def _save_final_scan_data(self, scan_result: str):
|
||||
def _save_final_scan_data(self, scan_result: str) -> None:
|
||||
"""Update scan data JSON file with data only known at the end of the scan.
|
||||
|
||||
Takes scan_result, a string that is either "success", "cancelled by user",
|
||||
|
|
@ -316,7 +330,7 @@ class SmartScanThing(lt.Thing):
|
|||
self._ongoing_scan.save_scan_data(self._scan_data)
|
||||
|
||||
@_scan_running
|
||||
def _manage_stitching_threads(self):
|
||||
def _manage_stitching_threads(self) -> None:
|
||||
"""Manage the stitching threads, starting them if needed and not already running."""
|
||||
# Assume 4 images means at least one offset in x and y, making the stitching
|
||||
# well constrained.
|
||||
|
|
@ -325,7 +339,7 @@ class SmartScanThing(lt.Thing):
|
|||
self._preview_stitcher.start()
|
||||
|
||||
@_scan_running
|
||||
def _run_scan(self):
|
||||
def _run_scan(self) -> None:
|
||||
"""Prepare and run the main scan, and perform final actions on completion.
|
||||
|
||||
The result (or exception) from the main scan loop determines whether the
|
||||
|
|
@ -377,7 +391,7 @@ class SmartScanThing(lt.Thing):
|
|||
self.purge_empty_scans(logger=self._scan_logger)
|
||||
|
||||
@_scan_running
|
||||
def _main_scan_loop(self):
|
||||
def _main_scan_loop(self) -> None:
|
||||
"""Run the main loop of the scan.
|
||||
|
||||
This loop runs during a scan, until no more scan x,y positions
|
||||
|
|
@ -444,7 +458,7 @@ class SmartScanThing(lt.Thing):
|
|||
self._ongoing_scan.zip_files()
|
||||
|
||||
@_scan_running
|
||||
def _return_to_starting_position(self):
|
||||
def _return_to_starting_position(self) -> None:
|
||||
"""Return to the initial scan position, if set."""
|
||||
self._scan_logger.info("Returning to starting position.")
|
||||
if self._scan_data is not None:
|
||||
|
|
@ -453,7 +467,7 @@ class SmartScanThing(lt.Thing):
|
|||
)
|
||||
|
||||
@_scan_running
|
||||
def _perform_final_stitch(self):
|
||||
def _perform_final_stitch(self) -> None:
|
||||
"""Update the scan zip and perform final stitch of the data."""
|
||||
if self._scan_data.image_count <= 3:
|
||||
self._scan_logger.info("Not performing a stitch as 3 or fewer images taken")
|
||||
|
|
@ -623,7 +637,7 @@ class SmartScanThing(lt.Thing):
|
|||
if scan_info.number_of_images == 0:
|
||||
self._delete_scan(scan_info.name, logger)
|
||||
|
||||
def _delete_scan(self, scan_name, logger: lt.deps.InvocationLogger) -> bool:
|
||||
def _delete_scan(self, scan_name: str, logger: lt.deps.InvocationLogger) -> bool:
|
||||
"""Delete a scan.
|
||||
|
||||
This is a wrapper around scan manager's delete_scan that logs to the
|
||||
|
|
@ -724,7 +738,7 @@ class SmartScanThing(lt.Thing):
|
|||
def download_zip(
|
||||
self,
|
||||
scan_name: str,
|
||||
):
|
||||
) -> ZipBlob:
|
||||
"""Return zip after including any files left until the end.
|
||||
|
||||
The zipfile is returned as a Blob.
|
||||
|
|
@ -735,7 +749,7 @@ class SmartScanThing(lt.Thing):
|
|||
@lt.thing_action
|
||||
def stitch_all_scans(
|
||||
self, logger: lt.deps.InvocationLogger, cancel: lt.deps.CancelHook
|
||||
):
|
||||
) -> None:
|
||||
"""Check the list of scans, and stitch any that don't have a DZI associated with it.
|
||||
|
||||
:raises RuntimeError: if the microscope is currently running a scan
|
||||
|
|
|
|||
|
|
@ -11,7 +11,7 @@ As the object will be used as a context manager create the hardware connection i
|
|||
|
||||
from __future__ import annotations
|
||||
from collections.abc import Sequence, Mapping
|
||||
from typing import Literal
|
||||
from typing import Literal, Never, Any
|
||||
|
||||
import labthings_fastapi as lt
|
||||
|
||||
|
|
@ -45,7 +45,7 @@ class BaseStage(lt.Thing):
|
|||
|
||||
_axis_names = ("x", "y", "z")
|
||||
|
||||
def __init__(self):
|
||||
def __init__(self) -> None:
|
||||
"""Initialise the stage.
|
||||
|
||||
:raises RedefinedBaseMovementError: if ``move_relative`` and/or
|
||||
|
|
@ -119,12 +119,12 @@ class BaseStage(lt.Thing):
|
|||
)
|
||||
|
||||
@property
|
||||
def thing_state(self):
|
||||
def thing_state(self) -> Mapping[str, Any]:
|
||||
"""Summary metadata describing the current state of the stage."""
|
||||
return {"position": self.position}
|
||||
|
||||
@lt.thing_action
|
||||
def invert_axis_direction(self, axis: Literal["x", "y", "z"]):
|
||||
def invert_axis_direction(self, axis: Literal["x", "y", "z"]) -> None:
|
||||
"""Invert the direction setting of the given axis.
|
||||
|
||||
:param axis: The axis name (x, y or z) to invert.
|
||||
|
|
@ -142,8 +142,8 @@ class BaseStage(lt.Thing):
|
|||
self,
|
||||
cancel: lt.deps.CancelHook,
|
||||
block_cancellation: bool = False,
|
||||
**kwargs: Mapping[str, int],
|
||||
):
|
||||
**kwargs: int,
|
||||
) -> None:
|
||||
"""Make a relative move. Keyword arguments should be axis names."""
|
||||
self._hardware_move_relative(
|
||||
cancel=cancel,
|
||||
|
|
@ -155,8 +155,8 @@ class BaseStage(lt.Thing):
|
|||
self,
|
||||
cancel: lt.deps.CancelHook,
|
||||
block_cancellation: bool = False,
|
||||
**kwargs: Mapping[str, int],
|
||||
):
|
||||
**kwargs: int,
|
||||
) -> Never:
|
||||
"""Make a relative move in the coordinate system used by the physical hardware.
|
||||
|
||||
Make sure to use and update ``self._hardware_position`` not ``self.position``.
|
||||
|
|
@ -170,8 +170,8 @@ class BaseStage(lt.Thing):
|
|||
self,
|
||||
cancel: lt.deps.CancelHook,
|
||||
block_cancellation: bool = False,
|
||||
**kwargs: Mapping[str, int],
|
||||
):
|
||||
**kwargs: int,
|
||||
) -> None:
|
||||
"""Make an absolute move. Keyword arguments should be axis names."""
|
||||
self._hardware_move_absolute(
|
||||
cancel=cancel,
|
||||
|
|
@ -183,8 +183,8 @@ class BaseStage(lt.Thing):
|
|||
self,
|
||||
cancel: lt.deps.CancelHook,
|
||||
block_cancellation: bool = False,
|
||||
**kwargs: Mapping[str, int],
|
||||
):
|
||||
**kwargs: int,
|
||||
) -> Never:
|
||||
"""Make a absolute move in the coordinate system used by the physical hardware.
|
||||
|
||||
Make sure to use and update ``self._hardware_position`` not ``self.position``.
|
||||
|
|
@ -194,7 +194,7 @@ class BaseStage(lt.Thing):
|
|||
)
|
||||
|
||||
@lt.thing_action
|
||||
def set_zero_position(self):
|
||||
def set_zero_position(self) -> None:
|
||||
"""Make the current position zero in all axes.
|
||||
|
||||
This action does not move the stage, but resets the position to zero.
|
||||
|
|
|
|||
|
|
@ -2,6 +2,8 @@
|
|||
|
||||
from __future__ import annotations
|
||||
|
||||
from typing import Optional, Any
|
||||
from types import TracebackType
|
||||
from collections.abc import Mapping
|
||||
import time
|
||||
|
||||
|
|
@ -17,7 +19,7 @@ class DummyStage(BaseStage):
|
|||
hardware attached.
|
||||
"""
|
||||
|
||||
def __init__(self, step_time: float = 0.001, **kwargs):
|
||||
def __init__(self, step_time: float = 0.001, **kwargs: Any) -> None:
|
||||
"""Initialise the Dummy stage, setting the step_time to adjust the speed.
|
||||
|
||||
:param step_time: The time in seconds per "motor" step. The default of 0.001
|
||||
|
|
@ -29,11 +31,16 @@ class DummyStage(BaseStage):
|
|||
self.step_time = step_time
|
||||
self.instantaneous_position = self._hardware_position
|
||||
|
||||
def __enter__(self):
|
||||
def __enter__(self) -> None:
|
||||
"""Register the stage position when the Thing context manager is opened."""
|
||||
self.instantaneous_position = self._hardware_position
|
||||
|
||||
def __exit__(self, _exc_type, _exc_value, _traceback):
|
||||
def __exit__(
|
||||
self,
|
||||
_exc_type: type[BaseException],
|
||||
_exc_value: Optional[BaseException],
|
||||
_traceback: Optional[TracebackType],
|
||||
) -> None:
|
||||
"""Nothing to do when the Thing context manager is closed."""
|
||||
|
||||
axis_inverted = lt.ThingSetting(
|
||||
|
|
@ -47,8 +54,8 @@ class DummyStage(BaseStage):
|
|||
self,
|
||||
cancel: lt.deps.CancelHook,
|
||||
block_cancellation: bool = False,
|
||||
**kwargs: Mapping[str, int],
|
||||
):
|
||||
**kwargs: int,
|
||||
) -> None:
|
||||
"""Make a relative move. Keyword arguments should be axis names."""
|
||||
displacement = [kwargs.get(k, 0) for k in self.axis_names]
|
||||
self.moving = True
|
||||
|
|
@ -85,8 +92,8 @@ class DummyStage(BaseStage):
|
|||
self,
|
||||
cancel: lt.deps.CancelHook,
|
||||
block_cancellation: bool = False,
|
||||
**kwargs: Mapping[str, int],
|
||||
):
|
||||
**kwargs: int,
|
||||
) -> None:
|
||||
"""Make an absolute move. Keyword arguments should be axis names."""
|
||||
displacement = {
|
||||
axis: int(pos) - self._hardware_position[axis]
|
||||
|
|
@ -98,7 +105,7 @@ class DummyStage(BaseStage):
|
|||
)
|
||||
|
||||
@lt.thing_action
|
||||
def set_zero_position(self):
|
||||
def set_zero_position(self) -> None:
|
||||
"""Make the current position zero in all axes.
|
||||
|
||||
This action does not move the stage, but resets the position to zero.
|
||||
|
|
|
|||
|
|
@ -5,7 +5,8 @@ import logging
|
|||
import threading
|
||||
import time
|
||||
from copy import copy
|
||||
from typing import Iterator, Literal
|
||||
from typing import Iterator, Literal, Optional, Any
|
||||
from types import TracebackType
|
||||
from contextlib import contextmanager
|
||||
from collections.abc import Mapping
|
||||
|
||||
|
|
@ -24,7 +25,7 @@ class SangaboardThing(BaseStage):
|
|||
functionality is accessed by directly querying the serial interface.
|
||||
"""
|
||||
|
||||
def __init__(self, port: str = None, **kwargs):
|
||||
def __init__(self, port: str = None, **kwargs: Any) -> None:
|
||||
"""Initialise SangaboardThing.
|
||||
|
||||
Initialise the "Thing", but do not initialise an underlying
|
||||
|
|
@ -41,7 +42,7 @@ class SangaboardThing(BaseStage):
|
|||
self.sangaboard_kwargs["port"] = port
|
||||
super().__init__(**kwargs)
|
||||
|
||||
def __enter__(self):
|
||||
def __enter__(self) -> None:
|
||||
"""Connect to the sangaboard when the Thing context manager is opened."""
|
||||
self._sangaboard = sangaboard.Sangaboard(**self.sangaboard_kwargs)
|
||||
self._sangaboard_lock = threading.RLock()
|
||||
|
|
@ -53,7 +54,12 @@ class SangaboardThing(BaseStage):
|
|||
sb.query("blocking_moves false")
|
||||
self.update_position()
|
||||
|
||||
def __exit__(self, _exc_type, _exc_value, _traceback):
|
||||
def __exit__(
|
||||
self,
|
||||
_exc_type: type[BaseException],
|
||||
_exc_value: Optional[BaseException],
|
||||
_traceback: Optional[TracebackType],
|
||||
) -> None:
|
||||
"""Close the sangaboard connection when the Thing context manager is closed."""
|
||||
with self.sangaboard() as sb:
|
||||
sb.close()
|
||||
|
|
@ -85,7 +91,7 @@ class SangaboardThing(BaseStage):
|
|||
self,
|
||||
cancel: lt.deps.CancelHook,
|
||||
block_cancellation: bool = False,
|
||||
**kwargs: Mapping[str, int],
|
||||
**kwargs: int,
|
||||
) -> None:
|
||||
"""Make a relative move in the coordinate system used by the sangaboard."""
|
||||
displacement = [kwargs.get(axis, 0) for axis in self.axis_names]
|
||||
|
|
@ -112,7 +118,7 @@ class SangaboardThing(BaseStage):
|
|||
self,
|
||||
cancel: lt.deps.CancelHook,
|
||||
block_cancellation: bool = False,
|
||||
**kwargs: Mapping[str, int],
|
||||
**kwargs: int,
|
||||
) -> None:
|
||||
"""Make a absolute move in the coordinate system used by the sangaboard."""
|
||||
with self.sangaboard():
|
||||
|
|
|
|||
|
|
@ -6,7 +6,7 @@ the microscope, server, and thing states to the web API.
|
|||
|
||||
from collections.abc import Mapping
|
||||
import socket
|
||||
from typing import Optional
|
||||
from typing import Optional, Any
|
||||
from uuid import UUID, uuid4
|
||||
import subprocess
|
||||
import os
|
||||
|
|
@ -51,7 +51,7 @@ class OpenFlexureSystem(lt.Thing):
|
|||
return UUID(self._microscope_id)
|
||||
|
||||
@microscope_id.setter
|
||||
def microscope_id(self, uuid: UUID):
|
||||
def microscope_id(self, uuid: UUID) -> None:
|
||||
# TODO make read only but still settable from disk
|
||||
self._microscope_id = uuid
|
||||
|
||||
|
|
@ -127,7 +127,7 @@ class OpenFlexureSystem(lt.Thing):
|
|||
return metadata_getter()
|
||||
|
||||
@property
|
||||
def thing_state(self) -> Mapping:
|
||||
def thing_state(self) -> Mapping[str, Any]:
|
||||
"""Summary metadata describing the current state of the Thing."""
|
||||
return {
|
||||
"hostname": self.hostname,
|
||||
|
|
|
|||
|
|
@ -1,7 +1,11 @@
|
|||
"""Functionality for communicating the required user interface for a thing."""
|
||||
|
||||
from typing import Callable, Any
|
||||
|
||||
from pydantic import BaseModel
|
||||
|
||||
import labthings_fastapi as lt
|
||||
|
||||
|
||||
class ActionButton(BaseModel):
|
||||
"""The data required for creating an actionButton in Vue.
|
||||
|
|
@ -43,7 +47,7 @@ class ActionButton(BaseModel):
|
|||
"""The message to show on successful completion."""
|
||||
|
||||
|
||||
def action_button_for(action, **kwargs) -> ActionButton:
|
||||
def action_button_for(action: Callable[..., Any], **kwargs: Any) -> ActionButton:
|
||||
"""Create a ActionButton data for the specified Thing Action.
|
||||
|
||||
:param action: The thing action to create a button for.
|
||||
|
|
@ -80,7 +84,9 @@ class PropertyControl(BaseModel):
|
|||
"""The delay in ms before reading back the property."""
|
||||
|
||||
|
||||
def property_control_for(thing, property_name, **kwargs) -> PropertyControl:
|
||||
def property_control_for(
|
||||
thing: lt.Thing, property_name: str, **kwargs: Any
|
||||
) -> PropertyControl:
|
||||
"""Create an PropertyControl data for the specified Thing Property.
|
||||
|
||||
:param thing: The instance of the thing that has the property to be controlled.
|
||||
|
|
|
|||
|
|
@ -1,6 +1,6 @@
|
|||
"""Utility functions and classes."""
|
||||
|
||||
from typing import TypeVar, Callable, ParamSpec
|
||||
from typing import TypeVar, Callable, ParamSpec, Optional, Any, Concatenate, Self
|
||||
import os
|
||||
import re
|
||||
import sys
|
||||
|
|
@ -24,7 +24,7 @@ COMMIT_REGEX = re.compile(r"[0-9a-f]{40}")
|
|||
REF_REGEX = re.compile(r"^ref:\s(.*)$")
|
||||
|
||||
|
||||
def _is_lock_like(obj):
|
||||
def _is_lock_like(obj: Any) -> bool:
|
||||
"""Check if an object is a lock.
|
||||
|
||||
Cannot use ``isinstance(obj, threading.RLock)``, may be possible to use
|
||||
|
|
@ -34,14 +34,16 @@ def _is_lock_like(obj):
|
|||
return hasattr(obj, "acquire") and hasattr(obj, "release")
|
||||
|
||||
|
||||
def requires_lock(method: Callable[P, T]) -> Callable[P, T]:
|
||||
def requires_lock(
|
||||
method: Callable[Concatenate[Self, P], T],
|
||||
) -> Callable[Concatenate[Self, P], T]:
|
||||
"""Decorate a class method so that it requires the class lock to run.
|
||||
|
||||
The class should have a reentrant lock with the name ``self._lock``.
|
||||
"""
|
||||
|
||||
@wraps(method)
|
||||
def wrapper(self, *args: P.args, **kwargs: P.kwargs) -> T:
|
||||
def wrapper(self: Self, *args: P.args, **kwargs: P.kwargs) -> T:
|
||||
# Confirm the object the method is attached to has a self._lock property
|
||||
if not hasattr(self, "_lock"):
|
||||
raise AttributeError(f"{self.__class__.__name__} has no '_lock' attribute")
|
||||
|
|
@ -67,7 +69,16 @@ class ErrorCapturingThread(Thread):
|
|||
join is run.
|
||||
"""
|
||||
|
||||
def __init__(self, group=None, target=None, args=None, kwargs=None, daemon=None):
|
||||
def __init__(
|
||||
self,
|
||||
group: Optional[Any] = None,
|
||||
target: Optional[Callable[..., Any]] = None,
|
||||
name: Optional[str] = None,
|
||||
args: tuple[Any, ...] = (),
|
||||
kwargs: Optional[dict[str, Any]] = None,
|
||||
*,
|
||||
daemon: Optional[bool] = None,
|
||||
) -> None:
|
||||
"""Initialise with the same arguments as Thread."""
|
||||
# As all inputs are keywords we need to set the default values for args and kwargs:
|
||||
if args is None:
|
||||
|
|
@ -88,12 +99,13 @@ class ErrorCapturingThread(Thread):
|
|||
super().__init__(
|
||||
group=group,
|
||||
target=_wrap_and_catch_errors,
|
||||
name=name,
|
||||
args=args,
|
||||
kwargs=kwargs,
|
||||
daemon=daemon,
|
||||
)
|
||||
|
||||
def join(self, timeout=None):
|
||||
def join(self, timeout: Optional[float] = None) -> None:
|
||||
"""Join when the thread is complete.
|
||||
|
||||
If the thread ended due to an unhandled exception, the exception will be raised
|
||||
|
|
@ -107,7 +119,9 @@ class ErrorCapturingThread(Thread):
|
|||
raise err
|
||||
|
||||
|
||||
def _wrap_and_catch_errors(target, error_buffer, *args, **kwargs):
|
||||
def _wrap_and_catch_errors(
|
||||
target: Callable[..., Any], error_buffer: list, *args: Any, **kwargs: Any
|
||||
) -> None:
|
||||
"""Run target function in a try-except block.
|
||||
|
||||
This function is designed only to be used by ErrorCapturingThread.
|
||||
|
|
|
|||
|
|
@ -1,6 +1,5 @@
|
|||
"""Test the stage without creating a full HTTP server and socket connection."""
|
||||
|
||||
from collections.abc import Mapping
|
||||
import tempfile
|
||||
import itertools
|
||||
|
||||
|
|
@ -69,7 +68,7 @@ def test_override_base_movement():
|
|||
self,
|
||||
cancel: lt.deps.CancelHook,
|
||||
block_cancellation: bool = False,
|
||||
**kwargs: Mapping[str, int],
|
||||
**kwargs: int,
|
||||
):
|
||||
pass
|
||||
|
||||
|
|
@ -82,7 +81,7 @@ def test_override_base_movement():
|
|||
self,
|
||||
cancel: lt.deps.CancelHook,
|
||||
block_cancellation: bool = False,
|
||||
**kwargs: Mapping[str, int],
|
||||
**kwargs: int,
|
||||
):
|
||||
pass
|
||||
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue