Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
48 changes: 48 additions & 0 deletions software/control/acquisition_yaml_loader.py
Original file line number Diff line number Diff line change
Expand Up @@ -7,6 +7,20 @@
from typing import Dict, List, Optional, Tuple


@dataclass
class ChannelYAMLSettings:
"""Per-channel settings parsed from an acquisition YAML's channels section.

Fields are None when absent from the YAML (e.g. files written by older
versions), so callers can restore only what was actually recorded.
"""

name: str
exposure_time_ms: Optional[float] = None
analog_gain: Optional[float] = None
illumination_intensity: Optional[float] = None


@dataclass
class AcquisitionYAMLData:
"""Parsed acquisition YAML data structure."""
Expand All @@ -32,6 +46,7 @@ class AcquisitionYAMLData:

# Channels
channel_names: List[str] = field(default_factory=list)
channel_settings: List[ChannelYAMLSettings] = field(default_factory=list)

# Autofocus
contrast_af: bool = False
Expand All @@ -51,6 +66,38 @@ class AcquisitionYAMLData:
flexible_positions: Optional[List[Dict]] = None # [{name, center_mm}, ...]


def _as_float(value) -> Optional[float]:
"""Coerce a YAML scalar to float, returning None for anything non-numeric."""
if isinstance(value, bool) or not isinstance(value, (int, float)):
return None
return float(value)


def _parse_channel_settings(channels: List[Dict]) -> List[ChannelYAMLSettings]:
"""Extract per-channel settings from serialized AcquisitionChannel dicts.

Channels are written by serialize_for_yaml(AcquisitionChannel), so settings
live in nested camera_settings / illumination_settings dicts. Entries without
a name are skipped; missing or non-numeric values parse to None.
"""
parsed = []
for ch in channels:
name = ch.get("name")
if not name:
continue
camera_settings = ch.get("camera_settings") or {}
illumination_settings = ch.get("illumination_settings") or {}
parsed.append(
ChannelYAMLSettings(
name=name,
exposure_time_ms=_as_float(camera_settings.get("exposure_time_ms")),
analog_gain=_as_float(camera_settings.get("gain_mode")),
illumination_intensity=_as_float(illumination_settings.get("intensity")),
)
)
return parsed


def parse_acquisition_yaml(file_path: str) -> AcquisitionYAMLData:
"""Parse acquisition YAML file and return structured data.

Expand Down Expand Up @@ -126,6 +173,7 @@ def parse_acquisition_yaml(file_path: str) -> AcquisitionYAMLData:
delta_t_s=time_series.get("delta_t_s", 0.0),
# Channels
channel_names=[ch.get("name") for ch in channels if ch.get("name")],
channel_settings=_parse_channel_settings(channels),
# Autofocus
contrast_af=autofocus.get("contrast_af", False),
laser_af=autofocus.get("laser_af", False),
Expand Down
6 changes: 6 additions & 0 deletions software/control/gui_hcs.py
Original file line number Diff line number Diff line change
Expand Up @@ -1461,11 +1461,17 @@ def make_connections(self):
if ENABLE_FLEXIBLE_MULTIPOINT:
self.flexibleMultiPointWidget.signal_acquisition_started.connect(self.toggleAcquisitionStart)
self.signal_performance_mode_changed.connect(self.flexibleMultiPointWidget.set_performance_mode)
self.flexibleMultiPointWidget.signal_channel_settings_restored.connect(
self.liveControlWidget.refresh_current_mode_settings
)

if ENABLE_WELLPLATE_MULTIPOINT:
self.wellplateMultiPointWidget.signal_acquisition_started.connect(self.toggleAcquisitionStart)
self.wellplateMultiPointWidget.signal_toggle_live_scan_grid.connect(self.toggle_live_scan_grid)
self.signal_performance_mode_changed.connect(self.wellplateMultiPointWidget.set_performance_mode)
self.wellplateMultiPointWidget.signal_channel_settings_restored.connect(
self.liveControlWidget.refresh_current_mode_settings
)

if RUN_FLUIDICS:
self.multiPointWithFluidicsWidget.signal_acquisition_started.connect(self.toggleAcquisitionStart)
Expand Down
82 changes: 82 additions & 0 deletions software/control/widgets.py
Original file line number Diff line number Diff line change
Expand Up @@ -834,6 +834,8 @@ class AcquisitionYAMLDropMixin:
2. Have `self._log`, `self.multipointController`, `self.objectiveStore` attributes
3. Implement `_get_expected_widget_type()` returning "wellplate" or "flexible"
4. Implement `_apply_yaml_settings(yaml_data)` to apply settings to the widget
5. Optionally define a `signal_channel_settings_restored` Signal, emitted after
per-channel settings (exposure/gain/intensity) are restored from the YAML
"""

def _is_valid_yaml_drop(self, file_path: str) -> bool:
Expand Down Expand Up @@ -953,13 +955,73 @@ def _load_acquisition_yaml(self, file_path: str) -> bool:

# Apply settings with signal blocking
self._apply_yaml_settings(yaml_data)
self._restore_channel_settings(yaml_data)
self._log.info(f"Loaded acquisition settings from: {file_path}")
return True

def _apply_yaml_settings(self, yaml_data):
"""Apply parsed YAML settings to widget controls. Override in subclass."""
raise NotImplementedError("Subclass must implement _apply_yaml_settings()")

# (setting name, value extractor, validator) for _restore_channel_settings.
# Validators mirror the AcquisitionChannel model constraints: values are written
# to the profile config with update_channel_setting, which bypasses pydantic
# assignment validation — persisting an out-of-range value would make the
# profile fail validation on next load.
_RESTORABLE_CHANNEL_SETTINGS = (
("ExposureTime", lambda ch: ch.exposure_time_ms, lambda v: v > 0),
("AnalogGain", lambda ch: ch.analog_gain, lambda v: v >= 0),
("IlluminationIntensity", lambda ch: ch.illumination_intensity, lambda v: 0 <= v <= 100),
)

def _restore_channel_settings(self, yaml_data) -> int:
"""Restore per-channel settings (exposure, gain, intensity) from the YAML.

Values are persisted to the current profile's objective config — the same
path used when editing them in the live control panel. Z-offset is
intentionally NOT restored: it is sample-dependent (relative to the laser
AF reference), so an old acquisition's offsets don't transfer.

Returns the number of channels whose settings were restored.
"""
live_controller = self.multipointController.liveController
config_repo = live_controller.microscope.config_repo
objective = self.objectiveStore.current_objective
confocal_mode = live_controller.is_confocal_mode()

restored = 0
for channel in yaml_data.channel_settings:
if live_controller.get_channel_by_name(objective, channel.name) is None:
self._log.warning(
f"Channel '{channel.name}' from YAML not found in current configuration; settings not restored"
)
continue
applied_any = False
for setting, get_value, is_valid in self._RESTORABLE_CHANNEL_SETTINGS:
value = get_value(channel)
if value is None:
continue
if not is_valid(value):
self._log.warning(
f"Skipping out-of-range {setting}={value} for channel '{channel.name}' from YAML"
)
continue
if config_repo.update_channel_setting(
objective, channel.name, setting, value, confocal_mode=confocal_mode
):
applied_any = True
else:
self._log.warning(f"Failed to restore {setting} for channel '{channel.name}' from YAML")
if applied_any:
restored += 1

if restored:
self._log.info(f"Restored settings for {restored} channel(s) from YAML")
signal = getattr(self, "signal_channel_settings_restored", None)
if signal is not None:
signal.emit()
return restored


class _ApplyChannelOffsetMixin:
"""Mixin providing the laser-AF per-channel Z-offset checkbox + handlers.
Expand Down Expand Up @@ -4336,6 +4398,24 @@ def select_new_microscope_mode_by_name(self, config_name):
self.update_ui_for_mode(maybe_new_config)
self._maybe_apply_live_channel_offset(maybe_new_config)

def refresh_current_mode_settings(self):
"""Re-read the current channel's settings from config and update the UI.

Used when channel settings are changed outside this widget (e.g. restored
from a dropped acquisition YAML). Unlike select_new_microscope_mode_by_name,
this does not apply the live per-channel Z-offset — the channel didn't
change, so no stage move is warranted.
"""
if self.currentConfiguration is None:
return
config = self.liveController.get_channel_by_name(
self.objectiveStore.current_objective, self.currentConfiguration.name
)
if config is None:
return
self.liveController.set_microscope_mode(config)
self.update_ui_for_mode(config)

def update_ui_for_mode(self, config):
try:
self.is_switching_mode = True
Expand Down Expand Up @@ -5702,6 +5782,7 @@ class FlexibleMultiPointWidget(AcquisitionYAMLDropMixin, _ApplyChannelOffsetMixi
signal_acquisition_started = Signal(bool) # true = started, false = finished
signal_acquisition_channels = Signal(list) # list channels
signal_acquisition_shape = Signal(int, float) # Nz, dz
signal_channel_settings_restored = Signal() # channel settings restored from dropped acquisition YAML

def __init__(
self,
Expand Down Expand Up @@ -7165,6 +7246,7 @@ class WellplateMultiPointWidget(AcquisitionYAMLDropMixin, _ApplyChannelOffsetMix
signal_acquisition_shape = Signal(int, float) # acquisition Nz, dz
signal_manual_shape_mode = Signal(bool) # enable manual shape layer on mosaic display
signal_toggle_live_scan_grid = Signal(bool) # enable/disable live scan grid
signal_channel_settings_restored = Signal() # channel settings restored from dropped acquisition YAML
# Signal to set acquisition running state from any thread (used by TCP server)
signal_set_acquisition_running = Signal(bool, int, float) # is_running, nz, delta_z_um

Expand Down
120 changes: 120 additions & 0 deletions software/tests/control/test_acquisition_yaml_channel_restore.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,120 @@
"""Tests for AcquisitionYAMLDropMixin._restore_channel_settings.

Dropping a previous acquisition onto a multipoint widget must restore the
per-channel settings (exposure, analog gain, illumination intensity) recorded
in its acquisition.yaml, persisting them through ConfigRepository the same way
live-control edits do. Channels missing from the current configuration and
out-of-range values are skipped: update_channel_setting bypasses pydantic
assignment validation, so persisting a bad value would corrupt the profile.
Z-offset is sample-dependent and must never be restored.
"""

from unittest.mock import MagicMock, call

from control.acquisition_yaml_loader import AcquisitionYAMLData, ChannelYAMLSettings
from control.widgets import AcquisitionYAMLDropMixin


class _Stub(AcquisitionYAMLDropMixin):
"""Minimal host for the mixin with mocked controller/repo plumbing."""

def __init__(self, known_channels, confocal_mode=False):
self._log = MagicMock()
self.objectiveStore = MagicMock()
self.objectiveStore.current_objective = "20x"
self.multipointController = MagicMock()
live = self.multipointController.liveController
live.is_confocal_mode.return_value = confocal_mode
live.get_channel_by_name.side_effect = lambda objective, name: (
MagicMock() if name in known_channels else None
)
self.repo = live.microscope.config_repo
self.repo.update_channel_setting.return_value = True


def _yaml_data(channel_settings):
return AcquisitionYAMLData(widget_type="wellplate", channel_settings=channel_settings)


def test_restores_all_settings_for_known_channel():
stub = _Stub(known_channels={"BF"})
data = _yaml_data(
[ChannelYAMLSettings(name="BF", exposure_time_ms=12.5, analog_gain=2.0, illumination_intensity=30.0)]
)

assert stub._restore_channel_settings(data) == 1
stub.repo.update_channel_setting.assert_has_calls(
[
call("20x", "BF", "ExposureTime", 12.5, confocal_mode=False),
call("20x", "BF", "AnalogGain", 2.0, confocal_mode=False),
call("20x", "BF", "IlluminationIntensity", 30.0, confocal_mode=False),
]
)


def test_passes_current_confocal_mode():
stub = _Stub(known_channels={"BF"}, confocal_mode=True)
data = _yaml_data([ChannelYAMLSettings(name="BF", exposure_time_ms=10.0)])

stub._restore_channel_settings(data)
stub.repo.update_channel_setting.assert_called_once_with("20x", "BF", "ExposureTime", 10.0, confocal_mode=True)


def test_skips_channel_missing_from_current_configuration():
stub = _Stub(known_channels={"BF"})
data = _yaml_data(
[
ChannelYAMLSettings(name="Removed Channel", exposure_time_ms=50.0),
ChannelYAMLSettings(name="BF", exposure_time_ms=12.0),
]
)

assert stub._restore_channel_settings(data) == 1
updated_channels = {c.args[1] for c in stub.repo.update_channel_setting.call_args_list}
assert updated_channels == {"BF"}


def test_skips_none_values_from_older_yamls():
stub = _Stub(known_channels={"BF"})
data = _yaml_data([ChannelYAMLSettings(name="BF")]) # name-only channel entry

assert stub._restore_channel_settings(data) == 0
stub.repo.update_channel_setting.assert_not_called()


def test_skips_out_of_range_values():
stub = _Stub(known_channels={"BF"})
data = _yaml_data(
[ChannelYAMLSettings(name="BF", exposure_time_ms=0.0, analog_gain=-1.0, illumination_intensity=150.0)]
)

assert stub._restore_channel_settings(data) == 0
stub.repo.update_channel_setting.assert_not_called()


def test_repo_failure_does_not_count_channel():
stub = _Stub(known_channels={"BF"})
stub.repo.update_channel_setting.return_value = False
data = _yaml_data([ChannelYAMLSettings(name="BF", exposure_time_ms=12.0)])

assert stub._restore_channel_settings(data) == 0


def test_emits_signal_only_when_settings_restored():
stub = _Stub(known_channels={"BF"})
stub.signal_channel_settings_restored = MagicMock()
data = _yaml_data([ChannelYAMLSettings(name="BF", exposure_time_ms=12.0)])

stub._restore_channel_settings(data)
stub.signal_channel_settings_restored.emit.assert_called_once()

stub.signal_channel_settings_restored.emit.reset_mock()
stub._restore_channel_settings(_yaml_data([ChannelYAMLSettings(name="BF")]))
stub.signal_channel_settings_restored.emit.assert_not_called()


def test_no_signal_attribute_is_tolerated():
stub = _Stub(known_channels={"BF"}) # no signal_channel_settings_restored defined
data = _yaml_data([ChannelYAMLSettings(name="BF", exposure_time_ms=12.0)])

assert stub._restore_channel_settings(data) == 1
Loading
Loading