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

Merge branch 'v3' into labthings-debug-logging-config

Browse source
This commit is contained in:
Beth Probert 2026-04-02 10:50:06 +01:00
commit 01954167ab
72 changed files with 2722 additions and 1046 deletions
Download patch file Download diff file Expand all files Collapse all files

3
src/openflexure_microscope_server/scan_directories.py
View file

@ -353,6 +353,9 @@ class ScanDirectoryManager:
For more explanation on the scan naming see `new_scan_dir`
"""
scan_name = make_name_safe(scan_name)
# Strip all trailing underscores from the base name
scan_name = scan_name.strip("_")
# A regex with the scan name and a group for the numbers
scan_regex = re.compile(
"^" + scan_name + "_([0-9]{" + str(SCAN_ZERO_PAD_DIGITS) + "})$"

40
src/openflexure_microscope_server/stitching.py
View file

@ -18,7 +18,7 @@ from pydantic import BaseModel
import labthings_fastapi as lt
from openflexure_microscope_server.utilities import make_path_safe
from openflexure_microscope_server.utilities import is_path_safe
IS_WINDOWS = os.name == "nt"
@ -29,6 +29,22 @@ STITCH_TILE_SIZE = 8192
DEFAULT_OVERLAP = 0.1
DEFAULT_RESIZE = 0.5
# A list of commands that are forbidden in any part of a generated CLI command.
# This provides defense-in-depth against trying to execute arbitrary shells
# or elevation tools.
FORBIDDEN_COMMANDS = {
"sudo",
"sh",
"bash",
"perl",
"ruby",
"php",
"nc",
"netcat",
"curl",
"wget",
}
class StitchingSettings(BaseModel):
"""The data needed to stitch a scan."""
@ -49,17 +65,26 @@ class StitcherValidationError(RuntimeError):
def validate_command(cmd: list[str]) -> None:
"""Validate that the command only characters that are allowed in a path.
"""Validate that the command only contains characters that are allowed in a path.
The values in the commands should be numbers, commandline flags, paths, and
executables. All of these should be allowed by ``make_path_safe``.
executables. All of these should be allowed by ``is_path_safe``.
This also checks against a blacklist of forbidden commands for defense-in-depth.
:raises StitcherValidationError: if any element in the command is not safe.
"""
for element in cmd:
if element != make_path_safe(element):
# Check against forbidden commands (case-insensitive)
if element.lower() in FORBIDDEN_COMMANDS:
raise StitcherValidationError(
"Invalid stiching command: Contains unsafe characters."
f"Invalid stitching command: Forbidden element '{element}' detected."
)
# Ensure characters are safe for a path component
if not is_path_safe(element):
raise StitcherValidationError(
f"Invalid stitching command: Element '{element}' contains unsafe characters."
)
@ -129,7 +154,7 @@ class BaseStitcher:
:raises RuntimeError: if inputs are unsafe.
"""
if self.images_dir != make_path_safe(self.images_dir):
if not is_path_safe(self.images_dir):
raise StitcherValidationError(
"Invalid directory path: Contains unsafe characters."
)
@ -200,7 +225,8 @@ class PreviewStitcher(BaseStitcher):
# Windows has no SIGKILL
self._popen_obj.kill()
else:
self._popen_obj.send_signal(signal.SIGKILL)
# ignore this line in mypy as mypy doesn't understand using the bool as a check
self._popen_obj.send_signal(signal.SIGKILL) # type: ignore[attr-defined]
raise (e)

6
src/openflexure_microscope_server/things/__init__.py
View file

@ -4,7 +4,7 @@ The microscope can be extended to be used with other hardware by creating a pack
with other Things and including them in the LabThings-FastAPI config file.
"""
import posixpath
import os
from typing import Optional, Self
import labthings_fastapi as lt
@ -23,7 +23,9 @@ class OFMThing(lt.Thing):
if application_config is None:
raise ValueError("No application configuration was supplied.")
app_data_dir = application_config["data_folder"]
self._data_dir = posixpath.join(str(app_data_dir), self.path.strip("/"))
self._data_dir = os.path.join(
os.path.normpath(str(app_data_dir)), os.path.normpath(self.path.strip("/"))
)
return self
@property

18
src/openflexure_microscope_server/things/autofocus.py
View file

@ -67,11 +67,6 @@ class StackParams(BaseModel):
settling_time: float = Field(default=0.3, ge=0)
"""Time (in seconds) between moving and capturing an image"""
backlash_correction: int = 250
"""
Distance (in steps) to overshoot a move and then undo, to account for backlash
"""
origin: StackOrigin = StackOrigin.START
"""Where the stack is positioned relative to the current z position."""
@ -760,7 +755,7 @@ class AutofocusThing(lt.Thing):
logic, or autofocus is performed.
:param stack_parameters: StackParams defining stack spacing,
image count and backlash correction.
and image count.
:param capture_parameters: CaptureParams defining save
resolution, images directory.
@ -780,15 +775,8 @@ class AutofocusThing(lt.Thing):
target_offset = -total_range
else:
target_offset = 0
# Apply backlash correction: overshoot and move back
overshoot = target_offset - stack_parameters.backlash_correction
if overshoot != 0:
self._stage.move_relative(z=overshoot)
# Move back to starting point if needed
if stack_parameters.backlash_correction != 0:
self._stage.move_relative(z=stack_parameters.backlash_correction)
if target_offset != 0:
self._stage.move_relative(z=target_offset)
# Capture images_to_save images
for move_count in range(stack_parameters.images_to_save):

64
src/openflexure_microscope_server/things/background_detect.py
View file

@ -13,7 +13,11 @@ from pydantic import BaseModel
import labthings_fastapi as lt
from openflexure_microscope_server.ui import PropertyControl, property_control_for
from openflexure_microscope_server.ui import (
UI_ELEMENT_RESPONSE,
UIElementList,
property_control_for,
)
SettingsType = TypeVar("SettingsType", bound=BaseModel)
BackgroundType = TypeVar("BackgroundType", bound=BaseModel)
@ -72,9 +76,9 @@ class BackgroundDetectAlgorithm(lt.Thing):
"Each background detect algorithm must implement an set_background method."
)
@lt.property
def settings_ui(self) -> list[PropertyControl]:
"""A list of PropertyControl objects to create the settings in the UI."""
@lt.endpoint("get", "settings_ui", responses=UI_ELEMENT_RESPONSE)
def settings_ui(self) -> UIElementList:
"""Return the UI for the background detect settings."""
raise NotImplementedError(
"Each background detect algorithm must implement an settings_ui method."
)
@ -99,29 +103,33 @@ class ColourChannelDetectLUV(BackgroundDetectAlgorithm):
# provided there.
min_stds = [0.5, 0.3, 0.5]
channel_tolerance: float = lt.setting(default=7.0)
channel_tolerance: float = lt.setting(default=7.0, ge=0)
"""Channel Tolerance
The number of standard deviations a pixel value must be from the background mean
to be considered sample.
"""
min_sample_coverage: float = lt.setting(default=25)
min_sample_coverage: float = lt.setting(default=25, ge=0, le=100)
"""Sample Coverage Required (%)
The minimum percentage of the image that needs to be identified as sample for the
image to be labeled as containing sample.
"""
@lt.property
def settings_ui(self) -> list[PropertyControl]:
"""A list of PropertyControl objects to create the settings in the UI."""
return [
property_control_for(self, "channel_tolerance", label="Channel Tolerance"),
property_control_for(
self, "min_sample_coverage", label="Sample Coverage Required (%)"
),
]
@lt.endpoint("get", "settings_ui", responses=UI_ELEMENT_RESPONSE)
def settings_ui(self) -> UIElementList:
"""Return the UI for the background detect settings."""
return UIElementList(
[
property_control_for(
self, "channel_tolerance", label="Channel Tolerance"
),
property_control_for(
self, "min_sample_coverage", label="Sample Coverage Required (%)"
),
]
)
@lt.property
def ready(self) -> bool:
@ -223,29 +231,33 @@ class ChannelDeviationLUV(BackgroundDetectAlgorithm):
# LUV colour space not converting to the CIELUV numbers)
min_stds = [0.5, 0.3, 0.5]
channel_tolerance: float = lt.setting(default=7.0)
channel_tolerance: float = lt.setting(default=7.0, ge=0)
"""Channel Tolerance
The number of standard deviations a pixel value must be from the background mean
to be considered sample.
"""
min_sample_coverage: float = lt.setting(default=25)
min_sample_coverage: float = lt.setting(default=25, ge=0, le=100)
"""Sample Coverage Required (%)
The minimum percentage of the image that needs to be identified as sample for the
image to be labeled as containing sample.
"""
@lt.property
def settings_ui(self) -> list[PropertyControl]:
"""A list of PropertyControl objects to create the settings in the UI."""
return [
property_control_for(self, "channel_tolerance", label="Channel Tolerance"),
property_control_for(
self, "min_sample_coverage", label="Sample Coverage Required (%)"
),
]
@lt.endpoint("get", "settings_ui", responses=UI_ELEMENT_RESPONSE)
def settings_ui(self) -> UIElementList:
"""Return the UI for the background detect settings."""
return UIElementList(
[
property_control_for(
self, "channel_tolerance", label="Channel Tolerance"
),
property_control_for(
self, "min_sample_coverage", label="Sample Coverage Required (%)"
),
]
)
@lt.property
def ready(self) -> bool:

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

@ -274,7 +274,7 @@ class BaseCamera(lt.Thing):
"CameraThings must define their own capture_array method"
)
downsampled_array_factor: int = lt.property(default=2)
downsampled_array_factor: int = lt.property(default=2, ge=1)
"""The downsampling factor when calling capture_downsampled_array."""
@lt.action
@ -565,7 +565,7 @@ class BaseCamera(lt.Thing):
except Exception as e:
raise IOError(f"An error occurred while saving {jpeg_path}") from e
settling_time: float = lt.setting(default=0.2)
settling_time: float = lt.setting(default=0.2, ge=0)
"""The settling time when calling the ``settle()`` method."""
@lt.action

3
src/openflexure_microscope_server/things/camera/picamera.py
View file

@ -640,7 +640,8 @@ class StreamingPiCamera2(BaseCamera):
:param target_white_level: Raw target white level, this should be an integer
within the range set by the bit-depth of the camera sensor (10-bit for
PiCamera v2, 12 Bit for Picamera HQ. If None the default will be used for
the current sensor. This is approximately 70% saturated.
the current sensor. This is approximately 40% saturated, but after gamma
curve is applied, the pixel values will have a value around 200.
:param percentile: The percentile to use instead of maximum. Default 99.9. When
calculating the brightest pixel, a percentile is used rather than the
maximum in order to be robust to a small number of noisy/bright pixels.

25
src/openflexure_microscope_server/things/camera/picamera_recalibrate_utils.py
View file

@ -85,7 +85,7 @@ IMX219_SENSOR_INFO = SensorInfo(
unpacked_pixel_format="SBGGR10",
bit_depth=10,
blacklevel=64,
default_target_white_level=700,
default_target_white_level=400,
short_pause=0.2,
long_pause=0.5,
)
@ -95,7 +95,7 @@ IMX477_SENSOR_INFO = SensorInfo(
unpacked_pixel_format="SBGGR12",
bit_depth=12,
blacklevel=256,
default_target_white_level=2800,
default_target_white_level=1600,
short_pause=0.2,
long_pause=1.0,
)
@ -120,10 +120,11 @@ def adjust_shutter_and_gain_from_raw(
:param camera: A Picamera2 object.
:param target_white_level: The raw value we aim for, the raw value of the brightest
pixels should be approximately this bright. The value to set depends on the
sensor bit depth. We recommend values of 700 for 10-bit sensors and 2800 for
12-bit sensors. This is about 70% of saturated once the blacklevel is
sensor bit depth. We recommend values of 400 for 10-bit sensors and 1600 for
12-bit sensors. This is about 40% of saturated once the blacklevel is
subtracted. The maximum possible value depends on the sensor bit depth, the
sensor blacklevel and the tolerance argument.
sensor blacklevel and the tolerance argument. While this only uses 40% of the
sensor range, after gamma this corresponds to pixel value ~200.
:param max_iterations: We will terminate once we perform this many iterations,
whether or not we converge. More than 10 shouldn't happen.
:param tolerance: How close to the target value we consider "done". Expressed as a
@ -240,6 +241,9 @@ def _set_minimum_exposure(camera: Picamera2, sensor_info: SensorInfo) -> None:
# to the minimum possible, which is ~8us for PiCamera v2
camera.set_controls({"AeEnable": False, "AnalogueGain": 1, "ExposureTime": 1})
time.sleep(sensor_info.long_pause)
# Flush stale frames
for _ in range(2):
_r = camera.capture_metadata()
def _test_exposure_settings(camera: Picamera2, percentile: float) -> _ExposureTest:
@ -252,9 +256,15 @@ def _test_exposure_settings(camera: Picamera2, percentile: float) -> _ExposureTe
percentile (which will be compared to the target), as well as
the camera's shutter and gain values.
"""
camera.capture_array("raw") # controls might not be updated for the first frame?
# A single request, to ensure metadata matches frame
request = camera.capture_request()
try:
metadata = request.get_metadata()
image = request.make_array("raw")
finally:
request.release()
max_brightness = np.percentile(
_channels_from_bayer_array(camera.capture_array("raw")),
_channels_from_bayer_array(image),
percentile,
)
# The reported brightness can, theoretically, be negative or zero
@ -267,7 +277,6 @@ def _test_exposure_settings(camera: Picamera2, percentile: float) -> _ExposureTe
"camera's black level compensation has gone wrong."
)
max_brightness = 1
metadata = camera.capture_metadata()
result = _ExposureTest(
level=max_brightness,
exposure_time=int(metadata["ExposureTime"]),

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

@ -172,6 +172,8 @@ class SimulatedCamera(BaseCamera):
@blob_density.setter
def _set_blob_density(self, value: int) -> None:
if value < 0:
raise ValueError("Sample density must be >= 0")
self._blob_density = value
if self._capture_enabled:
self.generate_canvas()
@ -457,7 +459,7 @@ class SimulatedCamera(BaseCamera):
return self._capture_thread.is_alive()
return False
noise_level: float = lt.property(default=2.0)
noise_level: float = lt.property(default=2.0, ge=0, le=50)
def _capture_frames(self) -> None:
last_frame_t = time.time()

586
src/openflexure_microscope_server/things/scan_workflows.py
View file

@ -4,14 +4,13 @@ This module contains the base ``ScanWorkflow`` class that all workflows should s
as well as specific workflows.
"""
from __future__ import annotations
import os
from typing import (
Generic,
Literal,
Mapping,
Optional,
Protocol,
TypeVar,
)
@ -34,6 +33,7 @@ from openflexure_microscope_server.things.autofocus import (
AutofocusParams,
AutofocusThing,
SmartStackParams,
StackParams,
)
from openflexure_microscope_server.things.background_detect import (
ChannelDeviationLUV,
@ -41,7 +41,16 @@ from openflexure_microscope_server.things.background_detect import (
from openflexure_microscope_server.things.camera import BaseCamera, CaptureParams
from openflexure_microscope_server.things.camera_stage_mapping import CameraStageMapper
from openflexure_microscope_server.things.stage import BaseStage
from openflexure_microscope_server.ui import PropertyControl, property_control_for
from openflexure_microscope_server.ui import (
UI_ELEMENT_RESPONSE,
Accordion,
HeaderBlock,
PropertyControl,
TextBlock,
UIElementList,
action_button_for,
property_control_for,
)
SettingModelType = TypeVar("SettingModelType", bound=BaseModel)
@ -160,9 +169,9 @@ class ScanWorkflow(Generic[SettingModelType], lt.Thing):
return True, focus_height
@lt.property
def settings_ui(self) -> list[PropertyControl]:
"""A list of PropertyControl objects to create the settings in the scan tab."""
@lt.endpoint("get", "settings_ui", responses=UI_ELEMENT_RESPONSE)
def settings_ui(self) -> UIElementList:
"""Return the UI for the workflow's settings in the scan tab."""
raise NotImplementedError(
"Each scan workflow must implement a settings_ui method."
)
@ -297,6 +306,145 @@ class RectGridWorkflow(
return not self._csm.calibration_required
class SmartStackCompatibleSettings(Protocol):
"""A protocol for the minimum settings needed for smart stack to work."""
capture_params: CaptureParams
autofocus_params: AutofocusParams
smart_stack_params: SmartStackParams
class SmartStackMixin:
"""A mixin for scan workflows that use smart stacking."""
stack_images_to_save: int = lt.setting(default=1, ge=1, le=9)
"""The number of images to save in a stack.
Defaults to 1 unless you need to see either side of focus
"""
stack_min_images_to_test: int = lt.setting(
default=9, ge=MIN_TEST_IMAGE_COUNT, le=MAX_TEST_IMAGE_COUNT
)
"""The minimum number of images to capture in a stack.
This many images are captured and tested for focus, if the focus is not central
enough more images may be captured. After new images are captured, this value sets
the number of images used for checking if focus is achieved.
Defaults to 9 which balances reliability and speed.
"""
stack_dz: int = lt.setting(default=50, ge=10, le=400)
"""Distance in steps between images in a z-stack.
Suggested values:
* 50 for 60-100x
* 100 for 40x
* 200 for 20x
"""
@property
def as_workflow(self) -> ScanWorkflow:
"""Return self as a ScanWorkflow.
Ensures this mixin is only used with ScanWorkflow instances,
raising TypeError otherwise.
"""
if not isinstance(self, ScanWorkflow):
raise TypeError("SmartStackMixin must be mixed into a ScanWorkflow")
return self
def create_smart_stack_params(self, save_on_failure: bool) -> SmartStackParams:
"""Set up the parameters used for all smart stacks in a scan.
:returns: A StackSmartParams object with the required parameters.
"""
# Coerce min_images_to_test parameter
min_images_to_test = self.stack_min_images_to_test
if min_images_to_test % 2 == 0:
min_images_to_test += 1
self.as_workflow.logger.warning(
"Minimum number of images to test should be odd, setting to "
f"{min_images_to_test}."
)
# Set the Thing property to the coerced value
self.stack_min_images_to_test = min_images_to_test
# Coerce the images to save parameter to be odd, and less than
# min_images_to_save
images_to_save = self.stack_images_to_save
if images_to_save > min_images_to_test:
self.as_workflow.logger.warning(
f"Cannot save {images_to_save} images as this above the minimum "
f"number to test. Setting images to save to {MAX_TEST_IMAGE_COUNT}."
)
images_to_save = min_images_to_test
elif images_to_save % 2 == 0:
images_to_save += 1
self.as_workflow.logger.warning(
f"Images to save should be odd, setting to {images_to_save}."
)
# Set the Thing property to the coerced value
self.stack_images_to_save = images_to_save
return SmartStackParams(
stack_dz=self.stack_dz,
images_to_save=self.stack_images_to_save,
min_images_to_test=self.stack_min_images_to_test,
save_on_failure=save_on_failure,
)
def _perform_smart_stack(
self, settings: SmartStackCompatibleSettings, xyz_pos: tuple[int, int, int]
) -> tuple[bool, Optional[int]]:
"""Perform acquisition a smart stack.
:param settings: The settings for this scan as a HistoScanSettingsModel
:param xyz_pos: The current position as a tuple or 3 ints.
:return: A tuple of whether an image was taken, and the z-position for focus.
If failed to find focus, returns for the focus z-position.
"""
focus_height: Optional[int]
focused, focus_height = self.as_workflow._autofocus.run_smart_stack(
stack_parameters=settings.smart_stack_params,
capture_parameters=settings.capture_params,
autofocus_parameters=settings.autofocus_params,
)
# An image was captured if we are focussed or we are not skipping background.
imaged = focused or settings.smart_stack_params.save_on_failure
if not imaged:
msg = f"Stack failed at {xyz_pos}. Treating as background."
self.as_workflow.logger.info(msg)
# run_smart_stage always returns a focus height for the sharpest image even
# if it failed to find a good focus. Set to None if not focussed.
if not focused:
focus_height = None
return imaged, focus_height
def smart_stack_property_controls(self) -> list[PropertyControl]:
"""Return smart stack property controls for the UI."""
return [
property_control_for(
self.as_workflow,
"stack_images_to_save",
label="Images in Stack to Save",
),
property_control_for(
self.as_workflow,
"stack_min_images_to_test",
label="Minimum number of images to test for focus",
),
property_control_for(
self.as_workflow, "stack_dz", label="Stack dz (steps)", step=5
),
]
class HistoScanSettingsModel(RectGridSettingsModel):
"""The settings for a scan with the HistoScanWorkflow.
@ -309,7 +457,7 @@ class HistoScanSettingsModel(RectGridSettingsModel):
smart_stack_params: SmartStackParams
class HistoScanWorkflow(RectGridWorkflow[HistoScanSettingsModel]):
class HistoScanWorkflow(RectGridWorkflow[HistoScanSettingsModel], SmartStackMixin):
"""A workflow optimised for scanning Histopathology samples.
This workflow automatically plans its own path around a sample spiralling out from
@ -342,36 +490,6 @@ class HistoScanWorkflow(RectGridWorkflow[HistoScanSettingsModel]):
max_range: int = lt.setting(default=45000, ge=0)
"""The maximum distance in steps from the centre of the scan."""
# Stacking settings
stack_images_to_save: int = lt.setting(default=1, ge=1, le=9)
"""The number of images to save in a stack.
Defaults to 1 unless you need to see either side of focus
"""
stack_min_images_to_test: int = lt.setting(
default=9, ge=MIN_TEST_IMAGE_COUNT, le=MAX_TEST_IMAGE_COUNT
)
"""The minimum number of images to capture in a stack.
This many images are captures and tested for focus, if the focus is not central
enough more images may be captured. After new images are captured, this value sets
the number of images used for checking if focus is achieved.
Defaults to 9 which balances reliability and speed.
"""
stack_dz: int = lt.setting(default=50, ge=10, le=400)
"""Distance in steps between images in a z-stack.
Suggested values:
* 50 for 60-100x
* 100 for 40x
* 200 for 20x
"""
equal_distances: bool = lt.setting(default=False)
"""Make the distances in x and y equal in motor steps, rather than in overlap.
@ -435,49 +553,9 @@ class HistoScanWorkflow(RectGridWorkflow[HistoScanSettingsModel]):
**base_kwargs,
max_dist=self.max_range,
skip_background=self.skip_background,
smart_stack_params=self.create_smart_stack_params(),
)
def create_smart_stack_params(
self,
) -> SmartStackParams:
"""Set up the parameters used for all smart stacks in a scan.
:returns: A StackSmartParams object with the required parameters.
"""
# Coerce min_images_to_test parameter
min_images_to_test = self.stack_min_images_to_test
if min_images_to_test % 2 == 0:
min_images_to_test += 1
self.logger.warning(
"Minimum number of images to test should be odd, setting to "
f"{min_images_to_test}."
)
# Set the Thing property to the coerced value
self.stack_min_images_to_test = min_images_to_test
# Coerce the images to save parameter to be odd, and less than
# min_images_to_save
images_to_save = self.stack_images_to_save
if images_to_save > min_images_to_test:
self.logger.warning(
f"Cannot save {images_to_save} images as this above the minimum "
f"number to test. Setting images to save to {MAX_TEST_IMAGE_COUNT}."
)
images_to_save = min_images_to_test
elif images_to_save % 2 == 0:
images_to_save += 1
self.logger.warning(
f"Images to save should be odd, setting to {images_to_save}."
)
# Set the Thing property to the coerced value
self.stack_images_to_save = images_to_save
return SmartStackParams(
stack_dz=self.stack_dz,
images_to_save=self.stack_images_to_save,
min_images_to_test=self.stack_min_images_to_test,
save_on_failure=not self.skip_background,
smart_stack_params=self.create_smart_stack_params(
save_on_failure=not self.skip_background
),
)
def pre_scan_routine(self, settings: HistoScanSettingsModel) -> None:
@ -536,55 +614,88 @@ class HistoScanWorkflow(RectGridWorkflow[HistoScanSettingsModel]):
self.logger.info(msg)
return False, None
focus_height: Optional[int]
focused, focus_height = self._autofocus.run_smart_stack(
stack_parameters=settings.smart_stack_params,
capture_parameters=settings.capture_params,
autofocus_parameters=settings.autofocus_params,
return self._perform_smart_stack(settings, xyz_pos)
@lt.action
def check_background(self) -> str:
"""Check if sample is background.
This action is a pre-run check for feeding back to the user.
"""
image_array = self._cam.grab_as_array(stream_name="lores")
is_sample, bg_message = self._background_detector.image_is_sample(image_array)
label = "sample" if is_sample else "background"
return f"Current image is {label} ({bg_message})"
@lt.action
def set_background(self) -> None:
"""Set the background for this background detector.
This sets the background for this workflow's background detector as opposed to
the active background detector for the camera.
"""
image_array = self._cam.grab_as_array(stream_name="lores")
self._background_detector.set_background(image_array)
@lt.endpoint("get", "settings_ui", responses=UI_ELEMENT_RESPONSE)
def settings_ui(self) -> UIElementList:
"""Return the UI for the workflow's settings in the scan tab."""
scan_settings = UIElementList(
[
property_control_for(
self, "overlap", label="Image Overlap (0.1-0.7)", step=0.05
),
*self.smart_stack_property_controls(),
property_control_for(
self, "autofocus_dz", label="Autofocus Range (steps)", step=200
),
property_control_for(
self, "max_range", label="Maximum Distance (steps)", step=1000
),
property_control_for(
self, "skip_background", label="Detect and Skip Empty Fields"
),
property_control_for(
self, "equal_distances", label="Set Equal x and y Distances"
),
]
)
background_ui = self._background_detector.settings_ui()
set_bg_button = action_button_for(
self,
"set_background",
poll_interval=0.1,
submit_label="Set Background",
can_terminate=False,
notify_on_success=True,
success_message="Background image has been updated",
update_interface_on_response=True,
)
check_bg_button = action_button_for(
self,
"check_background",
poll_interval=0.1,
submit_label="Check Current Image",
disabled=not self._background_detector.ready,
can_terminate=False,
notify_on_success=True,
response_is_success_message=True,
)
# An image was captured if we are focussed or we are not skipping background.
imaged = focused or settings.smart_stack_params.save_on_failure
if not imaged:
msg = f"Stack failed at {xyz_pos}. Treating as background."
self.logger.info(msg)
background_ui.root += [set_bg_button, check_bg_button]
# run_smart_stage always returns a focus height for the sharpest image even
# if it failed to find a good focus. Set to None if not focussed.
if not focused:
focus_height = None
return imaged, focus_height
@lt.property
def settings_ui(self) -> list[PropertyControl]:
"""A list of PropertyControl objects to create the settings in the scan tab."""
return [
property_control_for(
self, "overlap", label="Image Overlap (0.1-0.7)", step=0.05
),
property_control_for(
self, "stack_images_to_save", label="Images in Stack to Save"
),
property_control_for(
self,
"stack_min_images_to_test",
label="Minimum number of images to test for focus",
),
property_control_for(self, "stack_dz", label="Stack dz (steps)", step=5),
property_control_for(
self, "autofocus_dz", label="Autofocus Range (steps)", step=200
),
property_control_for(
self, "max_range", label="Maximum Distance (steps)", step=1000
),
property_control_for(
self, "skip_background", label="Detect and Skip Empty Fields"
),
property_control_for(
self, "equal_distances", label="Set Equal x and y Distances"
),
]
return UIElementList(
[
HeaderBlock(text=self.display_name, level=4),
TextBlock(text=self.ui_blurb),
Accordion(title="Background Detect", children=background_ui),
Accordion(
title="Scan Settings",
children=scan_settings,
),
]
)
class RegularGridSettingsModel(RectGridSettingsModel):
@ -596,10 +707,20 @@ class RegularGridSettingsModel(RectGridSettingsModel):
x_count: int
y_count: int
smart_stack_params: SmartStackParams
style: Literal["snake", "raster"]
class RegularGridWorkflow(RectGridWorkflow[RegularGridSettingsModel]):
RegGridSettingModelType = TypeVar(
"RegGridSettingModelType", bound=RegularGridSettingsModel
)
class RegularGridWorkflow(
RectGridWorkflow[RegGridSettingModelType],
SmartStackMixin,
Generic[RegGridSettingModelType],
):
"""A base workflow for any workflow that uses a regular rectangular grid."""
x_count: int = lt.setting(default=3, ge=1)
@ -607,20 +728,21 @@ class RegularGridWorkflow(RectGridWorkflow[RegularGridSettingsModel]):
y_count: int = lt.setting(default=2, ge=1)
"""The number of rows in the scan."""
_settings_model = RegularGridSettingsModel
_settings_model: type[RegGridSettingModelType]
_planner_cls = RegularGridPlanner
_grid_style: Literal["snake", "raster"]
def _build_scan_settings(self, base_kwargs: dict) -> RegularGridSettingsModel:
def _build_scan_settings(self, base_kwargs: dict) -> RegGridSettingModelType:
"""Construct the SettingModel for all_settings."""
return RegularGridSettingsModel(
return self._settings_model(
**base_kwargs,
x_count=self.x_count,
y_count=self.y_count,
style=self._grid_style,
smart_stack_params=self.create_smart_stack_params(save_on_failure=True),
)
def pre_scan_routine(self, settings: RegularGridSettingsModel) -> None:
def pre_scan_routine(self, settings: RegGridSettingModelType) -> None:
"""Perform these steps before starting the scan.
In this case, only autofocus.
@ -632,11 +754,11 @@ class RegularGridWorkflow(RectGridWorkflow[RegularGridSettingsModel]):
)
def new_scan_planner(
self, settings: RegularGridSettingsModel, position: Mapping[str, int]
self, settings: RegGridSettingModelType, position: Mapping[str, int]
) -> ScanPlanner:
"""Return a new scan planner object.
:param settings: The settings for this scan as a SnakeSettingsModel
:param settings: The settings for this scan as the relevant SettingsModel type.
:param position: The starting position as a mapping of axes names to int.
"""
planner_settings = {
@ -652,36 +774,46 @@ class RegularGridWorkflow(RectGridWorkflow[RegularGridSettingsModel]):
)
def acquisition_routine(
self, settings: RegularGridSettingsModel, xyz_pos: tuple[int, int, int]
self, settings: RegGridSettingModelType, xyz_pos: tuple[int, int, int]
) -> tuple[bool, Optional[int]]:
"""Autofocus and capture.
:param settings: The settings for this scan as a RegularGridSettingsModel
:param settings: The settings for this scan as the relevant SettingsModel type.
:param xyz_pos: The current position as a tuple or 3 ints.
:return: A tuple of whether an image was taken, and the z-position for focus.
If failed to find focus, returns for the focus z-position.
"""
return self._autofocus_and_capture(
xyz_pos=xyz_pos,
dz=settings.autofocus_params.dz,
images_dir=settings.capture_params.images_dir,
save_resolution=settings.capture_params.save_resolution,
return self._perform_smart_stack(settings, xyz_pos)
@lt.endpoint("get", "settings_ui", responses=UI_ELEMENT_RESPONSE)
def settings_ui(self) -> UIElementList:
"""Return the UI for the workflow's settings in the scan tab."""
scan_settings = UIElementList(
[
property_control_for(
self, "overlap", label="Image Overlap (0.1-0.7)", step=0.05
),
property_control_for(self, "x_count", label="Number of columns"),
property_control_for(self, "y_count", label="Number of rows"),
*self.smart_stack_property_controls(),
property_control_for(
self, "autofocus_dz", label="Autofocus Range (steps)"
),
]
)
return UIElementList(
[
HeaderBlock(text=self.display_name, level=4),
TextBlock(text=self.ui_blurb),
Accordion(
title="Scan Settings",
children=scan_settings,
),
]
)
@lt.property
def settings_ui(self) -> list[PropertyControl]:
"""A list of PropertyControl objects to create the settings in the scan tab."""
return [
property_control_for(
self, "overlap", label="Image Overlap (0.1-0.7)", step=0.05
),
property_control_for(self, "x_count", label="Number of columns"),
property_control_for(self, "y_count", label="Number of rows"),
property_control_for(self, "autofocus_dz", label="Autofocus Range (steps)"),
]
class SnakeWorkflow(RegularGridWorkflow):
class SnakeWorkflow(RegularGridWorkflow[RegularGridSettingsModel]):
"""A workflow optimised for snaking around samples.
This workflow generates a list of coordinates in a rectangle, and snakes
@ -696,10 +828,11 @@ class SnakeWorkflow(RegularGridWorkflow):
),
readonly=True,
)
_settings_model = RegularGridSettingsModel
_grid_style = "snake"
class RasterWorkflow(RegularGridWorkflow):
class RasterWorkflow(RegularGridWorkflow[RegularGridSettingsModel]):
"""A workflow optimised for snaking around samples.
This workflow generates a list of coordinates in a rectangle, and always
@ -715,5 +848,148 @@ class RasterWorkflow(RegularGridWorkflow):
),
readonly=True,
)
_settings_model = RegularGridSettingsModel
_grid_style = "raster"
class CChipScanSettingsModel(RectGridSettingsModel):
"""The settings for a scan with the CChipWorkflow.
This includes settings calculated when starting. This will be held by smart scan
during a scan and serialised to disk.
"""
x_count: int
y_count: int
stack_params: StackParams
style: Literal["snake", "raster"]
class CChipWorkflow(RectGridWorkflow[CChipScanSettingsModel]):
"""A workflow optimised for scanning the well of a CChip.
This workflow generates a list of coordinates in a rectangle, and snakes
around them from the top left (assuming positive dx and dy), stacking the
grid and above.
"""
display_name: str = lt.property(default="C-Chip Scan", readonly=True)
ui_blurb: str = lt.property(
default=(
"This scan workflow is optimised for scanning a C-Chip. It focuses on "
"a grid, then stacks images above the grid to complete a volumetric scan."
),
readonly=True,
)
_grid_style: Literal["snake"] = "snake"
_planner_cls = RegularGridPlanner
_settings_model = CChipScanSettingsModel
overlap: float = lt.setting(default=0.1, ge=0.1, le=0.7)
"""The fraction that adjacent images should overlap in x and y.
This must be between 0.1 and 0.7.
"""
x_count: int = lt.setting(default=5, readonly=False)
"""The number of columns in the scan."""
y_count: int = lt.setting(default=7, readonly=False)
"""The number of rows in the scan."""
stack_images_to_save: int = lt.setting(default=9, readonly=False)
"""The number of images to save in a stack.
Defaults to 1 unless you need to see either side of focus
"""
stack_dz: int = lt.setting(default=500, readonly=False)
"""Distance in steps between images in a z-stack."""
def create_stack_params(
self,
) -> StackParams:
"""Set up the parameters used for all stacks in a scan.
:returns: A StackSmartParams object with the required parameters.
"""
return StackParams(
stack_dz=self.stack_dz, images_to_save=self.stack_images_to_save
)
def _build_scan_settings(self, base_kwargs: dict) -> CChipScanSettingsModel:
"""Construct the SettingModel for all_settings."""
stack_params = self.create_stack_params()
return self._settings_model(
**base_kwargs,
x_count=self.x_count,
y_count=self.y_count,
style=self._grid_style,
stack_params=stack_params,
)
def new_scan_planner(
self, settings: CChipScanSettingsModel, position: Mapping[str, int]
) -> ScanPlanner:
"""Return a new scan planner object.
:param settings: The settings for this scan as the relevant SettingsModel type.
:param position: The starting position as a mapping of axes names to int.
"""
planner_settings = {
"dx": settings.dx,
"dy": settings.dy,
"x_count": settings.x_count,
"y_count": settings.y_count,
"style": settings.style,
}
return self._planner_cls(
initial_position=(position["x"], position["y"]),
planner_settings=planner_settings,
)
def pre_scan_routine(self, settings: CChipScanSettingsModel) -> None:
"""No autofocus, a looping autofocus on a CChip could corrupt the entire scan."""
pass
def acquisition_routine(
self,
settings: CChipScanSettingsModel,
xyz_pos: tuple[int, int, int], # noqa: ARG002
) -> tuple[bool, Optional[int]]:
"""Autofocus and capture a z-stack starting at the focused position.
The routine performs a fast autofocus using the provided ``dz``.
The focused z-height becomes the starting position for the stack
and is also the height of the first captured image.
A stack of ``self.stack_images_to_save`` images is then acquired.
Each subsequent image is captured after moving the stage upward
by ``self.stack_dz`` steps in z.
:param xyz_pos: The (x, y, z) position associated with this acquisition.
:param settings: The settings for this scan as a CChipSettingsModel
:return: (True, focus_height) where focus_height is the autofocus
z-position and the height of the first image in the stack.
"""
# Perform autofocus
self._autofocus.fast_autofocus(dz=settings.autofocus_params.dz)
focus_height = self._stage.get_xyz_position()[2]
self._autofocus.run_basic_stack(settings.stack_params, settings.capture_params)
return True, focus_height
@lt.property
def settings_ui(self) -> list[PropertyControl]:
"""A list of PropertyControl objects to create the settings in the scan tab."""
return [
property_control_for(self, "overlap", label="Image Overlap (0.1-0.7)"),
property_control_for(self, "x_count", label="Number of columns"),
property_control_for(self, "y_count", label="Number of rows"),
property_control_for(self, "autofocus_dz", label="Autofocus Range (steps)"),
property_control_for(
self, "stack_dz", label="Distance in z between images in stack (steps)"
),
property_control_for(
self, "stack_images_to_save", label="Images to save per xy site"
),
]

8
src/openflexure_microscope_server/things/smart_scan.py
View file

@ -376,11 +376,7 @@ class SmartScanThing(OFMThing):
starting x,y,z position.
"""
try:
self._cam.start_streaming(main_resolution=(3280, 2464))
self._scan_data = self._collect_scan_data(workflow)
workflow.pre_scan_routine(self._scan_data.workflow_settings)
self.ongoing_scan.save_scan_data(self._scan_data)
images_dir = self.ongoing_scan.images_dir
# Type narrowing
if images_dir is None:
@ -388,6 +384,10 @@ class SmartScanThing(OFMThing):
"Couldn't run scan, images directory was not created."
)
self.ongoing_scan.save_scan_data(self._scan_data)
self._cam.start_streaming(main_resolution=(3280, 2464))
workflow.pre_scan_routine(self._scan_data.workflow_settings)
# If stitching settings are None then this type of scan can't be stitched
if self.scan_data.stitching_settings is not None:
# Settings exist, so create preview stitcher

229
src/openflexure_microscope_server/ui.py
View file

@ -1,11 +1,155 @@
"""Functionality for communicating the required user interface for a thing."""
from typing import Any, Optional
from html import escape
from html.parser import HTMLParser
from typing import Annotated, Any, Literal, Optional
from urllib.parse import urlparse
from pydantic import BaseModel
from pydantic import BaseModel, BeforeValidator, Field, RootModel
import labthings_fastapi as lt
# Only allowed tags are italic, bold, and link. Also allows self closing br tags.
ALLOWED_TAGS = {"i", "b", "a"}
ALLOWED_SCHEMES = {"http", "https"}
def strip_control_chars(input_string: str) -> str:
"""Remove unprintable characters."""
return "".join(char for char in input_string if char.isprintable())
def sanitize_http_url(url: str) -> Optional[str]:
"""Santitise any url only allowing http and https without queries."""
url = strip_control_chars(url)
parsed = urlparse(url)
if parsed.scheme not in ALLOWED_SCHEMES:
return None
if not parsed.netloc:
return None
fragment_str = "#" + parsed.fragment if parsed.fragment else ""
return f"{parsed.scheme}://{parsed.netloc}{parsed.path}{fragment_str}"
class SafeHTMLParser(HTMLParser):
"""An HTMLParser that only allows a very limited subset of HTML."""
def __init__(self) -> None:
"""Initialise the parser."""
super().__init__()
self.output = ""
def handle_starttag(self, tag: str, attrs: list[tuple[str, Optional[str]]]) -> None:
"""Append a start tag found in the HTML if it is in the allowed list.
This is called by the base class when a start tag is found. We only append
the tag if it is our allowed list. We re-write the tag so no extra attributes
can be added.
No tags can have attributes except for ``a`` tags, why only allow ``href``. We then
append ``target="_blank" rel="noopener noreferrer"`` to ensure the link opens
in a new window.
:param tag: the tag name
:param attrs: the attributes for the tag.
"""
if tag not in ALLOWED_TAGS:
return
if tag == "a":
href = None
for attr, value in attrs:
if attr == "href" and value is not None:
# If it is not a httplink this will stay as None
href = sanitize_http_url(value.strip())
if href:
safe_href = escape(href, quote=True)
# Always in a new tab
self.output += (
f'<a href="{safe_href}" target="_blank" rel="noopener noreferrer">'
)
else:
self.output += "<a>"
else:
self.output += f"<{tag}>"
def handle_endtag(self, tag: str) -> None:
"""Append the end tag only if it is in the allowed list."""
if tag in ALLOWED_TAGS:
self.output += f"</{tag}>"
def handle_startendtag(
self, tag: str, _attrs: list[tuple[str, Optional[str]]]
) -> None:
"""Append a self-closing tag if it is a new line."""
if tag == "br":
self.output += "<br/>"
def handle_data(self, data: str) -> None:
"""Append any data inside tags after escaping it."""
self.output += escape(data)
def handle_entityref(self, name: str) -> None:
"""Append any named character."""
self.output += f"&{name};"
def handle_charref(self, name: str) -> None:
"""Append any character references."""
self.output += f"&#{name};"
def sanitise_html(html: str) -> str:
"""Santitise HTML to only have a small list of allowed tags.
Tags allowed without attrs: ``<i>, <b>,``
Tags allowed with attrs: ``<a>`` is allowed with ``href`` only. This automatically
appends ``target="_blank" rel="noopener noreferrer"`` so the link opens externally.
Self closing tags allowed ``<br>``.
:param html: The input HTML.
:return: The sanitised HTML.
"""
parser = SafeHTMLParser()
parser.feed(html)
parser.close()
return parser.output
HtmlFragment = Annotated[str, BeforeValidator(sanitise_html)]
class HeaderBlock(BaseModel):
"""The data required for header."""
element_type: Literal["header_block"] = "header_block"
text: HtmlFragment
"""The header text (can include basic HTML tags)."""
level: int = Field(default=2, ge=1, le=6)
"""The header level. A number from 1-6."""
class TextBlock(BaseModel):
"""The data required for creating a block text."""
element_type: Literal["text_block"] = "text_block"
text: HtmlFragment
"""The text (can include basic HTML tags)."""
class BulletBlock(BaseModel):
"""The data required for creating a block text."""
element_type: Literal["bullet_block"] = "bullet_block"
bullets: list[HtmlFragment]
"""A list of text elements (can include basic HTML tags)."""
class ActionButton(BaseModel):
"""The data required for creating an actionButton in Vue.
@ -13,13 +157,15 @@ class ActionButton(BaseModel):
Currently this cannot be used to submitData, or to submitOnEvent.
"""
element_type: Literal["action_button"] = "action_button"
thing: str
"""The Thing "path" for the Thing instance."""
action: str
"""The name of the action to be triggered."""
poll_interval: int = 1
poll_interval: float = 1
"""The interval for polling in seconds."""
submit_label: str = "Submit"
@ -37,6 +183,9 @@ class ActionButton(BaseModel):
button_primary: bool = True
"""Specify whether the button is styled as a primary button."""
disabled: bool = False
"""Specify whether the button is disabled."""
modal_progress: bool = False
"""Specify whether to show a progress modal."""
@ -50,9 +199,21 @@ class ActionButton(BaseModel):
notify_on_success: bool = False
"""Specify whether to a notification on successful completion."""
response_is_success_message: bool = False
"""Set True to use the action response as the success message.
If ``success_message`` is set, this is never used.
"""
success_message: str = "Success!"
"""The message to show on successful completion."""
update_interface_on_response: bool = False
"""If True the action button emits a requestUpdate signal when completed.
Downstream components can subscribe to this and request an update.
"""
def action_button_for(thing: lt.Thing, action_name: str, **kwargs: Any) -> ActionButton:
"""Create a ActionButton data for the specified Thing Action.
@ -72,6 +233,8 @@ class PropertyControl(BaseModel):
Currently this cannot be used to submitData, or to submitOnEvent.
"""
element_type: Literal["property_control"] = "property_control"
thing: str
"""The Thing "path" for the Thing instance."""
@ -121,3 +284,63 @@ def property_control_for(
if "label" not in kwargs:
kwargs["label"] = property_name
return PropertyControl(thing=thing.name, property_name=property_name, **kwargs)
class Accordion(BaseModel):
"""The data required for creating an accordion in the UI.
The HTML is limited to a small number of tags.
"""
element_type: Literal["accordion"] = "accordion"
title: str
children: "UIElementList"
class Container(BaseModel):
"""The data required for creating ``<div>`` in the UI.
The HTML is limited to a small number of tags.
"""
element_type: Literal["container"] = "container"
css_class: str
children: "UIElementList"
UIElementModels = (
HeaderBlock
| TextBlock
| BulletBlock
| PropertyControl
| ActionButton
| Accordion
| Container
)
class UIElementList(RootModel[list[UIElementModels]]):
"""A list of user interface elements.
Note! Two important things to consider:
* This cannot be the return of a lt.property. To fully describe the UI recursion
is used. Web of Things DataSchema doesn't support recursion. Use an
lt.endpoint instead
* If the module using this class imports future annotations then this will cause
havoc with pydantics forward references.
"""
# Resolve forward references so Pydantic can build a sensible recursive schema
Accordion.model_rebuild()
Container.model_rebuild()
UIElementList.model_rebuild()
# This constant can be used as the value for the `responses` argument of lt.endpoint.
UI_ELEMENT_RESPONSE = {
200: {
"description": "A list of UI element specifications.",
"model": UIElementList,
}
}

160
src/openflexure_microscope_server/utilities.py
View file

@ -76,34 +76,131 @@ def requires_lock(
# Compiled regular expressions for unsafe characters
# Matches anything that isn't a-z, A-Z, 0-9, _, ., -, :, /, \
_WINDOWS_UNSAFE_PATTERN = re.compile(r"[^a-zA-Z0-9_.\-:/\\]")
# Matches anything that isn't a-z, A-Z, 0-9, _, ., -, \
_POSIX_UNSAFE_PATTERN = re.compile(r"[^a-zA-Z0-9_.\-/]")
# Matches anything that isn't a-z, A-Z, 0-9, _, ., -, :, /, \, space
_WINDOWS_UNSAFE_PATTERN = re.compile(r"[^a-zA-Z0-9_.\-:/\ \\]")
# Matches anything that isn't a-z, A-Z, 0-9, _, ., -, \, space
_POSIX_UNSAFE_PATTERN = re.compile(r"[^a-zA-Z0-9_.\-/ ]")
# Matches anything that isn't a-z, A-Z, 0-9, _, ., -
_NAME_UNSAFE_PATTERN = re.compile(r"[^a-zA-Z0-9_.\-]")
# Windows reserved names (case-insensitive)
_WINDOWS_RESERVED_NAMES = {
"CON",
"PRN",
"AUX",
"NUL",
"COM1",
"COM2",
"COM3",
"COM4",
"COM5",
"COM6",
"COM7",
"COM8",
"COM9",
"LPT1",
"LPT2",
"LPT3",
"LPT4",
"LPT5",
"LPT6",
"LPT7",
"LPT8",
"LPT9",
}
def make_path_safe(unsafe_path_string: str) -> str:
"""Replace unsafe characters in a file path with underscores, preserving separators.
This can be used to check that user inputs have not been concatenated into an
unsafe filepath.
def is_path_safe(unsafe_path_string: str) -> bool:
"""Check if a file path has any unsafe elements in it.
This ensures compatibility across platforms by preserving only valid characters,
including path separators (forward slash on POSIX, and forward
slash/backslash/colon on Windows).
The path is not coerced into a safe form because if we ask
for a file to be written to a location we shouldn't be changing
that location as we have no "consent" to write to this other location.
:param unsafe_path_string: The original path string to sanitise.
:returns: A version of the input string safe to use as a file path.
:returns: A boolean indicating if any unsafe features were found.
"""
# Split by separators first to sanitise components independently
components = re.split(r"([/\\])", unsafe_path_string)
unsafe_character_pattern = (
_WINDOWS_UNSAFE_PATTERN
if sys.platform.startswith("win")
else _POSIX_UNSAFE_PATTERN
)
return unsafe_character_pattern.sub("_", unsafe_path_string)
# Ensure each warning only gets logged once per path
unsafe_relative_path = False
trailing_dots = False
trailing_whitespace = False
unsafe_characters = False
reserved_word = False
for i, component in enumerate(components):
# Skip separators and empty strings
if component in ("/", "\\") or not component:
continue
# 1. Check for relative paths in the wrong place
if component in (".", "..") and not unsafe_relative_path:
is_first = i == 0
is_second = (
i == 2
and components[i - 1] in ("/", "\\")
and components[i - 2] == ".."
)
if not (is_first or is_second):
LOGGER.warning(
f"File path {unsafe_path_string} may be unsafe due to unexpected relative navigation."
)
unsafe_relative_path = True
continue
# 2. Check for trailing dots in path
if "." in component and i != len(components) - 1 and not trailing_dots:
# Check for trailing dots in the file path before the file name
# e.g.: foo/bar./file is bad, but foo/bar.file.py is fine
LOGGER.warning(
f"File path {unsafe_path_string} may be unsafe due to trailing dots."
)
trailing_dots = True
# Check for trailing spaces - Windows specific
if (
sys.platform.startswith("win")
and component.endswith(" ")
and not trailing_whitespace
):
LOGGER.warning(f"{unsafe_path_string} contains unsafe trailing spaces.")
trailing_whitespace = True
# 3. Check for unsafe characters (Regex)
if unsafe_character_pattern.search(component) and not unsafe_characters:
LOGGER.warning(
f"{unsafe_path_string} contains characters that may be unsafe on this platform."
)
unsafe_characters = True
# 4. Check for Reserved Names (e.g., CON, PRN, LPT1)
# Assuming _sanitise_reserved returns a different string if it's a reserved name
if (
_sanitise_reserved(component, is_filename=False) != component
and not reserved_word
):
LOGGER.warning(
f"{unsafe_path_string} contains a reserved system name and may cause issues."
)
reserved_word = True
return not (
unsafe_relative_path
or trailing_dots
or trailing_whitespace
or unsafe_characters
or reserved_word
)
def make_name_safe(unsafe_name_string: str) -> str:
@ -112,11 +209,44 @@ def make_name_safe(unsafe_name_string: str) -> str:
This excludes all path separators, ensuring the result is safe to use as a
standalone filename or identifier component.
This also handles Windows reserved names and trailing dots/spaces.
:param unsafe_name_string: The original name string to sanitise.
:returns: A version of the input string safe to use as a file name or identifier.
"""
return _NAME_UNSAFE_PATTERN.sub("_", unsafe_name_string)
# 1. Strip trailing dots and spaces
# 2. Apply unsafe character regex
# 3. Check for reserved names
name = unsafe_name_string.rstrip(". ")
if not name:
return "_"
name = _NAME_UNSAFE_PATTERN.sub("_", name)
return _sanitise_reserved(name)
def _sanitise_reserved(name: str, is_filename: bool = True) -> str:
"""Check a name against Windows reserved names.
:param name: The component to check.
:param is_filename: Whether the component is a filename.
If True, names will be forced into lowercase.
:returns: The sanitised component, in lower case.
"""
# Check for Windows reserved names.
# These are reserved even with an extension (e.g. NUL.txt).
# We check the part before the first dot.
base_name = name.split(".", maxsplit=1)[0].upper()
if base_name in _WINDOWS_RESERVED_NAMES:
return f"{name.lower()}_" if is_filename else f"{name}_"
# We force names to be lowercase to avoid issues on
# Windows where files with the same name
# but different case can cause problems when checking if
# a file already exists.
return name.lower() if is_filename else name
class VersionData(BaseModel):
@ -291,7 +421,7 @@ def merge_patch(
:param target: The target object
:param patch: The patch to be applied
:param enforce_dict: Boolean, set True enfoces that the target and patch are both
:param enforce_dict: Boolean, set True enforces that the target and patch are both
dictionaries.
"""
if enforce_dict and not (isinstance(target, dict) and isinstance(patch, dict)):