1028 lines
30 KiB
Python
1028 lines
30 KiB
Python
import pytest
|
|
from slic.core.adjustable.baseadjustable import BaseAdjustable
|
|
from slic.core.adjustable.dummyadjustable import DummyAdjustable
|
|
from slic.core.adjustable.genericadjustable import GenericAdjustable
|
|
from slic.core.adjustable.converted import Converted
|
|
from slic.core.adjustable.scaler import Scaler
|
|
from slic.core.adjustable.combined import Combined
|
|
from slic.core.adjustable.linked import Linked
|
|
from slic.core.adjustable.collection import Collection
|
|
|
|
|
|
# BaseAdjustable
|
|
|
|
def test_baseadjustable_cannot_instantiate():
|
|
with pytest.raises(TypeError):
|
|
BaseAdjustable()
|
|
|
|
|
|
def test_baseadjustable_missing_methods():
|
|
class IncompleteAdj(BaseAdjustable):
|
|
pass
|
|
|
|
with pytest.raises(TypeError):
|
|
IncompleteAdj()
|
|
|
|
|
|
def test_baseadjustable_working_subclass():
|
|
class WorkingAdj(BaseAdjustable):
|
|
def __init__(self):
|
|
self.value = 0
|
|
|
|
def get_current_value(self):
|
|
return self.value
|
|
|
|
def set_target_value(self, value):
|
|
self.value = value
|
|
|
|
def is_moving(self):
|
|
return False
|
|
|
|
adj = WorkingAdj()
|
|
assert adj.get_current_value() == 0
|
|
adj.set_target_value(42)
|
|
assert adj.get_current_value() == 42
|
|
assert adj.is_moving() == False
|
|
|
|
|
|
# DummyAdjustable
|
|
|
|
def test_dummyadjustable_basic():
|
|
adj = DummyAdjustable(name="TestAdj", ID="test_id")
|
|
assert adj.name == "TestAdj"
|
|
assert adj.ID == "test_id"
|
|
assert adj.get_current_value() == 0
|
|
|
|
|
|
def test_dummyadjustable_set_get_with_wait():
|
|
adj = DummyAdjustable(name="TestAdj", ID="test_id")
|
|
task = adj.set_target_value(100)
|
|
task.wait()
|
|
assert adj.get_current_value() == 100
|
|
|
|
task = adj.set_target_value(-50.5)
|
|
task.wait()
|
|
assert adj.get_current_value() == -50.5
|
|
|
|
def test_dummyadjustable_is_moving():
|
|
adj = DummyAdjustable(name="TestAdj", ID="test_id")
|
|
assert adj.is_moving() == False
|
|
|
|
def test_dummyadjustable_initial_value():
|
|
adj = DummyAdjustable(name="TestAdj", ID="test_id", initial_value=42)
|
|
assert adj.get_current_value() == 42
|
|
|
|
def test_dummyadjustable_float_values():
|
|
adj = DummyAdjustable(name="TestAdj", ID="test_id")
|
|
task = adj.set_target_value(3.14159)
|
|
task.wait()
|
|
assert adj.get_current_value() == 3.14159
|
|
|
|
def test_dummyadjustable_large_values():
|
|
adj = DummyAdjustable(name="TestAdj", ID="test_id")
|
|
task = adj.set_target_value(1e10)
|
|
task.wait()
|
|
assert adj.get_current_value() == 1e10
|
|
|
|
@pytest.mark.parametrize("process_time,target", [
|
|
(0.1, 100),
|
|
(0.2, 50),
|
|
(0.15, 200),
|
|
(0.3, -100),
|
|
])
|
|
def test_dummyadjustable_process_time(process_time, target):
|
|
import time
|
|
adj = DummyAdjustable(name="TestAdj", ID="test_id", process_time=process_time)
|
|
|
|
start = time.time()
|
|
task = adj.set_target_value(target)
|
|
task.wait()
|
|
elapsed = time.time() - start
|
|
|
|
assert elapsed >= process_time * 0.9, f"Too fast: {elapsed:.3f}s < {process_time*0.9:.3f}s"
|
|
assert elapsed <= process_time * 1.5, f"Too slow: {elapsed:.3f}s > {process_time*1.5:.3f}s"
|
|
assert adj.get_current_value() == target
|
|
|
|
|
|
@pytest.mark.parametrize("process_time,initial,target", [
|
|
(0.2, 0, 100),
|
|
(0.3, 0, 200),
|
|
(0.4, 0, 50),
|
|
(0.25, 0, -100),
|
|
(0.3, 100, -50),
|
|
(0.35, -50, -150),
|
|
])
|
|
def test_dummyadjustable_process_time_progressive_values(process_time, initial, target):
|
|
# Test that DummyAdjustable progresses gradually through intermediate values
|
|
import time
|
|
import threading
|
|
|
|
adj = DummyAdjustable(name="TestAdj", ID="test_id", process_time=process_time, initial_value=initial)
|
|
|
|
intermediate_data = []
|
|
stop_collecting = threading.Event()
|
|
start_time = time.time()
|
|
|
|
def collect_values():
|
|
while not stop_collecting.is_set():
|
|
current_time = time.time() - start_time
|
|
current_value = adj.get_current_value()
|
|
intermediate_data.append((current_time, current_value))
|
|
time.sleep(0.02)
|
|
|
|
collector_thread = threading.Thread(target=collect_values)
|
|
collector_thread.start()
|
|
|
|
task = adj.set_target_value(target)
|
|
task.wait()
|
|
|
|
stop_collecting.set()
|
|
collector_thread.join()
|
|
|
|
intermediate_values = [v for t, v in intermediate_data]
|
|
|
|
assert len(intermediate_values) >= 3, f"Expected at least 3 samples, got {len(intermediate_values)}"
|
|
|
|
distance = target - initial
|
|
|
|
if target > initial:
|
|
increasing_count = sum(1 for i in range(1, len(intermediate_values))
|
|
if intermediate_values[i] >= intermediate_values[i-1])
|
|
assert increasing_count >= len(intermediate_values) * 0.7, \
|
|
f"Expected mostly increasing values, got {increasing_count}/{len(intermediate_values)-1}"
|
|
elif target < initial:
|
|
decreasing_count = sum(1 for i in range(1, len(intermediate_values))
|
|
if intermediate_values[i] <= intermediate_values[i-1])
|
|
assert decreasing_count >= len(intermediate_values) * 0.7, \
|
|
f"Expected mostly decreasing values, got {decreasing_count}/{len(intermediate_values)-1}"
|
|
|
|
distances_to_target = [abs(value - target) for value in intermediate_values]
|
|
|
|
decreasing_distance_count = sum(1 for i in range(1, len(distances_to_target))
|
|
if distances_to_target[i] <= distances_to_target[i-1])
|
|
|
|
assert decreasing_distance_count >= len(distances_to_target) * 0.7, \
|
|
f"Expected values to approach target, got {decreasing_distance_count}/{len(distances_to_target)-1} decreasing distances. " \
|
|
f"Distances: {distances_to_target[:10]}..."
|
|
|
|
assert adj.get_current_value() == target, f"Final value {adj.get_current_value()} != target {target}"
|
|
|
|
unique_values = sorted(set(intermediate_values))
|
|
assert len(unique_values) >= 2, f"Expected progressive motion, got values: {unique_values}"
|
|
|
|
min_val = min(initial, target)
|
|
max_val = max(initial, target)
|
|
for i, value in enumerate(intermediate_values):
|
|
assert min_val <= value <= max_val, \
|
|
f"Value {value} at index {i} outside range [{min_val}, {max_val}]"
|
|
|
|
tolerance_factor = 0.3
|
|
for timestamp, value in intermediate_data[1:-1]:
|
|
progress_ratio = timestamp / process_time
|
|
expected_value = initial + distance * progress_ratio
|
|
|
|
value_range = abs(distance) * tolerance_factor
|
|
lower_bound = expected_value - value_range
|
|
upper_bound = expected_value + value_range
|
|
|
|
if not (lower_bound <= value <= upper_bound):
|
|
pass
|
|
|
|
|
|
@pytest.mark.parametrize("jitter", [
|
|
1,
|
|
5,
|
|
10,
|
|
20,
|
|
0.5,
|
|
])
|
|
def test_dummyadjustable_jitter(jitter):
|
|
initial_value = 100
|
|
adj = DummyAdjustable(name="TestAdj", ID="test_id", initial_value=initial_value, jitter=jitter)
|
|
|
|
num_samples = 20
|
|
readings = [adj.get_current_value() for _ in range(num_samples)]
|
|
|
|
unique_readings = set(readings)
|
|
assert len(unique_readings) > 1, f"Expected variation with jitter={jitter}, got all same value"
|
|
|
|
min_expected = initial_value - jitter
|
|
max_expected = initial_value + jitter
|
|
|
|
for reading in readings:
|
|
assert min_expected <= reading <= max_expected, \
|
|
f"Reading {reading} outside expected range [{min_expected}, {max_expected}] for jitter={jitter}"
|
|
|
|
import statistics
|
|
std_dev = statistics.stdev(readings)
|
|
|
|
assert std_dev >= jitter * 0.2, \
|
|
f"Standard deviation {std_dev:.2f} too small for jitter={jitter} (expected >= {jitter*0.2:.2f})"
|
|
assert std_dev <= jitter * 0.8, \
|
|
f"Standard deviation {std_dev:.2f} too large for jitter={jitter} (expected <= {jitter*0.8:.2f})"
|
|
|
|
|
|
def test_dummyadjustable_stop():
|
|
import threading
|
|
adj = DummyAdjustable(name="TestAdj", ID="test_id", process_time=1.0)
|
|
|
|
def move():
|
|
adj.set_target_value(100)
|
|
|
|
thread = threading.Thread(target=move)
|
|
thread.start()
|
|
|
|
import time
|
|
time.sleep(0.1)
|
|
|
|
adj.stop()
|
|
thread.join()
|
|
|
|
assert adj.get_current_value() < 100
|
|
|
|
|
|
# GenericAdjustable
|
|
|
|
def test_genericadjustable_with_callbacks():
|
|
storage = {"value": 0}
|
|
|
|
def getter():
|
|
return storage["value"]
|
|
|
|
def setter(val):
|
|
storage["value"] = val
|
|
|
|
adj = GenericAdjustable(ID="gen_id",
|
|
get = getter,
|
|
set =setter,
|
|
name="GenAdj")
|
|
|
|
assert adj.get_current_value() == 0
|
|
task = adj.set_target_value(42)
|
|
assert adj._last_target == 42
|
|
task.wait()
|
|
assert adj.get_current_value() == 42
|
|
assert storage["value"] == 42
|
|
|
|
|
|
def test_genericadjustable_with_wait_callback():
|
|
# The wait callback should return True when motion is COMPLETE (not moving)
|
|
# and False when still moving. is_moving() returns not wait().
|
|
motion_complete = {"complete": True}
|
|
|
|
def getter():
|
|
return 0
|
|
|
|
def setter(val):
|
|
pass
|
|
|
|
def wait_func():
|
|
return motion_complete["complete"]
|
|
|
|
adj = GenericAdjustable(ID="gen_id",
|
|
get=getter,
|
|
set=setter,
|
|
wait=wait_func,
|
|
name="GenAdj")
|
|
|
|
assert adj.is_moving() == False
|
|
|
|
motion_complete["complete"] = False
|
|
assert adj.is_moving() == True
|
|
|
|
motion_complete["complete"] = True
|
|
assert adj.is_moving() == False
|
|
|
|
|
|
# Converted
|
|
|
|
@pytest.mark.parametrize("scale_factor,test_value", [
|
|
(10, 100),
|
|
(2, 50),
|
|
(5, 200),
|
|
(0.5, 25),
|
|
(100, 1000),
|
|
])
|
|
def test_converted_with_scaling(scale_factor, test_value):
|
|
base = DummyAdjustable(name="Base", ID="base_id")
|
|
|
|
converted = Converted("scaled_id", base,
|
|
conv_get=lambda x: x * scale_factor,
|
|
conv_set=lambda x: x / scale_factor,
|
|
name="Scaled")
|
|
|
|
assert converted.get_current_value() == 0
|
|
|
|
converted.set_target_value(test_value).wait()
|
|
expected_base = test_value / scale_factor
|
|
assert abs(base.get_current_value() - expected_base) < 0.0001
|
|
assert abs(converted.get_current_value() - test_value) < 0.0001
|
|
|
|
|
|
@pytest.mark.parametrize("offset,test_value", [
|
|
(50, 100),
|
|
(10, 30),
|
|
(-20, 50),
|
|
(100, 200),
|
|
(0, 42),
|
|
])
|
|
def test_converted_with_offset(offset, test_value):
|
|
base = DummyAdjustable(name="Base", ID="base_id")
|
|
|
|
converted = Converted("offset_id", base,
|
|
conv_get=lambda x: x + offset,
|
|
conv_set=lambda x: x - offset,
|
|
name="Offset")
|
|
|
|
assert converted.get_current_value() == offset
|
|
|
|
converted.set_target_value(test_value).wait()
|
|
assert base.get_current_value() == test_value - offset
|
|
assert converted.get_current_value() == test_value
|
|
|
|
|
|
@pytest.mark.parametrize("scale,offset,base_val,expected_conv", [
|
|
(10, 50, 0, 50),
|
|
(10, 50, 5, 100),
|
|
(2, 10, 20, 50),
|
|
(5, -15, 3, 0),
|
|
(0.5, 100, 40, 120),
|
|
])
|
|
def test_converted_with_scaling_and_offset(scale, offset, base_val, expected_conv):
|
|
base = DummyAdjustable(name="Base", ID="base_id")
|
|
|
|
converted = Converted("both_id", base,
|
|
conv_get=lambda x: x * scale + offset,
|
|
conv_set=lambda x: (x - offset) / scale,
|
|
name="Both")
|
|
|
|
assert converted.get_current_value() == offset
|
|
|
|
base.set_target_value(base_val).wait()
|
|
assert abs(converted.get_current_value() - expected_conv) < 0.0001
|
|
|
|
converted.set_target_value(expected_conv).wait()
|
|
assert abs(base.get_current_value() - base_val) < 0.0001
|
|
assert abs(converted.get_current_value() - expected_conv) < 0.0001
|
|
|
|
|
|
def test_converted_units_conversion():
|
|
base_mm = DummyAdjustable(name="Position_mm", ID="pos_mm", units="mm")
|
|
|
|
position_um = Converted("pos_um", base_mm,
|
|
conv_get=lambda x: x * 1000,
|
|
conv_set=lambda x: x / 1000,
|
|
name="Position_μm",
|
|
units="μm")
|
|
|
|
assert position_um.get_current_value() == 0
|
|
|
|
base_mm.set_target_value(1).wait()
|
|
assert position_um.get_current_value() == 1000
|
|
|
|
position_um.set_target_value(2500).wait()
|
|
assert base_mm.get_current_value() == 2.5
|
|
assert position_um.get_current_value() == 2500
|
|
|
|
|
|
def test_converted_negative_scale():
|
|
base = DummyAdjustable(name="Base", ID="base_id")
|
|
|
|
converted = Converted("inv_id", base,
|
|
conv_get=lambda x: x * -1,
|
|
conv_set=lambda x: x * -1,
|
|
name="Inverted")
|
|
|
|
base.set_target_value(10).wait()
|
|
assert converted.get_current_value() == -10
|
|
|
|
converted.set_target_value(20).wait()
|
|
assert base.get_current_value() == -20
|
|
assert converted.get_current_value() == 20
|
|
|
|
|
|
def test_converted_is_moving():
|
|
base = DummyAdjustable(name="Base", ID="base_id")
|
|
converted = Converted("conv_id", base,
|
|
conv_get=lambda x: x * 10,
|
|
conv_set=lambda x: x / 10,
|
|
name="Conv")
|
|
|
|
assert converted.is_moving() == False
|
|
|
|
|
|
def test_converted_only_get_conversion():
|
|
base = DummyAdjustable(name="Base", ID="base_id")
|
|
|
|
converted = Converted("get_only_id", base,
|
|
conv_get=lambda x: x * 10,
|
|
conv_set=lambda x: x,
|
|
name="GetOnly")
|
|
|
|
base.set_target_value(5).wait()
|
|
assert converted.get_current_value() == 50
|
|
assert base.get_current_value() == 5
|
|
|
|
converted.set_target_value(100).wait()
|
|
assert base.get_current_value() == 100
|
|
assert converted.get_current_value() == 1000
|
|
|
|
|
|
def test_converted_only_set_conversion():
|
|
base = DummyAdjustable(name="Base", ID="base_id")
|
|
|
|
converted = Converted("set_only_id", base,
|
|
conv_get=lambda x: x,
|
|
conv_set=lambda x: x / 10,
|
|
name="SetOnly")
|
|
|
|
base.set_target_value(50).wait()
|
|
assert converted.get_current_value() == 50
|
|
assert base.get_current_value() == 50
|
|
|
|
converted.set_target_value(100).wait()
|
|
assert base.get_current_value() == 10
|
|
assert converted.get_current_value() == 10
|
|
|
|
|
|
# Scaler
|
|
|
|
@pytest.mark.parametrize("init1,init2,factor_init,factor_target", [
|
|
(10, 20, 2, 4),
|
|
(5, 15, 1, 3),
|
|
(100, 200, 0.5, 1),
|
|
(8, 16, 4, 2),
|
|
(50, 100, 10, 20),
|
|
])
|
|
def test_scaler_basic(init1, init2, factor_init, factor_target):
|
|
adj1 = DummyAdjustable(name="Adj1", ID="id1", initial_value=init1)
|
|
adj2 = DummyAdjustable(name="Adj2", ID="id2", initial_value=init2)
|
|
|
|
scaler = Scaler("scaler_id", [adj1, adj2], factor=factor_init, name="Scaler")
|
|
|
|
assert scaler.get_current_value() == factor_init
|
|
|
|
scaler.set_target_value(factor_target).wait()
|
|
ratio = factor_target / factor_init
|
|
assert abs(adj1.get_current_value() - init1 * ratio) < 0.0001
|
|
assert abs(adj2.get_current_value() - init2 * ratio) < 0.0001
|
|
assert abs(scaler.get_current_value() - factor_target) < 0.0001
|
|
|
|
|
|
def test_scaler_fractional_factor():
|
|
adj1 = DummyAdjustable(name="Adj1", ID="id1", initial_value=100)
|
|
|
|
scaler = Scaler("scaler_id", [adj1], factor=0.5, name="Half")
|
|
|
|
assert scaler.get_current_value() == 0.5
|
|
|
|
scaler.set_target_value(1.0).wait()
|
|
|
|
assert adj1.get_current_value() == 200
|
|
assert scaler.get_current_value() == 1.0
|
|
|
|
|
|
def test_scaler_is_moving():
|
|
adj1 = DummyAdjustable(name="Adj1", ID="id1")
|
|
adj2 = DummyAdjustable(name="Adj2", ID="id2")
|
|
|
|
scaler = Scaler("scaler_id", [adj1, adj2], factor=1, name="Scaler")
|
|
|
|
assert scaler.is_moving() == False
|
|
|
|
|
|
# Combined
|
|
|
|
def test_combined_two_adjustables():
|
|
adj1 = DummyAdjustable(name="Adj1", ID="id1")
|
|
adj2 = DummyAdjustable(name="Adj2", ID="id2")
|
|
|
|
combined = Combined("comb_id", [adj1, adj2], name="Combined")
|
|
|
|
combined.set_target_value(50).wait()
|
|
|
|
assert adj1.get_current_value() == 50
|
|
assert adj2.get_current_value() == 50
|
|
assert combined.get_current_value() == 50
|
|
|
|
|
|
def test_combined_three_adjustables():
|
|
adj1 = DummyAdjustable(name="Adj1", ID="id1")
|
|
adj2 = DummyAdjustable(name="Adj2", ID="id2")
|
|
adj3 = DummyAdjustable(name="Adj3", ID="id3")
|
|
|
|
combined = Combined("comb3_id", [adj1, adj2, adj3], name="Combined3")
|
|
|
|
combined.set_target_value(100).wait()
|
|
|
|
assert adj1.get_current_value() == 100
|
|
assert adj2.get_current_value() == 100
|
|
assert adj3.get_current_value() == 100
|
|
assert combined.get_current_value() == 100
|
|
|
|
|
|
def test_combined_get_current_value_returns_mean():
|
|
adj1 = DummyAdjustable(name="Adj1", ID="id1", initial_value=10)
|
|
adj2 = DummyAdjustable(name="Adj2", ID="id2", initial_value=20)
|
|
|
|
combined = Combined("comb_id", [adj1, adj2], name="Combined")
|
|
|
|
current = combined.get_current_value()
|
|
assert current == 15.0
|
|
|
|
combined.set_target_value(100).wait()
|
|
assert combined.get_current_value() == 100
|
|
|
|
|
|
def test_combined_mean_with_different_initial_values():
|
|
adj1 = DummyAdjustable(name="Adj1", ID="id1", initial_value=5)
|
|
adj2 = DummyAdjustable(name="Adj2", ID="id2", initial_value=15)
|
|
adj3 = DummyAdjustable(name="Adj3", ID="id3", initial_value=25)
|
|
|
|
combined = Combined("comb_id", [adj1, adj2, adj3], name="Combined")
|
|
|
|
assert combined.get_current_value() == 15.0
|
|
|
|
adj1.set_target_value(10).wait()
|
|
import numpy as np
|
|
assert np.isclose(combined.get_current_value(), 50/3)
|
|
|
|
combined.set_target_value(60).wait()
|
|
assert adj1.get_current_value() == 60
|
|
assert adj2.get_current_value() == 60
|
|
assert adj3.get_current_value() == 60
|
|
assert combined.get_current_value() == 60
|
|
|
|
|
|
def test_combined_is_moving():
|
|
adj1 = DummyAdjustable(name="Adj1", ID="id1")
|
|
adj2 = DummyAdjustable(name="Adj2", ID="id2")
|
|
|
|
combined = Combined("comb_id", [adj1, adj2], name="Combined")
|
|
|
|
assert combined.is_moving() == False
|
|
|
|
|
|
# Linked
|
|
|
|
def test_linked_basic():
|
|
master = DummyAdjustable(name="Master", ID="master_id")
|
|
slave = DummyAdjustable(name="Slave", ID="slave_id")
|
|
|
|
linked = Linked("linked_id", master, slave, name="Linked")
|
|
|
|
linked.set_target_value(10).wait()
|
|
|
|
assert master.get_current_value() == 10
|
|
assert slave.get_current_value() == 10
|
|
|
|
|
|
@pytest.mark.parametrize("scale,target_val,expected_slave", [
|
|
(2, 10, 20),
|
|
(3, 15, 45),
|
|
(0.5, 20, 10),
|
|
(10, 5, 50),
|
|
(-1, 10, -10),
|
|
(-2, 15, -30),
|
|
])
|
|
def test_linked_with_scale(scale, target_val, expected_slave):
|
|
master = DummyAdjustable(name="Master", ID="master_id")
|
|
slave = DummyAdjustable(name="Slave", ID="slave_id")
|
|
|
|
linked = Linked("linked_id", master, slave, scale=scale, name="Linked")
|
|
|
|
linked.set_target_value(target_val).wait()
|
|
|
|
assert master.get_current_value() == target_val
|
|
assert abs(slave.get_current_value() - expected_slave) < 0.0001
|
|
|
|
|
|
@pytest.mark.parametrize("scale,offset,target_val", [
|
|
(1, 50, 10),
|
|
(2, 10, 15),
|
|
(3, -5, 10),
|
|
(0.5, 100, 20),
|
|
(-1, 50, 10),
|
|
(2, 0, 25),
|
|
])
|
|
def test_linked_with_scale_and_offset(scale, offset, target_val):
|
|
master = DummyAdjustable(name="Master", ID="master_id")
|
|
slave = DummyAdjustable(name="Slave", ID="slave_id")
|
|
|
|
linked = Linked("linked_id", master, slave, scale=scale, offset=offset, name="Linked")
|
|
|
|
linked.set_target_value(target_val).wait()
|
|
|
|
assert master.get_current_value() == target_val
|
|
expected_slave = target_val * scale + offset
|
|
assert abs(slave.get_current_value() - expected_slave) < 0.0001
|
|
|
|
|
|
def test_linked_get_current_value():
|
|
master = DummyAdjustable(name="Master", ID="master_id", initial_value=42)
|
|
slave = DummyAdjustable(name="Slave", ID="slave_id", initial_value=100)
|
|
|
|
linked = Linked("linked_id", master, slave, name="Linked")
|
|
|
|
assert linked.get_current_value() == 42
|
|
|
|
|
|
def test_linked_repr():
|
|
master = DummyAdjustable(name="Master", ID="master_id", initial_value=10)
|
|
slave = DummyAdjustable(name="Slave", ID="slave_id", initial_value=20)
|
|
|
|
linked = Linked("linked_id", master, slave, scale=2, offset=5, name="Linked")
|
|
|
|
linked.set_target_value(15).wait()
|
|
|
|
repr_str = repr(linked)
|
|
|
|
assert "Primary:" in repr_str
|
|
assert "Secondary:" in repr_str
|
|
|
|
assert "Master" in repr_str
|
|
assert "Slave" in repr_str
|
|
assert "15" in repr_str
|
|
assert "35" in repr_str
|
|
|
|
|
|
# Collection
|
|
|
|
def test_collection_basic():
|
|
adj1 = DummyAdjustable(name="Adj1", ID="id1")
|
|
adj2 = DummyAdjustable(name="Adj2", ID="id2")
|
|
adj3 = DummyAdjustable(name="Adj3", ID="id3")
|
|
|
|
collection = Collection("coll_id", [adj1, adj2, adj3], name="MyCollection")
|
|
|
|
assert len(collection.adjs) == 3
|
|
|
|
|
|
def test_collection_set_individual_values():
|
|
adj1 = DummyAdjustable(name="Adj1", ID="id1")
|
|
adj2 = DummyAdjustable(name="Adj2", ID="id2")
|
|
|
|
collection = Collection("coll_id", [adj1, adj2], name="MyCollection")
|
|
|
|
collection.set_target_value(10, 20).wait()
|
|
|
|
assert adj1.get_current_value() == 10
|
|
assert adj2.get_current_value() == 20
|
|
|
|
|
|
def test_collection_get_current_value():
|
|
adj1 = DummyAdjustable(name="Adj1", ID="id1", initial_value=42)
|
|
adj2 = DummyAdjustable(name="Adj2", ID="id2", initial_value=84)
|
|
|
|
collection = Collection("coll_id", [adj1, adj2], name="MyCollection")
|
|
|
|
current = collection.get_current_value()
|
|
assert current == (42, 84)
|
|
|
|
|
|
def test_collection_empty():
|
|
collection = Collection("empty_id", [], name="EmptyCollection")
|
|
assert len(collection.adjs) == 0
|
|
|
|
|
|
def test_collection_wrong_number_of_values():
|
|
# BUG: ValueError is wrapped in TaskError due to threading
|
|
from slic.core.task.task import TaskError
|
|
|
|
adj1 = DummyAdjustable(name="Adj1", ID="id1")
|
|
adj2 = DummyAdjustable(name="Adj2", ID="id2")
|
|
|
|
collection = Collection("coll_id", [adj1, adj2], name="MyCollection")
|
|
|
|
with pytest.raises(TaskError, match="ValueError.*number of values.*3.*is not equal.*2"):
|
|
task = collection.set_target_value(10, 20, 30)
|
|
task.wait()
|
|
|
|
with pytest.raises(TaskError, match="ValueError.*number of values.*1.*is not equal.*2"):
|
|
task = collection.set_target_value(10)
|
|
task.wait()
|
|
|
|
|
|
def test_collection_repr():
|
|
adj1 = DummyAdjustable(name="Adj1", ID="id1", initial_value=10)
|
|
adj2 = DummyAdjustable(name="Adj2", ID="id2", initial_value=20)
|
|
adj3 = DummyAdjustable(name="Adj3", ID="id3", initial_value=30)
|
|
|
|
collection = Collection("coll_id", [adj1, adj2, adj3], name="MyCollection")
|
|
|
|
collection.set_target_value(100, 200, 300).wait()
|
|
|
|
repr_str = repr(collection)
|
|
|
|
assert "Adj1" in repr_str
|
|
assert "Adj2" in repr_str
|
|
assert "Adj3" in repr_str
|
|
|
|
assert "100" in repr_str
|
|
assert "200" in repr_str
|
|
assert "300" in repr_str
|
|
|
|
assert "\n" in repr_str
|
|
|
|
|
|
def test_collection_is_moving():
|
|
adj1 = DummyAdjustable(name="Adj1", ID="id1")
|
|
adj2 = DummyAdjustable(name="Adj2", ID="id2")
|
|
|
|
collection = Collection("coll_id", [adj1, adj2], name="MyCollection")
|
|
|
|
assert collection.is_moving() == False
|
|
|
|
|
|
# Integration
|
|
|
|
def test_nested_conversions():
|
|
base = DummyAdjustable(name="Base", ID="base_id")
|
|
|
|
converted1 = Converted("conv1_id", base,
|
|
conv_get=lambda x: x * 10,
|
|
conv_set=lambda x: x / 10,
|
|
name="Conv1")
|
|
|
|
converted2 = Converted("conv2_id", converted1,
|
|
conv_get=lambda x: x * 2,
|
|
conv_set=lambda x: x / 2,
|
|
name="Conv2")
|
|
|
|
base.set_target_value(1).wait()
|
|
assert converted2.get_current_value() == 20
|
|
|
|
converted2.set_target_value(100).wait()
|
|
assert base.get_current_value() == 5
|
|
|
|
|
|
def test_combined_with_converted():
|
|
base1 = DummyAdjustable(name="Base1", ID="base1_id")
|
|
base2 = DummyAdjustable(name="Base2", ID="base2_id")
|
|
|
|
scaled1 = Converted("scaled1_id", base1,
|
|
conv_get=lambda x: x * 10,
|
|
conv_set=lambda x: x / 10,
|
|
name="Scaled1")
|
|
|
|
scaled2 = Converted("scaled2_id", base2,
|
|
conv_get=lambda x: x * 100,
|
|
conv_set=lambda x: x / 100,
|
|
name="Scaled2")
|
|
|
|
combined = Combined("combscaled_id", [scaled1, scaled2], name="CombScaled")
|
|
|
|
combined.set_target_value(50).wait()
|
|
|
|
assert base1.get_current_value() == 5
|
|
assert base2.get_current_value() == 0.5
|
|
assert combined.get_current_value() == 50
|
|
|
|
|
|
def test_collection_with_converted():
|
|
base1 = DummyAdjustable(name="Base1", ID="base1_id")
|
|
base2 = DummyAdjustable(name="Base2", ID="base2_id")
|
|
|
|
scaled1 = Converted("scaled1_id", base1,
|
|
conv_get=lambda x: x * 10,
|
|
conv_set=lambda x: x / 10,
|
|
name="Scaled1")
|
|
|
|
scaled2 = Converted("scaled2_id", base2,
|
|
conv_get=lambda x: x * 100,
|
|
conv_set=lambda x: x / 100,
|
|
name="Scaled2")
|
|
|
|
collection = Collection("collscaled_id", [scaled1, scaled2], name="CollScaled")
|
|
|
|
collection.set_target_value(50, 200).wait()
|
|
|
|
assert base1.get_current_value() == 5
|
|
assert base2.get_current_value() == 2
|
|
assert collection.get_current_value() == (50, 200)
|
|
|
|
|
|
# Adjustable Base Class
|
|
|
|
@pytest.mark.parametrize("initial,delta1,delta2,expected_final", [
|
|
(10, 5, -3, 12),
|
|
(0, 100, -50, 50),
|
|
(42, -10, 8, 40),
|
|
(-5, 15, -20, -10),
|
|
(100, 0, 0, 100),
|
|
])
|
|
def test_adjustable_tweak(initial, delta1, delta2, expected_final):
|
|
adj = DummyAdjustable(name="Test", ID="test_id", initial_value=initial)
|
|
|
|
adj.tweak(delta1).wait()
|
|
assert adj.get_current_value() == initial + delta1
|
|
|
|
adj.tweak(delta2).wait()
|
|
assert adj.get_current_value() == expected_final
|
|
|
|
|
|
def test_adjustable_call_syntax():
|
|
adj = DummyAdjustable(name="Test", ID="test_id", initial_value=42)
|
|
|
|
assert adj() == 42
|
|
|
|
adj(100).wait()
|
|
assert adj() == 100
|
|
|
|
|
|
def test_adjustable_set_get_aliases():
|
|
adj = DummyAdjustable(name="Test", ID="test_id", initial_value=5)
|
|
|
|
assert adj.get() == 5
|
|
|
|
adj.set(20).wait()
|
|
assert adj.get() == 20
|
|
|
|
|
|
def test_adjustable_moving_property():
|
|
adj = DummyAdjustable(name="Test", ID="test_id")
|
|
|
|
assert isinstance(adj.moving, bool)
|
|
assert adj.moving == False
|
|
|
|
|
|
def test_adjustable_repr_with_units():
|
|
adj = DummyAdjustable(name="Position", ID="pos_id", initial_value=42, units="mm")
|
|
|
|
repr_str = repr(adj)
|
|
assert "Position" in repr_str
|
|
assert "42" in repr_str
|
|
assert "mm" in repr_str
|
|
|
|
|
|
def test_adjustable_repr_with_degrees():
|
|
adj = DummyAdjustable(name="Angle", ID="angle_id", initial_value=90, units="deg")
|
|
|
|
repr_str = repr(adj)
|
|
assert "90°" in repr_str or "90 deg" in repr_str
|
|
|
|
|
|
def test_adjustable_str():
|
|
adj = DummyAdjustable(name="Test", ID="test_id", initial_value=3.14, units="m")
|
|
|
|
str_val = str(adj)
|
|
assert "3.14" in str_val
|
|
assert "m" in str_val
|
|
|
|
|
|
# NumericConvenience
|
|
|
|
@pytest.mark.parametrize("value,expected_int,expected_float", [
|
|
(42.7, 42, 42.7),
|
|
(3.14159, 3, 3.14159),
|
|
(99.99, 99, 99.99),
|
|
(-5.8, -5, -5.8),
|
|
(0.1, 0, 0.1),
|
|
])
|
|
def test_numeric_convenience_int_float(value, expected_int, expected_float):
|
|
adj = DummyAdjustable(name="Test", ID="test_id", initial_value=value)
|
|
|
|
assert int(adj) == expected_int
|
|
assert isinstance(int(adj), int)
|
|
assert float(adj) == expected_float
|
|
assert isinstance(float(adj), float)
|
|
|
|
|
|
@pytest.mark.parametrize("value,round0,round1,round2", [
|
|
(3.14159, 3, 3.1, 3.14),
|
|
(2.71828, 3, 2.7, 2.72),
|
|
(9.8765, 10, 9.9, 9.88),
|
|
(-4.567, -5, -4.6, -4.57),
|
|
(100.123, 100, 100.1, 100.12),
|
|
])
|
|
def test_numeric_convenience_round(value, round0, round1, round2):
|
|
adj = DummyAdjustable(name="Test", ID="test_id", initial_value=value)
|
|
|
|
assert round(adj) == round0
|
|
assert round(adj, 1) == round1
|
|
assert round(adj, 2) == round2
|
|
|
|
|
|
@pytest.mark.parametrize("value,expected_trunc,expected_floor,expected_ceil", [
|
|
(3.9, 3, 3, 4),
|
|
(3.1, 3, 3, 4),
|
|
(-2.8, -2, -3, -2),
|
|
(-2.1, -2, -3, -2),
|
|
(5.5, 5, 5, 6),
|
|
])
|
|
def test_numeric_convenience_math_funcs(value, expected_trunc, expected_floor, expected_ceil):
|
|
import math
|
|
adj = DummyAdjustable(name="Test", ID="test_id", initial_value=value)
|
|
|
|
assert math.trunc(adj) == expected_trunc
|
|
assert math.floor(adj) == expected_floor
|
|
assert math.ceil(adj) == expected_ceil
|
|
|
|
|
|
# SpecConvenience
|
|
|
|
def test_spec_convenience_wm():
|
|
adj = DummyAdjustable(name="Test", ID="test_id", initial_value=42)
|
|
|
|
assert adj.wm() == 42
|
|
|
|
|
|
def test_spec_convenience_mv():
|
|
adj = DummyAdjustable(name="Test", ID="test_id", initial_value=10)
|
|
|
|
adj.mv(50).wait()
|
|
assert adj.get_current_value() == 50
|
|
|
|
|
|
@pytest.mark.parametrize("initial,move1,move2,expected_final", [
|
|
(10, 5, -3, 12),
|
|
(0, 50, 30, 80),
|
|
(100, -20, -10, 70),
|
|
(-5, 15, -8, 2),
|
|
(42, 0, 8, 50),
|
|
])
|
|
def test_spec_convenience_mvr(initial, move1, move2, expected_final):
|
|
adj = DummyAdjustable(name="Test", ID="test_id", initial_value=initial)
|
|
|
|
adj.mvr(move1).wait()
|
|
assert adj.get_current_value() == initial + move1
|
|
|
|
adj.mvr(move2).wait()
|
|
assert adj.get_current_value() == expected_final
|
|
|
|
|
|
# Limited
|
|
|
|
def test_limited_set_limits():
|
|
adj = DummyAdjustable(name="Test", ID="test_id")
|
|
|
|
adj.set_limits(low=0, high=100)
|
|
assert adj.limit_low == 0
|
|
assert adj.limit_high == 100
|
|
|
|
|
|
@pytest.mark.parametrize("low,high,valid_values,invalid_low,invalid_high", [
|
|
(0, 100, [0, 50, 100], -10, 150),
|
|
(-50, 50, [-50, 0, 50], -100, 100),
|
|
(10, 20, [10, 15, 20], 5, 25),
|
|
(-100, -10, [-100, -50, -10], -150, 0),
|
|
(0, 1000, [0, 500, 1000], -1, 1001),
|
|
])
|
|
def test_limited_with_various_ranges(low, high, valid_values, invalid_low, invalid_high):
|
|
from slic.core.adjustable.limited import OutsideLimits
|
|
|
|
adj = DummyAdjustable(name="Test", ID="test_id")
|
|
adj.set_limits(low=low, high=high)
|
|
|
|
for val in valid_values:
|
|
adj.set_target_value(val).wait()
|
|
assert adj.get_current_value() == val
|
|
|
|
with pytest.raises(OutsideLimits):
|
|
adj.set_target_value(invalid_low)
|
|
|
|
with pytest.raises(OutsideLimits):
|
|
adj.set_target_value(invalid_high)
|
|
|
|
|
|
def test_limited_no_limits():
|
|
adj = DummyAdjustable(name="Test", ID="test_id")
|
|
|
|
adj.set_target_value(-999).wait()
|
|
assert adj.get_current_value() == -999
|
|
|
|
adj.set_target_value(999).wait()
|
|
assert adj.get_current_value() == 999
|
|
|
|
|
|
def test_limited_only_low_limit():
|
|
adj = DummyAdjustable(name="Test", ID="test_id")
|
|
adj.set_limits(low=0)
|
|
|
|
adj.set_target_value(1000).wait()
|
|
assert adj.get_current_value() == 1000
|
|
|
|
from slic.core.adjustable.limited import OutsideLimits
|
|
with pytest.raises(OutsideLimits):
|
|
adj.set_target_value(-1)
|
|
|
|
|
|
def test_limited_only_high_limit():
|
|
adj = DummyAdjustable(name="Test", ID="test_id")
|
|
adj.set_limits(high=100)
|
|
|
|
adj.set_target_value(-1000).wait()
|
|
assert adj.get_current_value() == -1000
|
|
|
|
from slic.core.adjustable.limited import OutsideLimits
|
|
with pytest.raises(OutsideLimits):
|
|
adj.set_target_value(150)
|
|
|
|
|
|
def test_limited_reversed_limits():
|
|
adj = DummyAdjustable(name="Test", ID="test_id")
|
|
adj.set_limits(low=100, high=10)
|
|
|
|
adj.set_target_value(50).wait()
|
|
assert adj.get_current_value() == 50
|