This commit is contained in:
@@ -0,0 +1,852 @@
|
||||
import pytest
|
||||
from pathlib import Path
|
||||
import os
|
||||
|
||||
from slic.core.scanner.scanbackend import (
|
||||
ScanBackend,
|
||||
is_sfdaq, is_only_sfdaq,
|
||||
print_all_current_values, get_all_current_values,
|
||||
set_all_target_values_and_wait, set_all_target_values,
|
||||
wait_for_all, stop_all,
|
||||
)
|
||||
|
||||
from slic.core.acquisition import SFAcquisition
|
||||
from slic.core.acquisition.fakeacquisition import FakeAcquisition
|
||||
from slic.core.adjustable.dummyadjustable import DummyAdjustable
|
||||
from slic.core.task import DAQTask
|
||||
|
||||
|
||||
# Dummies for conditions/sensors
|
||||
|
||||
class DummyCondition:
|
||||
def __init__(self, repeats=0):
|
||||
self.repeats = repeats
|
||||
self._stopped = False
|
||||
def wants_repeat(self):
|
||||
self.repeats -= 1
|
||||
return self.repeats >= 0
|
||||
def stop(self):
|
||||
self._stopped = True
|
||||
|
||||
|
||||
class DummySensor:
|
||||
counter = 0
|
||||
def __init__(self, name=None):
|
||||
DummySensor.counter += 1
|
||||
self.name = name or f"sensor_{DummySensor.counter}"
|
||||
self.started = False
|
||||
self.stopped = False
|
||||
def start(self):
|
||||
self.started = True
|
||||
def stop(self):
|
||||
self.stopped = True
|
||||
def get(self):
|
||||
return 3.14
|
||||
|
||||
|
||||
class DummyRemotePlot:
|
||||
def __init__(self, fail=False):
|
||||
self.fail = fail
|
||||
self.created = False
|
||||
self.appended = False
|
||||
self.last_data = None
|
||||
self.last_filename = None
|
||||
|
||||
def new_plot(self, filename, cfg):
|
||||
self.created = True
|
||||
if self.fail:
|
||||
raise ConnectionRefusedError
|
||||
|
||||
def append_data(self, filename, data):
|
||||
self.appended = True
|
||||
self.last_data = data
|
||||
self.last_filename = filename
|
||||
if self.fail:
|
||||
raise ConnectionRefusedError
|
||||
|
||||
|
||||
# Main ScanBackend tests
|
||||
|
||||
def test_is_sfdaq_and_only_sfdaq():
|
||||
# Test SFDAQ detection functions with different acquisition types
|
||||
s1, s2 = SFAcquisition(), SFAcquisition()
|
||||
f1 = FakeAcquisition()
|
||||
random_obj = object()
|
||||
|
||||
assert is_sfdaq(s1)
|
||||
assert is_sfdaq(f1)
|
||||
assert not is_sfdaq(random_obj)
|
||||
|
||||
assert is_only_sfdaq([s1, s2])
|
||||
assert is_only_sfdaq([f1, s1])
|
||||
assert not is_only_sfdaq([s1, random_obj])
|
||||
|
||||
def test_get_filename(tmp_path):
|
||||
# Test filename generation with both make_scan_sub_dir modes
|
||||
adjs = [DummyAdjustable(name="AX", ID="AX")]
|
||||
acqs = [FakeAcquisition()]
|
||||
|
||||
# Case 1 - make_scan_sub_dir = False
|
||||
sb1 = ScanBackend(
|
||||
adjs, [[1]], acqs, "scanfile",
|
||||
detectors=[], channels=[], pvs=[],
|
||||
n_pulses=1, data_base_dir="data", scan_info_dir=tmp_path,
|
||||
make_scan_sub_dir=False, condition=None,
|
||||
return_to_initial_values=True, n_repeat=1,
|
||||
sensor=None, remote_plot=None
|
||||
)
|
||||
f1 = sb1.get_filename(7)
|
||||
assert f1.endswith("scanfile_step0007")
|
||||
assert os.path.basename(f1).startswith("scanfile")
|
||||
|
||||
# Case 2 - make_scan_sub_dir = True
|
||||
sb2 = ScanBackend(
|
||||
adjs, [[1]], acqs, "scanfile",
|
||||
detectors=[], channels=[], pvs=[],
|
||||
n_pulses=1, data_base_dir="data", scan_info_dir=tmp_path,
|
||||
make_scan_sub_dir=True, condition=None,
|
||||
return_to_initial_values=True, n_repeat=1,
|
||||
sensor=None, remote_plot=None
|
||||
)
|
||||
f2 = sb2.get_filename(3)
|
||||
expected_sub = os.path.join("scanfile", "scanfile_step0003")
|
||||
assert f2.endswith(expected_sub)
|
||||
|
||||
sb3 = ScanBackend(
|
||||
adjs, [[1]], acqs, "/tmp/path/to/custom_name",
|
||||
detectors=[], channels=[], pvs=[],
|
||||
n_pulses=1, data_base_dir="data", scan_info_dir=tmp_path,
|
||||
make_scan_sub_dir=True, condition=None,
|
||||
return_to_initial_values=True, n_repeat=1,
|
||||
sensor=None, remote_plot=None
|
||||
)
|
||||
f3 = sb3.get_filename(1)
|
||||
assert f3.endswith(os.path.join("custom_name", "custom_name_step0001"))
|
||||
|
||||
|
||||
def test_create_output_dirs(tmp_path):
|
||||
# Test directory creation behavior with different acquisition types and settings
|
||||
adjs = [DummyAdjustable(ID="A1")]
|
||||
|
||||
# Test case 1: SFDAQ only, make_scan_sub_dir=True
|
||||
sfdaq_acq = SFAcquisition()
|
||||
sb = ScanBackend(
|
||||
adjs, [[1]], [sfdaq_acq],
|
||||
filename="scan_sfdaq",
|
||||
detectors=[], channels=[], pvs=[],
|
||||
n_pulses=1, data_base_dir="data", scan_info_dir=tmp_path,
|
||||
make_scan_sub_dir=True, condition=None,
|
||||
return_to_initial_values=True, n_repeat=1,
|
||||
sensor=None, remote_plot=None
|
||||
)
|
||||
|
||||
sb.create_output_dirs()
|
||||
for root, dirs, files in os.walk(tmp_path):
|
||||
assert not dirs
|
||||
|
||||
# Test case 2: FakeAcquisition only, make_scan_sub_dir=False
|
||||
fake_acq = FakeAcquisition("non_sfdaq", "p1")
|
||||
fake_acq.default_dir = str(tmp_path / "fake_default")
|
||||
sb = ScanBackend(
|
||||
adjs, [[1]], [fake_acq],
|
||||
filename="scan_fake",
|
||||
detectors=[], channels=[], pvs=[],
|
||||
n_pulses=1, data_base_dir="data", scan_info_dir=tmp_path,
|
||||
make_scan_sub_dir=False, condition=None,
|
||||
return_to_initial_values=True, n_repeat=1,
|
||||
sensor=None, remote_plot=None
|
||||
)
|
||||
|
||||
sb.create_output_dirs()
|
||||
assert (tmp_path / sb.scan_info.base_dir).exists()
|
||||
expected_data_dir = os.path.join(fake_acq.default_dir, sb.data_base_dir)
|
||||
assert os.path.isdir(expected_data_dir)
|
||||
assert not os.path.exists(os.path.join(expected_data_dir, "scan_fake"))
|
||||
|
||||
# Test case 3: FakeAcquisition only, make_scan_sub_dir=True
|
||||
fake_acq2 = FakeAcquisition("non_sfdaq2", "p2")
|
||||
fake_acq2.default_dir = str(tmp_path / "fake_default2")
|
||||
sb = ScanBackend(
|
||||
adjs, [[1]], [fake_acq2],
|
||||
filename="scan_fake_subdir",
|
||||
detectors=[], channels=[], pvs=[],
|
||||
n_pulses=1, data_base_dir="data", scan_info_dir=tmp_path,
|
||||
make_scan_sub_dir=True, condition=None,
|
||||
return_to_initial_values=True, n_repeat=1,
|
||||
sensor=None, remote_plot=None
|
||||
)
|
||||
|
||||
sb.create_output_dirs()
|
||||
expected_subdir = os.path.join(fake_acq2.default_dir, sb.data_base_dir, "scan_fake_subdir")
|
||||
assert os.path.isdir(expected_subdir)
|
||||
|
||||
# Test case 4: Mixed SFDAQ + FakeAcquisition, make_scan_sub_dir=False
|
||||
fake_acq3 = FakeAcquisition("mix_non_sfdaq", "p3")
|
||||
fake_acq3.default_dir = str(tmp_path / "mix_default")
|
||||
sb = ScanBackend(
|
||||
adjs, [[1]], [sfdaq_acq, fake_acq3],
|
||||
filename="scan_mix_no_subdir",
|
||||
detectors=[], channels=[], pvs=[],
|
||||
n_pulses=1, data_base_dir="data", scan_info_dir=tmp_path,
|
||||
make_scan_sub_dir=False, condition=None,
|
||||
return_to_initial_values=True, n_repeat=1,
|
||||
sensor=None, remote_plot=None
|
||||
)
|
||||
|
||||
sb.create_output_dirs()
|
||||
expected_data_dir_no_sub = os.path.join(fake_acq3.default_dir, sb.data_base_dir)
|
||||
assert os.path.isdir(expected_data_dir_no_sub)
|
||||
assert not os.path.exists(os.path.join(expected_data_dir_no_sub, "scan_mix_no_subdir"))
|
||||
|
||||
# Test case 5: Mixed SFDAQ + FakeAcquisition, make_scan_sub_dir=True
|
||||
fake_acq4 = FakeAcquisition("mix_non_sfdaq2", "p4")
|
||||
fake_acq4.default_dir = str(tmp_path / "mix_default2")
|
||||
sb = ScanBackend(
|
||||
adjs, [[1]], [sfdaq_acq, fake_acq4],
|
||||
filename="scan_mix_subdir",
|
||||
detectors=[], channels=[], pvs=[],
|
||||
n_pulses=1, data_base_dir="data", scan_info_dir=tmp_path,
|
||||
make_scan_sub_dir=True, condition=None,
|
||||
return_to_initial_values=True, n_repeat=1,
|
||||
sensor=None, remote_plot=None
|
||||
)
|
||||
|
||||
sb.create_output_dirs()
|
||||
expected_subdir_mix = os.path.join(fake_acq4.default_dir, sb.data_base_dir, "scan_mix_subdir")
|
||||
assert os.path.isdir(expected_subdir_mix)
|
||||
sfdaq_dir = os.path.join("data", "sfdaq")
|
||||
assert not os.path.exists(sfdaq_dir)
|
||||
|
||||
|
||||
def test_store_and_change_initial_values_restores_correctly(tmp_path):
|
||||
# Test that initial values are properly stored and restored
|
||||
adjs = [
|
||||
DummyAdjustable(ID="A", initial_value=9, process_time=0),
|
||||
DummyAdjustable(ID="B", initial_value=8, process_time=0)
|
||||
]
|
||||
|
||||
sb = ScanBackend(
|
||||
adjs, [[1]], [FakeAcquisition()],
|
||||
filename="fn", detectors=[], channels=[], pvs=[],
|
||||
n_pulses=1, data_base_dir="data", scan_info_dir=tmp_path,
|
||||
make_scan_sub_dir=True, condition=None,
|
||||
return_to_initial_values=True, n_repeat=1,
|
||||
sensor=None, remote_plot=None
|
||||
)
|
||||
|
||||
# Store initial values
|
||||
sb.store_initial_values()
|
||||
initial_values = [a.get_current_value() for a in adjs]
|
||||
assert initial_values == [9, 8]
|
||||
assert sb.initial_values == [9, 8]
|
||||
|
||||
# Change values
|
||||
for a, new_val in zip(adjs, [100, 200]):
|
||||
a.set_target_value(new_val)
|
||||
|
||||
changed_values = [a.get_current_value() for a in adjs]
|
||||
assert changed_values == [100, 200]
|
||||
|
||||
# Restore initial values
|
||||
sb.change_to_initial_values()
|
||||
|
||||
restored_values = [a.get_current_value() for a in adjs]
|
||||
assert restored_values == [9, 8]
|
||||
|
||||
|
||||
def test_acquire_all_with_fake_acquisitions(tmp_path):
|
||||
# Test acquisition with multiple fake acquisitions
|
||||
adjs = [DummyAdjustable(name="A", ID="A")]
|
||||
|
||||
fake_acq1 = FakeAcquisition(name="fake_acq1")
|
||||
fake_acq2 = FakeAcquisition(name="fake_acq2")
|
||||
acqs = [fake_acq1, fake_acq2]
|
||||
|
||||
sb = ScanBackend(
|
||||
adjs, [[1]], acqs, "test_scan",
|
||||
["detector1"], ["bs_channel1"], ["pv1"], 3,
|
||||
"data", tmp_path, True, None, True, 1, None, None
|
||||
)
|
||||
|
||||
# Test acquire_all
|
||||
filenames = sb.acquire_all("test_filename")
|
||||
|
||||
# Verify tasks were created and stored
|
||||
assert hasattr(sb, 'current_tasks')
|
||||
assert len(sb.current_tasks) == 2
|
||||
|
||||
# Verify all tasks completed
|
||||
assert all(t.done for t in sb.current_tasks)
|
||||
|
||||
# Verify each fake acquisition was called with correct parameters
|
||||
for acq in acqs:
|
||||
assert acq.acquire_called
|
||||
assert acq.last_filename == "test_filename"
|
||||
assert acq.last_data_base_dir == "data"
|
||||
assert acq.last_channels == ["bs_channel1"]
|
||||
assert acq.last_n_pulses == 3
|
||||
assert acq.last_wait == False
|
||||
|
||||
# Verify filenames returned
|
||||
assert len(filenames) >= 2
|
||||
assert all(isinstance(fname, str) for fname in filenames)
|
||||
assert all(len(fname) > 0 for fname in filenames)
|
||||
|
||||
# Test stop cleans up properly
|
||||
sb.stop()
|
||||
assert not sb.running
|
||||
assert sb.current_tasks == []
|
||||
|
||||
|
||||
def test_do_step_with_fake_acquisitions(tmp_path):
|
||||
# Test individual scan step execution
|
||||
adjs = [DummyAdjustable(name="motor1", ID="M1"), DummyAdjustable(name="motor2", ID="M2")]
|
||||
|
||||
fake_acq = FakeAcquisition(name="fake_acq")
|
||||
|
||||
sb = ScanBackend(
|
||||
adjs, [[1, 2], [3, 4]], [fake_acq], "test_scan",
|
||||
[], ["bs_channel"], [], 1,
|
||||
"data", tmp_path, True, None, True, 1, None, None
|
||||
)
|
||||
|
||||
# Test do_step for first step
|
||||
step_values = [1, 3]
|
||||
n_step = 0
|
||||
|
||||
sb.do_step(n_step, step_values)
|
||||
|
||||
# Verify adjustables were moved to target values
|
||||
current_values = [adj.get_current_value() for adj in adjs]
|
||||
assert current_values == step_values
|
||||
|
||||
# Verify scan_info_sfdaq was updated
|
||||
assert hasattr(sb.scan_info_sfdaq, 'steps')
|
||||
assert len(sb.scan_info_sfdaq.steps) > 0
|
||||
|
||||
# Verify acquisition was called with correct filename
|
||||
expected_filename = sb.get_filename(n_step)
|
||||
assert fake_acq.last_filename == expected_filename
|
||||
|
||||
# Verify scan_info was updated
|
||||
assert hasattr(sb.scan_info, 'steps')
|
||||
assert len(sb.scan_info.steps) > 0
|
||||
|
||||
|
||||
def test_do_step_with_sensor_and_remote_plot(tmp_path):
|
||||
# Test scan step with sensor and remote plot integration
|
||||
adjs = [DummyAdjustable(name="motor", ID="M1")]
|
||||
|
||||
fake_acq = FakeAcquisition(name="fake_acq")
|
||||
dummy_sensor = DummySensor()
|
||||
dummy_remote_plot = DummyRemotePlot()
|
||||
|
||||
sb = ScanBackend(
|
||||
adjs, [[1, 2, 3]], [fake_acq], "test_scan",
|
||||
[], ["bs_channel"], [], 1,
|
||||
"data", tmp_path, True, None, True, 1, dummy_sensor, dummy_remote_plot
|
||||
)
|
||||
|
||||
step_values = [2]
|
||||
n_step = 1
|
||||
|
||||
sb.do_step(n_step, step_values)
|
||||
|
||||
# Verify sensor was started and stopped
|
||||
assert dummy_sensor.started
|
||||
assert dummy_sensor.stopped
|
||||
|
||||
# Verify remote plot received data
|
||||
assert dummy_remote_plot.appended
|
||||
|
||||
# Verify data sent to plot matches current motor position and sensor reading
|
||||
x_value = adjs[0].get_current_value()
|
||||
y_value = dummy_sensor.get()
|
||||
expected_data = (float(x_value), float(y_value))
|
||||
assert dummy_remote_plot.last_data == expected_data
|
||||
|
||||
|
||||
def test_do_step_multiple_steps(tmp_path):
|
||||
# Test multiple consecutive scan steps
|
||||
adjs = [DummyAdjustable(name="motor", ID="M1")]
|
||||
|
||||
fake_acq = FakeAcquisition(name="fake_acq")
|
||||
|
||||
sb = ScanBackend(
|
||||
adjs, [[1, 2, 3]], [fake_acq], "test_scan",
|
||||
[], ["bs_channel"], [], 1,
|
||||
"data", tmp_path, True, None, True, 1, None, None
|
||||
)
|
||||
|
||||
# Test multiple steps
|
||||
step_sequences = [
|
||||
(0, [1]),
|
||||
(1, [2]),
|
||||
(2, [3])
|
||||
]
|
||||
|
||||
for n_step, step_values in step_sequences:
|
||||
fake_acq.reset()
|
||||
|
||||
sb.do_step(n_step, step_values)
|
||||
|
||||
# Verify motor moved to correct position
|
||||
assert adjs[0].get_current_value() == step_values[0]
|
||||
|
||||
# Verify acquisition was called with correct step filename
|
||||
expected_filename = sb.get_filename(n_step)
|
||||
assert fake_acq.last_filename == expected_filename
|
||||
|
||||
# Verify scan info was updated for each step
|
||||
assert len(sb.scan_info.steps) == n_step + 1
|
||||
|
||||
|
||||
def test_do_checked_step_with_condition_repeats(tmp_path):
|
||||
# Test step execution with condition that requires repeats
|
||||
adjs = [DummyAdjustable(name="motor", ID="M1")]
|
||||
fake_acq = FakeAcquisition(name="fake_acq")
|
||||
|
||||
# Condition that wants 2 repeats
|
||||
condition = DummyCondition(repeats=2)
|
||||
|
||||
sb = ScanBackend(
|
||||
adjs, [[1, 2]], [fake_acq], "test_scan",
|
||||
[], ["bs_channel"], [], 1,
|
||||
"data", tmp_path, True, condition, True, 1, None, None
|
||||
)
|
||||
|
||||
step_values = [1]
|
||||
n_step = 0
|
||||
|
||||
# Track do_step calls directly
|
||||
do_step_call_count = 0
|
||||
original_do_step = sb.do_step
|
||||
|
||||
def mock_do_step(*args, **kwargs):
|
||||
nonlocal do_step_call_count
|
||||
do_step_call_count += 1
|
||||
return original_do_step(*args, **kwargs)
|
||||
|
||||
sb.do_step = mock_do_step
|
||||
sb.running = True
|
||||
|
||||
sb.do_checked_step(n_step, step_values)
|
||||
|
||||
# Verify do_step was called 3 times (original + 2 repeats)
|
||||
assert do_step_call_count == 3
|
||||
|
||||
# Verify condition was checked
|
||||
assert condition.repeats == -1
|
||||
|
||||
|
||||
def test_make_summary_and_repr(tmp_path):
|
||||
# Test summary generation and string representation
|
||||
adjs = [DummyAdjustable(name="A", ID="A")]
|
||||
sb = ScanBackend(
|
||||
adjs, [[1, 2]], [FakeAcquisition()],
|
||||
"fn", [], [], [], 2,
|
||||
"data", tmp_path, True, None, True, 2, None, None
|
||||
)
|
||||
s = sb._make_summary()
|
||||
assert "record" in s and "pulse" in s
|
||||
assert isinstance(repr(sb), str)
|
||||
|
||||
|
||||
def test_make_summary_single_repeat(tmp_path):
|
||||
# Test summary generation with single repeat
|
||||
adjs = [DummyAdjustable(name="motor1", ID="M1"), DummyAdjustable(name="motor2", ID="M2")]
|
||||
fake_acq = FakeAcquisition(name="fake_daq")
|
||||
|
||||
sb = ScanBackend(
|
||||
adjs, [[1, 2], [3, 4]], [fake_acq], "test_scan",
|
||||
[], ["bs_channel"], [], 5,
|
||||
"data", tmp_path, True, None, True, 1,
|
||||
None, None
|
||||
)
|
||||
|
||||
summary = sb._make_summary()
|
||||
|
||||
# Verify single repeat wording
|
||||
assert "perform the following scan" in summary
|
||||
|
||||
# Verify adjustable names are included
|
||||
assert "motor1" in summary
|
||||
assert "motor2" in summary
|
||||
|
||||
# Verify pulse count
|
||||
assert "5 pulses" in summary
|
||||
|
||||
# Verify filename
|
||||
assert "test_scan" in summary
|
||||
|
||||
# Verify acquisition is mentioned
|
||||
assert "fake_daq" in summary
|
||||
|
||||
|
||||
def test_make_summary_multiple_repeats(tmp_path):
|
||||
# Test summary generation with multiple repeats
|
||||
adjs = [DummyAdjustable(name="motor", ID="M1")]
|
||||
fake_acq = FakeAcquisition(name="daq_system")
|
||||
|
||||
sb = ScanBackend(
|
||||
adjs, [[1]], [fake_acq], "multi_scan",
|
||||
[], ["bs_channel"], [], 1,
|
||||
"data", tmp_path, True, None, True, 3,
|
||||
None, None
|
||||
)
|
||||
|
||||
summary = sb._make_summary()
|
||||
|
||||
# Verify multiple repeats wording
|
||||
assert "repeat the following scan 3 times" in summary
|
||||
|
||||
# Verify pulse count
|
||||
assert "1 pulse" in summary
|
||||
|
||||
# Verify filename
|
||||
assert "multi_scan" in summary
|
||||
|
||||
|
||||
def test_scan_loop_fake_only(tmp_path, capsys):
|
||||
# Test complete scan loop execution with fake acquisition
|
||||
adjs = [
|
||||
DummyAdjustable(name="A", ID="A", initial_value=0),
|
||||
DummyAdjustable(name="B", ID="B", initial_value=0),
|
||||
]
|
||||
|
||||
fake = FakeAcquisition()
|
||||
|
||||
values = [[1, 10], [2, 20], [3, 30]]
|
||||
|
||||
sb = ScanBackend(
|
||||
adjs, values, [fake], "scan1",
|
||||
[], ["ch"], [], 1,
|
||||
"data", tmp_path, False,
|
||||
condition=None, return_to_initial_values=True, n_repeat=1,
|
||||
sensor=None, remote_plot=None
|
||||
)
|
||||
|
||||
sb.running = True
|
||||
sb.scan_loop()
|
||||
|
||||
out = capsys.readouterr().out
|
||||
|
||||
# Verify steps printed
|
||||
assert "Scan step 1 of 3" in out
|
||||
assert "Scan step 2 of 3" in out
|
||||
assert "Scan step 3 of 3" in out
|
||||
assert "All scan steps done" in out
|
||||
|
||||
# Verify acquisitions done
|
||||
assert fake.call_count == 3
|
||||
|
||||
# Verify final positions
|
||||
assert adjs[0].get_current_value() == 3
|
||||
assert adjs[1].get_current_value() == 30
|
||||
|
||||
|
||||
def test_repeated_scan_loop_fake_only(tmp_path, capsys):
|
||||
# Test repeated scan loop execution
|
||||
adjs = [DummyAdjustable(name="A", ID="A")]
|
||||
fake = FakeAcquisition()
|
||||
|
||||
values = [[1], [2]]
|
||||
|
||||
sb = ScanBackend(
|
||||
adjs, values, [fake], "rscan",
|
||||
[], ["ch"], [], 1,
|
||||
"data", tmp_path, False,
|
||||
condition=None,
|
||||
return_to_initial_values=True, n_repeat=3,
|
||||
sensor=None, remote_plot=None
|
||||
)
|
||||
|
||||
sb.running = True
|
||||
sb.repeated_scan_loop()
|
||||
|
||||
out = capsys.readouterr().out
|
||||
|
||||
assert "Repetition 1 of 3" in out
|
||||
assert "Repetition 2 of 3" in out
|
||||
assert "Repetition 3 of 3" in out
|
||||
|
||||
# Verify fake acquisitions: 3 reps * 2 steps = 6
|
||||
assert fake.call_count == 6
|
||||
|
||||
# Verify filename restored
|
||||
assert sb.filename == "rscan"
|
||||
|
||||
|
||||
def test_full_multidimensional_scan_end_to_end(tmp_path):
|
||||
# Comprehensive test of multi-dimensional scan with verification of all components
|
||||
d = 4 # scan dimension
|
||||
n_steps = 5 # number of steps
|
||||
|
||||
# Create d adjustables
|
||||
adjs = [
|
||||
DummyAdjustable(name=f"M{i}", ID=f"ID{i}", initial_value=0, process_time=0)
|
||||
for i in range(d)
|
||||
]
|
||||
|
||||
# Create values
|
||||
values = [
|
||||
list(range(t, t + d))
|
||||
for t in range(1, n_steps + 1)
|
||||
]
|
||||
|
||||
# Create fake acquisition
|
||||
fake = FakeAcquisition(name="fake_master")
|
||||
|
||||
# Create ScanBackend
|
||||
sb = ScanBackend(
|
||||
adjs, values, [fake],
|
||||
filename="multidim_test",
|
||||
detectors=[], channels=["ch"], pvs=[],
|
||||
n_pulses=3,
|
||||
data_base_dir="data",
|
||||
scan_info_dir=tmp_path,
|
||||
make_scan_sub_dir=True,
|
||||
condition=None,
|
||||
return_to_initial_values=True,
|
||||
n_repeat=1,
|
||||
sensor=None,
|
||||
remote_plot=None
|
||||
)
|
||||
|
||||
# Execute scan
|
||||
sb.run()
|
||||
|
||||
# Verification checks
|
||||
|
||||
# 1. All acquisitions done
|
||||
assert fake.call_count == n_steps
|
||||
|
||||
# 2. Adjustables returned to initial values
|
||||
assert [a.get_current_value() for a in adjs] == [0] * d
|
||||
|
||||
# 3. Scan info has correct number of steps
|
||||
assert len(sb.scan_info.steps) == n_steps
|
||||
assert len(sb.scan_info_sfdaq.steps) == n_steps
|
||||
|
||||
# 4. Each step has correct values
|
||||
for i in range(n_steps):
|
||||
step_values = sb.scan_info.steps[i]["target_values"]
|
||||
assert step_values == values[i]
|
||||
|
||||
# 5. Filenames are coherent
|
||||
base = sb.filename
|
||||
filebase = os.path.basename(base)
|
||||
for i in range(n_steps):
|
||||
expected = os.path.join(base, filebase + f"_step{i:04d}")
|
||||
assert expected.endswith(f"{filebase}_step{i:04d}")
|
||||
|
||||
# 6. Directories created
|
||||
data_root = tmp_path / "fake_master" / "data"
|
||||
assert data_root.exists()
|
||||
|
||||
expected_subfolder = data_root / "multidim_test"
|
||||
assert expected_subfolder.exists()
|
||||
|
||||
# 7. All tasks finished
|
||||
assert all(t.done for t in sb.current_tasks)
|
||||
|
||||
# 8. Each task has at least one file
|
||||
all_files = []
|
||||
for t in sb.current_tasks:
|
||||
assert len(t.filenames) > 0
|
||||
all_files += t.filenames
|
||||
|
||||
assert len(all_files) >= n_steps
|
||||
|
||||
# 9. Intermediate values correctly positioned
|
||||
for i in range(n_steps):
|
||||
target = values[i]
|
||||
readback = sb.scan_info_sfdaq.steps[i]["readback_values"]
|
||||
assert readback == target
|
||||
|
||||
# 10. Step order respected
|
||||
assert sb.scan_info.steps[0]["target_values"] == values[0]
|
||||
assert sb.scan_info.steps[-1]["target_values"] == values[-1]
|
||||
|
||||
|
||||
def test_scanND_relative_positions_only(tmp_path):
|
||||
# Test relative position calculation for N-dimensional scans
|
||||
d = 5
|
||||
|
||||
# Create adjustables with distinct initial values
|
||||
initial_values = [10 * (i + 1) for i in range(d)]
|
||||
adjustables = [
|
||||
DummyAdjustable(name=f"M{i}", ID=f"ID{i}", initial_value=initial_values[i], process_time=0)
|
||||
for i in range(d)
|
||||
]
|
||||
|
||||
# Create N-dim grid
|
||||
positions_per_dim = [list(range(-1, 2))] * d
|
||||
|
||||
fake = FakeAcquisition(name="fake")
|
||||
|
||||
# Create ScanBackend
|
||||
sb = ScanBackend(
|
||||
adjustables,
|
||||
values=positions_per_dim,
|
||||
acquisitions=[fake],
|
||||
filename="relND",
|
||||
detectors=[], channels=["ch"], pvs=[],
|
||||
n_pulses=1,
|
||||
data_base_dir="data",
|
||||
scan_info_dir=tmp_path,
|
||||
make_scan_sub_dir=False,
|
||||
condition=None,
|
||||
return_to_initial_values=True,
|
||||
n_repeat=1,
|
||||
sensor=None, remote_plot=None
|
||||
)
|
||||
|
||||
# Simulate relative position calculation
|
||||
offset_values = [
|
||||
[p + initial_values[i] for p in [-1, 0, 1]]
|
||||
for i in range(d)
|
||||
]
|
||||
|
||||
# Apply relative offset
|
||||
sb.values = offset_values
|
||||
|
||||
# Verification
|
||||
|
||||
# Check each dimension is correctly offset
|
||||
for i in range(d):
|
||||
expected = [
|
||||
initial_values[i] - 1,
|
||||
initial_values[i],
|
||||
initial_values[i] + 1,
|
||||
]
|
||||
assert sb.values[i] == expected
|
||||
|
||||
# Verify dimension count
|
||||
assert len(sb.values) == d
|
||||
|
||||
# Verify each dimension has 3 positions
|
||||
assert all(len(axis) == 3 for axis in sb.values)
|
||||
|
||||
|
||||
# Utility function tests
|
||||
|
||||
def test_print_current_values_displays_correct_output(capsys):
|
||||
# Test current values printing functionality
|
||||
class Obj:
|
||||
def __init__(self, adjustables):
|
||||
self.adjustables = adjustables
|
||||
|
||||
def print_current_values(self):
|
||||
print_all_current_values(self.adjustables)
|
||||
|
||||
adjs = [
|
||||
DummyAdjustable(name="MotorA", ID="A1", value=10),
|
||||
DummyAdjustable(name="MotorB", ID="B2", value=20),
|
||||
]
|
||||
|
||||
obj = Obj(adjs)
|
||||
obj.print_current_values()
|
||||
|
||||
captured = capsys.readouterr().out
|
||||
|
||||
assert "Current values" in captured
|
||||
assert "A1" in captured and "B2" in captured
|
||||
assert "10" in captured and "20" in captured
|
||||
|
||||
|
||||
def test_get_all_current_values_returns_correct_list():
|
||||
# Test current values retrieval
|
||||
adjs = [
|
||||
DummyAdjustable(name="M1", ID="M1", value=5),
|
||||
DummyAdjustable(name="M2", ID="M2", value=15),
|
||||
]
|
||||
values = get_all_current_values(adjs)
|
||||
assert values == [5, 15]
|
||||
|
||||
|
||||
def test_set_all_target_values_and_wait_full_chain(monkeypatch):
|
||||
# Test complete chain of setting target values and waiting
|
||||
adjs = [
|
||||
DummyAdjustable(ID="1", initial_value=0, process_time=0),
|
||||
DummyAdjustable(ID="2", initial_value=0, process_time=0),
|
||||
]
|
||||
|
||||
called = []
|
||||
|
||||
class DummyTask:
|
||||
def __init__(self, name):
|
||||
self.name = name
|
||||
def wait(self):
|
||||
called.append(self.name)
|
||||
|
||||
# Monkeypatch set_all_target_values
|
||||
def fake_set_all_target_values(adjs, values):
|
||||
for adj, v in zip(adjs, values):
|
||||
adj.set_target_value(v)
|
||||
return [DummyTask(f"task_{a.ID}") for a in adjs]
|
||||
|
||||
monkeypatch.setattr("slic.core.scanner.scanbackend.set_all_target_values", fake_set_all_target_values)
|
||||
|
||||
# Execute
|
||||
tasks = set_all_target_values(adjs, [10, 20])
|
||||
wait_for_all(tasks)
|
||||
|
||||
# Verify values were updated
|
||||
values = [a.get_current_value() for a in adjs]
|
||||
assert values == [10, 20]
|
||||
|
||||
# Verify all tasks were waited for
|
||||
assert called == ["task_1", "task_2"]
|
||||
|
||||
# Verify returned objects have wait method
|
||||
assert all(hasattr(t, "wait") for t in tasks)
|
||||
|
||||
|
||||
def test_wait_for_all_calls_wait_on_all_tasks(monkeypatch):
|
||||
# Test wait functionality on all tasks
|
||||
called = []
|
||||
|
||||
class DummyTask:
|
||||
def __init__(self, name):
|
||||
self.name = name
|
||||
def wait(self):
|
||||
called.append(self.name)
|
||||
|
||||
tasks = [DummyTask("t1"), DummyTask("t2"), DummyTask("t3")]
|
||||
|
||||
wait_for_all(tasks)
|
||||
|
||||
assert called == ["t1", "t2", "t3"]
|
||||
|
||||
|
||||
def test_stop_all_calls_stop_and_handles_exceptions(capsys):
|
||||
# Test stop functionality with exception handling
|
||||
called = []
|
||||
|
||||
class WorkingTask:
|
||||
def __init__(self, name):
|
||||
self.name = name
|
||||
self.stopped = False
|
||||
def stop(self):
|
||||
self.stopped = True
|
||||
called.append(self.name)
|
||||
|
||||
class FailingTask:
|
||||
def stop(self):
|
||||
raise RuntimeError("boom")
|
||||
|
||||
tasks = [WorkingTask("T1"), FailingTask(), WorkingTask("T2")]
|
||||
|
||||
stop_all(tasks)
|
||||
|
||||
# Verify normal tasks were stopped
|
||||
assert all(t.stopped for t in tasks if isinstance(t, WorkingTask))
|
||||
|
||||
# Verify task names recorded
|
||||
assert called == ["T1", "T2"]
|
||||
|
||||
# Verify error was handled and printed
|
||||
out = capsys.readouterr().out
|
||||
assert "Stopping caused" in out
|
||||
assert "boom" in out
|
||||
Reference in New Issue
Block a user