Henrik Lemke
5b6f2d2838
- shared eco.widgets.live_values: writes on a background thread with a status dot (and working Stop during a move), reads off the GUI/kernel thread, monitored items subscribed (trusted only once an event arrives, demoted if a poll sees an unannounced change), tweak step from the IOC's TWV for motor/SmarAct records. - assembly_browser Set no longer falls back to mvr (a relative move with an absolute value); runs off the kernel thread. - fix process aborts: IOC finder worker destroyed while running, simple gadget polling after close; tweak panel message timer guarded.
290 lines
8.7 KiB
Python
290 lines
8.7 KiB
Python
"""Tests for eco.widgets.live_values: the toolkit-independent reading/writing
|
|
behind the Qt, ipywidgets and Tk property grids."""
|
|
|
|
import threading
|
|
import time
|
|
|
|
from eco.widgets import live_values
|
|
|
|
|
|
def wait_for(condition, seconds=3.0):
|
|
deadline = time.monotonic() + seconds
|
|
while time.monotonic() < deadline:
|
|
if condition():
|
|
return True
|
|
time.sleep(0.01)
|
|
return condition()
|
|
|
|
|
|
class Item:
|
|
def __init__(self, name, value=1.0, delay=0.0):
|
|
self.name = name
|
|
self.value = value
|
|
self.delay = delay
|
|
self.reads = 0
|
|
|
|
def get_current_value(self):
|
|
self.reads += 1
|
|
time.sleep(self.delay)
|
|
return self.value
|
|
|
|
|
|
class Monitored(Item):
|
|
def __init__(self, name):
|
|
super().__init__(name)
|
|
self.callbacks = {}
|
|
|
|
def add_value_callback(self, callback, index=None):
|
|
index = len(self.callbacks) + 1
|
|
self.callbacks[index] = callback
|
|
return index
|
|
|
|
def clear_value_callback(self, index=None):
|
|
self.callbacks.pop(index, None)
|
|
|
|
def push(self, value):
|
|
self.value = value
|
|
for callback in list(self.callbacks.values()):
|
|
callback(pvname=self.name, value=value)
|
|
|
|
|
|
class CallbackEpicsLike:
|
|
"""What set_current_value_callback returns for PV-backed items."""
|
|
|
|
def __init__(self, func):
|
|
self.func = func
|
|
self.started = False
|
|
|
|
def start(self, add_current_value=True, with_ctrlvars=True, auto_monitor=True):
|
|
self.started = (add_current_value, with_ctrlvars)
|
|
|
|
def stop(self):
|
|
self.started = None
|
|
|
|
|
|
class PvMonitored(Item):
|
|
def set_current_value_callback(self, func="accumulate", **kwargs):
|
|
self.monitor = CallbackEpicsLike(func)
|
|
return self.monitor
|
|
|
|
|
|
def collect():
|
|
results = []
|
|
return results, lambda index, ok, payload: results.append((index, ok, payload))
|
|
|
|
|
|
def test_monitored_items_update_at_once_and_unsubscribe_on_stop():
|
|
results, on_result = collect()
|
|
item = Monitored("m")
|
|
reader = live_values.LiveReader(on_result, poll_interval=30.0)
|
|
reader.add(item)
|
|
reader.start()
|
|
try:
|
|
assert wait_for(lambda: item.callbacks)
|
|
assert wait_for(lambda: results) # the first read
|
|
assert not reader.monitored(0) # subscribed, but no event yet: not trusted
|
|
item.push(5.0)
|
|
assert wait_for(lambda: results[-1] == (0, True, 5.0), 1.0) # not after 30 s
|
|
assert reader.monitored(0)
|
|
finally:
|
|
reader.stop()
|
|
assert wait_for(lambda: not item.callbacks)
|
|
|
|
|
|
def test_a_monitor_without_parents_is_not_trusted():
|
|
"""e.g. DetectorVirtual([], fn) -- vacuum valve readbacks, the prepump state."""
|
|
|
|
class Composed:
|
|
children = []
|
|
|
|
def start(self, **kwargs):
|
|
pass
|
|
|
|
def stop(self):
|
|
pass
|
|
|
|
item = Item("readback")
|
|
item.set_current_value_callback = lambda func=None, **kw: Composed()
|
|
assert live_values.subscribe(item, lambda: None) is None
|
|
|
|
|
|
def test_a_monitor_that_misses_changes_is_demoted(monkeypatch):
|
|
"""e.g. PshellMotor: calls back only when a move ends."""
|
|
monkeypatch.setattr(live_values, "MONITORED_POLL_INTERVAL", 0.3)
|
|
results, on_result = collect()
|
|
item = Monitored("pshell")
|
|
reader = live_values.LiveReader(on_result, poll_interval=0.2)
|
|
reader.add(item)
|
|
reader.start()
|
|
try:
|
|
assert wait_for(lambda: item.callbacks)
|
|
item.push(1.0) # a move ended: an event
|
|
assert wait_for(lambda: reader.monitored(0))
|
|
time.sleep(0.5) # the read that event asked for is done
|
|
item.value = 2.0 # the next move: no event
|
|
assert wait_for(lambda: not reader.monitored(0), 2.0)
|
|
assert wait_for(lambda: results[-1] == (0, True, 2.0))
|
|
finally:
|
|
reader.stop()
|
|
|
|
|
|
def test_pv_style_monitor_is_started_without_blocking_reads():
|
|
reader = live_values.LiveReader(lambda *a: None, poll_interval=30.0)
|
|
item = PvMonitored("pv")
|
|
reader.add(item)
|
|
reader.start()
|
|
try:
|
|
assert wait_for(lambda: getattr(item, "monitor", None) is not None and item.monitor.started)
|
|
assert item.monitor.started == (False, False) # no seeding get, no ctrlvars get
|
|
finally:
|
|
reader.stop()
|
|
assert wait_for(lambda: item.monitor.started is None)
|
|
|
|
|
|
def test_a_slow_item_does_not_hold_up_the_others():
|
|
results, on_result = collect()
|
|
slow, fast = Item("slow", delay=2.0), Item("fast", value=3.0)
|
|
reader = live_values.LiveReader(on_result, poll_interval=0.1)
|
|
reader.add(slow)
|
|
reader.add(fast)
|
|
reader.start()
|
|
try:
|
|
assert wait_for(lambda: sum(1 for r in results if r[0] == 1) >= 3, 1.5)
|
|
finally:
|
|
reader.stop()
|
|
|
|
|
|
def test_no_value_and_errors_are_reported():
|
|
results, on_result = collect()
|
|
|
|
class Broken:
|
|
def get_current_value(self):
|
|
raise RuntimeError("IOC down")
|
|
|
|
dead = Item("dead", value=None)
|
|
reader = live_values.LiveReader(on_result, poll_interval=30.0)
|
|
reader.add(dead)
|
|
reader.add(Broken())
|
|
reader.start()
|
|
try:
|
|
assert wait_for(lambda: {r[0] for r in results} == {0, 1})
|
|
finally:
|
|
reader.stop()
|
|
by_index = {r[0]: r for r in results}
|
|
assert by_index[0] == (0, False, "no value (not connected?)")
|
|
assert by_index[1] == (1, False, "RuntimeError: IOC down")
|
|
|
|
|
|
def test_inactive_items_are_not_read_until_activated_and_restart_works():
|
|
results, on_result = collect()
|
|
item = Item("hidden")
|
|
reader = live_values.LiveReader(on_result, poll_interval=30.0)
|
|
reader.add(item, active=False)
|
|
reader.start()
|
|
time.sleep(0.3)
|
|
assert item.reads == 0
|
|
reader.set_active(0, True)
|
|
assert wait_for(lambda: item.reads == 1)
|
|
reader.stop()
|
|
reader.start() # e.g. make_assembly_widget's start() after stop()
|
|
try:
|
|
assert wait_for(lambda: item.reads == 2)
|
|
finally:
|
|
reader.stop()
|
|
|
|
|
|
class Changer:
|
|
def __init__(self, seconds):
|
|
self.done = threading.Event()
|
|
self.seconds = seconds
|
|
self.stopped = False
|
|
|
|
def wait(self):
|
|
self.done.wait(self.seconds)
|
|
|
|
def stop(self):
|
|
self.stopped = True
|
|
self.done.set()
|
|
|
|
|
|
def test_start_write_reports_the_move_and_errors():
|
|
states = []
|
|
finished = threading.Event()
|
|
item = Item("mot")
|
|
holder = {}
|
|
item.set_target_value = lambda value: Changer(0.3)
|
|
live_values.start_write(item, lambda: 2.0, lambda *s: states.append(s), finished.set, holder)
|
|
assert wait_for(finished.is_set)
|
|
assert states == [("moving", "to 2.0"), ("idle", "")]
|
|
assert isinstance(holder["changer"], Changer)
|
|
|
|
states.clear()
|
|
finished.clear()
|
|
|
|
def blocked(value):
|
|
raise PermissionError("mot is in use by queue 'default', task #1 -- not moved.")
|
|
|
|
item.set_target_value = blocked
|
|
live_values.start_write(item, lambda: 2.0, lambda *s: states.append(s), finished.set)
|
|
assert wait_for(finished.is_set)
|
|
assert states == [("error", "mot is in use by queue 'default', task #1 -- not moved.")]
|
|
|
|
|
|
def test_stop_move_stops_the_changer_or_the_item():
|
|
changer = Changer(5)
|
|
live_values.stop_move(None, changer, lambda *s: None)
|
|
assert wait_for(lambda: changer.stopped)
|
|
|
|
item = Item("mot")
|
|
item.stopped = False
|
|
item.stop = lambda: setattr(item, "stopped", True)
|
|
live_values.stop_move(item, None, lambda *s: None)
|
|
assert wait_for(lambda: item.stopped)
|
|
|
|
|
|
def test_prefetch_has_a_deadline():
|
|
slow, fast = Item("slow", delay=1.5), Item("fast", value=4.0)
|
|
t0 = time.monotonic()
|
|
values = live_values.prefetch_values([slow, fast], deadline=0.3)
|
|
assert time.monotonic() - t0 < 0.8
|
|
assert values[id(slow)] is live_values.PENDING
|
|
assert values[id(fast)] == 4.0
|
|
|
|
|
|
def test_step_sources():
|
|
item = Item("mot")
|
|
assert live_values.initial_step(item, 1) == 1
|
|
assert live_values.initial_step(item, 1.0) == 0.1
|
|
item._tweak_instance = type("T", (), {"step_sizes": [0.02]})()
|
|
assert live_values.initial_step(item, 1.0) == 0.02
|
|
|
|
class Pv:
|
|
def __init__(self):
|
|
self.puts = []
|
|
|
|
def get(self, timeout=None):
|
|
return 0.5
|
|
|
|
def put(self, value):
|
|
self.puts.append(value)
|
|
|
|
class Motor(Item):
|
|
def _tweak_ioc(self):
|
|
pass
|
|
|
|
motor = Motor("m")
|
|
pv = Pv()
|
|
motor._motor = type("M", (), {"get_pv": lambda _, field: pv})()
|
|
assert live_values.has_ioc_step(motor) and not live_values.has_ioc_step(item)
|
|
got = []
|
|
holder = live_values.load_ioc_step(motor, got.append)
|
|
assert wait_for(lambda: got == [0.5])
|
|
live_values.store_ioc_step(holder, 0.25)
|
|
assert wait_for(lambda: pv.puts == [0.25])
|
|
|
|
|
|
def test_dot_state():
|
|
assert live_values.dot_state("moving", "to 2", ("polled", ""))[1] == "moving: to 2"
|
|
assert live_values.dot_state("idle", "", ("live", ""))[1] == "connected, updated by monitor"
|
|
assert live_values.dot_state("error", "x", ("live", ""))[0] == "#d93025"
|