diff --git a/pyproject.toml b/pyproject.toml index 58852c0..42de3e8 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -16,6 +16,7 @@ classifiers = [ "Topic :: Utilities", ] dependencies = [ + 'caproto', 'lcls-tools @ git+https://github.com/slaclab/lcls-tools', 'pyepics', 'pykern', diff --git a/slicops/config.py b/slicops/config.py index ce5a13d..27abe8b 100644 --- a/slicops/config.py +++ b/slicops/config.py @@ -39,5 +39,10 @@ def cfg(): tcp_port=(8000, pykern.pkasyncio.cfg_port, "port of server"), vue_port=(8008, pykern.pkasyncio.cfg_port, "port of Vue dev server"), ), + package_path=( + tuple(["slicops"]), + tuple, + "Names of root packages that should be checked for codes and resources. Order is important, the first package with a matching code/resource will be used.", + ), ) return _cfg diff --git a/slicops/device.py b/slicops/device/__init__.py similarity index 67% rename from slicops/device.py rename to slicops/device/__init__.py index 6150bc9..dfa59b4 100644 --- a/slicops/device.py +++ b/slicops/device/__init__.py @@ -10,6 +10,9 @@ import slicops.device_db import threading +# TODO(robnagler) configure via device_db +_TIMEOUT = 5 + class AccessorPutError(RuntimeError): """The PV for this accessor is not writable""" @@ -34,6 +37,7 @@ class Device: def __init__(self, device_name): self.device_name = device_name self.meta = slicops.device_db.meta_for_device(device_name) + self._destroyed = False self._accessor = PKDict() self.connected = False @@ -45,16 +49,21 @@ def accessor(self, accessor_name): Returns: _Accessor: object holding PV state """ + if self._destroyed: + raise AssertionError(f"destroyed {self}") return self._accessor.pksetdefault( accessor_name, lambda: _Accessor(self, accessor_name) )[accessor_name] def destroy(self): """Disconnect from PV's and remove state about device""" + if self._destroyed: + return + self._destroyed = True x = list(self._accessor.values()) self._accessor = PKDict() for a in x: - a.disconnect() + a.destroy() def get(self, accessor_name): """Read from PV @@ -85,6 +94,9 @@ def put(self, accessor_name, value): """ return self.accessor(accessor_name).put(value) + def __repr__(self): + return f"" + class _Accessor: """Container for a PV, metadata, and dynamic state @@ -96,26 +108,33 @@ class _Accessor: def __init__(self, device, accessor_name): self.device = device + self.accessor_name = accessor_name self.meta = device.meta.accessor[accessor_name] self._callback = None - self._mutex = threading.Lock() - # TODO(pjm): connection and PV timeouts need to be configurable? - self._pv = epics.PV( - self.meta.pv_name, - connection_callback=self._on_connection, - connection_timeout=4.0, - ) - if accessor_name == "image": - # TODO(robnagler) this has to be done here, because you can't get pvs - # from within a monitor callback - self._image_shape = (self.device.get("n_row"), self.device.get("n_col")) - - def disconnect(self): + self._destroyed = False + self._lock = threading.Lock() + self._initialized = threading.Event() + self._initializing = False + # Defer initialization + self._pv = None + + def destroy(self): """Stop all monitoring and disconnect from PV""" - self._callback = None + if self._destroyed: + return + with self._lock: + if self._destroyed: + return + self._destroyed = True + self._initializing = False + self._callback = None + if (p := self._pv) is None: + return + self._pv = None + self._initialized.set() try: # Clears all callbacks - self._pv.disconnect() + p.disconnect() except Exception as e: pkdlog("error={} {} stack={}", e, self, pkdexc()) @@ -125,11 +144,10 @@ def get(self): Returns: object: the value from the PV converted to a Python type """ - - # TODO(pjm): connection and PV timeouts need to be configurable? - if (rv := self._pv.get(timeout=5.0)) is None: + p = self.__pv() + if (rv := p.get(timeout=_TIMEOUT)) is None: raise DeviceError(f"unable to get {self}") - if not self._pv.connected: + if not p.connected: raise DeviceError(f"disconnected {self}") return self._fixup_value(rv) @@ -147,23 +165,21 @@ def monitor(self, callback): Args: callback (callable): accepts a single `PKDict` as ag """ - with self._mutex: + with self._lock: + self._assert_not_destroyed() if self._callback: - raise ValueError(f"already monitoring {self}") - # should lock - self._callback_index = self._pv.add_callback(self._on_value) - self._pv.auto_monitor = True + raise AssertionError("may only call monitor once") + if self._pv or self._initializing: + raise AssertionError("monitor must be called before get/put") self._callback = callback + self.__pv() def monitor_stop(self): """Stops monitoring PV""" - with self._mutex: - if not self._callback: + with self._lock: + if self._destroyed or not self._callback: return self._callback = None - self._pv.auto_monitor = False - self._pv.remove_callback(self._callback_index) - self._callback_index = None def put(self, value): """Set PV to value @@ -180,18 +196,23 @@ def put(self, value): else: raise AccessorPutError(f"unhandled py_type={self.meta.py_type} {self}") # ECA_NORMAL == 0 and None is normal, too, apparently - if (e := self._pv.put(v)) != 1: + p = self.__pv() + if (e := p.put(v)) != 1: raise DeviceError(f"put error={e} value={v} {self}") - if not self._pv.connected: + if not p.connected: raise DeviceError(f"disconnected {self}") + def _assert_not_destroyed(self): + if self._destroyed: + raise AssertionError(f"destroyed {self}") + def _fixup_value(self, raw): def _reshape(image): return image.reshape(self._image_shape) if self.meta.py_type == bool: return bool(raw) - if self.meta.accessor_name == "image": + if self.accessor_name == "image": return _reshape(raw) return raw @@ -221,12 +242,41 @@ def _on_value(self, **kwargs): pkdlog("error={} {} stack={}", e, self, pkdexc()) raise + def __pv(self): + with self._lock: + self._assert_not_destroyed() + if self._pv: + return self._pv + if not (i := self._initializing): + self._initializing = True + if i: + self._initialized.wait(timeout=_TIMEOUT) + else: + k = ( + PKDict(callback=self._on_value, auto_monitor=True) + if self._callback + else PKDict() + ) + if self.accessor_name == "image": + # TODO(robnagler) this has to be done here, because you can't get pvs + # from within a monitor callback. + # TODO(robnagler) need a better way of dealing with this + self._image_shape = (self.device.get("n_row"), self.device.get("n_col")) + self._pv = epics.PV( + self.meta.pv_name, + connection_callback=self._on_connection, + connection_timeout=_TIMEOUT, + **k, + ) + self._initialized.set() + return self._pv + def __repr__(self): - return f"<_Accessor {self.device.device_name}.{self.meta.accessor_name} {self.meta.pv_name}>" + return f"<_Accessor {self.device.device_name}.{self.accessor_name} {self.meta.pv_name}>" def _run_callback(self, **kwargs): k = PKDict(accessor=self, **kwargs) - with self._mutex: + with self._lock: c = self._callback if c: c(k) diff --git a/slicops/device/screen.py b/slicops/device/screen.py new file mode 100644 index 0000000..afe79d4 --- /dev/null +++ b/slicops/device/screen.py @@ -0,0 +1,345 @@ +"""Control a Screen + +:copyright: Copyright (c) 2024 The Board of Trustees of the Leland Stanford Junior University, through SLAC National Accelerator Laboratory (subject to receipt of any required approvals from the U.S. Dept. of Energy). All Rights Reserved. +:license: http://github.com/slaclab/slicops/LICENSE +""" + +from pykern.pkcollections import PKDict +from pykern.pkdebug import pkdc, pkdexc, pkdlog, pkdp +from slicops.device import DeviceError +import abc +import enum +import logging +import pykern.pkconfig +import queue +import slicops.device +import slicops.device_db +import threading + +# TODO(robnagler) these should be reused for both cases +_MOVE_TARGET_IN = PKDict({False: 0, True: 1}) +_STATUS_IN = 2 +_STATUS_OUT = 1 + +_BLOCKING_MSG = "upstream target is in" +_TIMEOUT_MSG = "upstream target status accessor timed out" +_ERROR_PREFIX_MSG = "upstream target error: " + + +class Screen(slicops.device.Device): + """Augment `Device` with screen specific operations""" + + def __init__(self, beam_path, device_name, handler, *args, **kwargs): + super().__init__(device_name, *args, **kwargs) + if not isinstance(handler, EventHandler): + raise AssertionError( + f"handler is not subclass EventHandler type={type(handler)}" + ) + self.__worker = _Worker(beam_path, handler, self) + + def destroy(self): + self.__worker.destroy() + super().destroy() + + def move_target(self, want_in): + """Insert or remove the target + + Args: + want_in (bool): True to insert, and False to remove + """ + self.__worker.req_action( + self.__worker.action_req_move_target, PKDict(want_in=want_in) + ) + + +class ErrorKind(enum.Enum): + """Errors passed to on_screen_device_error""" + + fsm = enum.auto() + monitor = enum.auto() + upstream = enum.auto() + + +class EventHandler: + """Clients of DeviceScreen must implement this""" + + @abc.abstractmethod + def on_screen_device_error(self, accessor_name, error_kind, error_msg): + pass + + @abc.abstractmethod + def on_screen_device_update(self, accessor_name, value): + pass + + +class _ActionLoop: + """Generic thread that processes actions in a loop on request""" + + _LOOP_END = object() + + def __init__(self): + self.destroyed = False + self.__lock = threading.Lock() + self.__actions = queue.Queue() + self.__thread = threading.Thread(target=self._start, daemon=True) + if self._loop_timeout_secs > 0 and not hasattr(self, "action_loop_timeout"): + raise AssertionError( + f"_loop_timeout_secs={self._loop_timeout_secs} and not action_loop_timeout" + ) + self.__thread.start() + + def action(self, method, arg): + self.__actions.put_nowait((method, arg)) + + def destroy(self): + try: + with self.__lock: + if self.destroyed: + return + self.destroyed = True + self.__actions.put_nowait((None, None)) + self._destroy() + except Exception as e: + pkdlog("error={} {} stack={}", e, self, pkdexc(simplify=True)) + + def __repr__(self): + def _destroyed(): + return " DESTROYED" if self.destroyed else "" + + return f"<{self.__class__.__name__}{_destroyed()} self._repr()>" + + def _start(self): + timeout_kwarg = PKDict() + if self._loop_timeout_secs: + timeout_kwarg.timeout = self._loop_timeout_secs + try: + while True: + try: + m, a = self.__actions.get(**timeout_kwarg) + except queue.Empty: + m, a = self.action_loop_timeout(), None + with self.__lock: + if self.destroyed: + return + if m(a) is self._LOOP_END: + return + except Exception as e: + pkdlog("error={} {} stack={}", e, self, pkdexc(simplify=True)) + finally: + self.destroy() + + +class _FSM: + """Finite State Machine for DeviceScreen""" + + def __init__(self, worker, handler): + self.worker = worker + self.handler = handler + self.curr = PKDict( + acquire=False, + check_upstream=False, + move_target_arg=None, + target_status=None, + upstream_problems=None, + ) + self.prev = self.curr.copy() + + def event(self, name, arg): + self.prev = self.curr.copy() + if u := getattr(self, f"_event_{name}")(arg, **self.curr): + self.curr.update(u) + + def _event_handle_monitor(self, arg, **kwargs): + n = arg.accessor.accessor_name + if "error" in arg: + self.handler.on_screen_device_error( + error_kind=ErrorKind.monitor, accessor_name=n, error_msg=arg.error + ) + if n == "target_status": + # TODO(robnagler) is resetting move_target_arg right? + return PKDict(target_status=None, move_target_arg=None) + return + if "connected" in arg: + return + if n == "image": + v = arg.value + rv = None + elif n == "acquire": + v = arg.value + rv = PKDict(acquire=arg.value) + elif n == "target_status": + v = _STATUS_IN == arg.value + rv = PKDict(move_target_arg=None, target_status=v) + else: + raise AssertionError(f"unsupported accessor={n} {self}") + self.handler.on_screen_device_update(accessor_name=n, value=v) + return rv + + def _event_move_target( + self, + arg, + check_upstream, + move_target_arg, + target_status, + upstream_problems, + **kwargs, + ): + if move_target_arg: + self.handler.on_screen_device_error( + error_kind="fsm", error_msg="target already moving" + ) + return + if target_status is not None and arg.want_in == target_status: + # TODO(robnagler) could be a race condition so probably fine to do nothing + pkdlog("same target_status={} self.want_in={}", target_status, arg.want_in) + return + # TODO(robnagler) allow moving without checking upstream + rv = PKDict(move_target_arg=arg) + if arg.want_in and upstream_problems is None or upstream_problems: + # Recheck the upstream + self.worker.action(self.worker.action_check_upstream, None) + rv.check_upstream = True + else: + self.worker.action(self.worker.action_move_target, arg) + return rv + + def _event_upstream_status(self, arg, move_target_arg, **kwargs): + rv = PKDict(check_upstream=False, upstream_problems=arg.problems) + if arg.problems: + self.handler.on_screen_device_error( + error_kind="upstream", error_msg=arg.problems + ) + return rv.pkupdate(move_target_arg=None) + self.worker.action(self.worker.action_move_target, move_target_arg) + return rv + + def __repr__(self): + def _states(curr): + return " ".join(f"{k}={curr[k]}" for k in sorted(curr.keys())) + + return f"<_FSM {self.worker.device.device_name} {_states(self.curr)}>" + + +class _Upstream(_ActionLoop): + """Action loop to check targets of upstream screens""" + + def __init__(self, worker): + def _names(): + return slicops.device_db.upstream_devices( + "PROF", "target_control", worker.beam_path, worker.device.device_name + ) + + self.__worker = worker + self.__problems = PKDict() + self.__devices = PKDict({u: slicops.device.Device(u) for u in _names()}) + self._loop_timeout_secs = _cfg.upstream_timeout_secs + super().__init__() + + def action_handle_status(self, arg): + n = arg.accessor.device.device_name + self.__devices.pkdel(n).destroy() + if e := arg.get("error"): + pkdlog("device={} error={}", n, e) + self.__problems[n] = f"{_ERROR_PREFIX_MSG}{e}" + elif arg.value == _STATUS_IN: + self.__problems[n] = _BLOCKING_MSG + if not self.__devices: + return self.__done() + return None + + def action_loop_timeout(self): + for n in self.__devices: + self.__problems[n] = _TIMEOUT_MSG + return self.__done() + + def _destroy(self): + (d, self.__devices) = (self.__devices, PKDict()) + for x in d.values(): + x.destroy() + + def __done(self): + self.__worker.action( + self.__worker.action_upstream_status, PKDict(problems=self.__problems) + ) + return self._LOOP_END + + def __handle_status(self, kwargs): + if "connected" in kwargs: + return + self.action(self.action_handle_status, kwargs) + + def _start(self, *args, **kwargs): + for d in self.__devices.values(): + d.accessor("target_status").monitor(self.__handle_status) + super()._start(*args, **kwargs) + + def _repr(self): + return f"pending={sorted(self.__devices)} problems={sorted(self.__problems)}" + + +class _Worker(_ActionLoop): + """Action loop for DeviceScreen""" + + def __init__(self, beam_path, handler, device): + self.beam_path = beam_path + self.device = device + self.__upstream = None + self.__status = None + self.__fsm = _FSM(self, handler) + self.__target_control = None + self._loop_timeout_secs = 0 + super().__init__() + + def action_check_upstream(self, arg): + self.__upstream = _Upstream(self) + return None + + def action_handle_monitor(self, arg): + self.__fsm.event("handle_monitor", arg) + return None + + def action_move_target(self, arg): + if not self.__target_control: + self.__target_control = self.device.accessor("target_control") + self.__target_control.put(_MOVE_TARGET_IN[arg.want_in]) + return None + + def action_req_move_target(self, arg): + self.__fsm.event("move_target", arg) + return None + + def action_upstream_status(self, arg): + self.__fsm.event("upstream_status", arg) + self.__upstream = None + return None + + def req_action(self, method, arg): + """Called by DeviceScreen which has separate life cycle""" + if self.destroyed: + raise AssertionError("object is destroyed") + self.action(method, arg) + + def _destroy(self): + if self.__upstream: + (u, self.__upstream) = (self.__upstream, None) + u.destroy() + + def __handle_monitor(self, change): + self.action(self.action_handle_monitor, change) + + def _start(self, *args, **kwargs): + for a in "acquire", "image", "target_status": + self.device.accessor(a).monitor(self.__handle_monitor) + super()._start(*args, **kwargs) + + def _repr(self): + return f"device={self.device.device_name}" + + +_cfg = pykern.pkconfig.init( + upstream_timeout_secs=( + 15, + pykern.pkconfig.parse_seconds, + "how long to wait for updates from devices", + ), +) diff --git a/slicops/device_db.py b/slicops/device_db.py index 1cc8208..30f2e7a 100644 --- a/slicops/device_db.py +++ b/slicops/device_db.py @@ -22,10 +22,14 @@ class DeviceDbError(Exception): _ACCESSOR_META = PKDict( acquire=PKDict(py_type=bool, pv_writable=True), + enabled=PKDict(py_type=int, pv_writable=False), image=PKDict(py_type=numpy.ndarray, pv_writable=False), n_bits=PKDict(py_type=int, pv_writable=False), n_col=PKDict(py_type=int, pv_writable=False), n_row=PKDict(py_type=int, pv_writable=False), + start_scan=PKDict(py_type=int, pv_writable=True), + target_control=PKDict(py_type=int, pv_writable=True), + target_status=PKDict(py_type=int, pv_writable=False), ) @@ -194,18 +198,18 @@ def beam_paths(): return rv -def device_names(beam_path, device_type): +def device_names(device_type, beam_path): """Query devices for device_type and beam_path Args: - beam_path (str): which beam path device_type (str): type of device to filter by + beam_path (str): which beam path Returns: tuple: sorted device names """ - if device_type not in slicops.const.DEVICE_TYPES: - raise DeviceDbError(f"no such device_type={device_type}") - if rv := slicops.device_sql_db.device_names(beam_path, device_type): + if rv := slicops.device_sql_db.device_names( + _assert_device_type(device_type), beam_path + ): return rv # TODO(robnagler) refine because beam_path could exist, just not for device raise DeviceDbError(f"no devices for beam_path={beam_path}") @@ -236,3 +240,19 @@ def _static(device, accessor): for r in rv.accessor.values(): _static(rv, r) return rv + + +def upstream_devices(device_type, accessor_name, beam_path, device_name): + """returns in z order""" + return slicops.device_sql_db.upstream_devices( + _assert_device_type(device_type), + accessor_name, + beam_path, + device_name, + ) + + +def _assert_device_type(value): + if value not in slicops.const.DEVICE_TYPES: + raise DeviceDbError(f"no such device_type={value}") + return value diff --git a/slicops/device_sql_db.py b/slicops/device_sql_db.py index a10e1af..e2ff4d1 100644 --- a/slicops/device_sql_db.py +++ b/slicops/device_sql_db.py @@ -8,8 +8,9 @@ from pykern.pkcollections import PKDict from pykern.pkdebug import pkdc, pkdlog, pkdp -import pykern.sql_db import pykern.pkresource +import pykern.sql_db +import slicops.config import sqlalchemy _BASE_PATH = "device_db.sqlite3" @@ -25,19 +26,19 @@ def beam_paths(): ) -def device(device_name): +def device(name): with _session() as s: - return PKDict(s.select_one("device", PKDict(device_name=device_name))).pkupdate( + return PKDict(s.select_one("device", PKDict(device_name=name))).pkupdate( accessor=PKDict( { r.accessor_name: PKDict(r) - for r in s.select("device_pv", PKDict(device_name=device_name)) + for r in s.select("device_pv", PKDict(device_name=name)) } ), ) -def device_names(beam_path, device_type): +def device_names(device_type, beam_path): with _session() as s: c = s.t.device.c.device_name return tuple( @@ -57,6 +58,68 @@ def device_names(beam_path, device_type): ) +def upstream_devices(device_type, required_accessor, beam_path, end_device): + with _session() as s: + # select device.device_name from device_meta_float, device where device_meta_name = 'sum_l_meters' and device_meta_value < 33 and device.device_type = 'PROF' and device.device_name = device_meta_float.device_name; + c = s.t.device_meta_float.c.device_name + _assert_on_beampath(end_device, beam_path, s) + return tuple( + r.device_name + for r in s.select( + sqlalchemy.select(c) + .select_from( + s.t.device_meta_float.join( + s.t.device, + s.t.device.c.device_name == c, + ) + .join( + s.t.beam_path, + s.t.beam_path.c.beam_area == s.t.device.c.beam_area, + ) + .join( + s.t.device_pv, + s.t.device_pv.c.device_name == c, + ) + ) + .where( + s.t.beam_path.c.beam_path == beam_path, + s.t.device_meta_float.c.device_meta_name == "sum_l_meters", + s.t.device_meta_float.c.device_meta_value + < _device_meta(end_device, "sum_l_meters", s), + s.t.device.c.device_type == device_type, + s.t.device_pv.c.accessor_name == required_accessor, + ) + .order_by(s.t.device_meta_float.c.device_meta_value) + ) + ) + + +def _assert_on_beampath(device, beam_path, select): + c = select.t.device.c.device_name + v = select.select_one_or_none( + sqlalchemy.select(c) + .select_from( + select.t.device.join( + select.t.beam_path, + select.t.beam_path.c.beam_area == select.t.device.c.beam_area, + ) + ) + .where( + select.t.device.c.device_name == device, + select.t.beam_path.c.beam_path == beam_path, + ), + None, + ) + if v is None: + raise ValueError(f"device={device} is not in beam_path={beam_path}") + + +def _device_meta(device, meta, select): + return select.select_one( + "device_meta_float", PKDict(device_name=device, device_meta_name=meta) + ).device_meta_value + + def recreate(parser): """Recreates db""" assert not _meta @@ -153,7 +216,9 @@ def _init(): def _path(): - return pykern.pkresource.file_path(".").join(_BASE_PATH) + return pykern.pkresource.file_path( + ".", packages=slicops.config.cfg().package_path + ).join(_BASE_PATH) def _session(): diff --git a/slicops/mock_epics.py b/slicops/mock_epics.py index 7a52503..b19a432 100644 --- a/slicops/mock_epics.py +++ b/slicops/mock_epics.py @@ -20,22 +20,39 @@ _X_SIZE = 50 _Y_FACTOR = 1.3 +_PV_VALUE = None + _PV = None class PV: _CB_INDEX = 1 - def __init__(self, name, connection_timeout=0, connection_callback=None): + def __init__( + self, + name, + connection_timeout=0, + callback=None, + auto_monitor=False, + connection_callback=None, + ): + if name in _PV: + raise AssertionError(f"already exists PV={name}") self.pvname = name self.connected = True self.connection_callback = connection_callback self.monitor_callback = None - self._auto_monitor = False self._monitor_queue = None + self._auto_monitor = False + _PV[name] = self + if callback: + self.add_callback(callback) + self.auto_monitor = auto_monitor def disconnect(self): - pass + self._auto_monitor = False + if self._monitor_queue: + self._monitor_queue.put_nowait(None) def add_callback(self, callback): if self.monitor_callback: @@ -49,9 +66,8 @@ def auto_monitor(self): @auto_monitor.setter def auto_monitor(self, value): - def _acquire(): + def _simple(): self.connection_callback(conn=True) - self._monitor_queue = queue.Queue() while True: v = self._monitor_queue.get() if v is None: @@ -62,16 +78,21 @@ def _acquire(): def _image(): self.connection_callback(conn=True) for s in MONITOR_X_SIZE: + if not self._auto_monitor: + break time.sleep(MONITOR_SLEEP) - _PV.pkupdate(_pv_image(s)) - self.monitor_callback(value=_PV[self.pvname]) + _PV_VALUE.pkupdate(_pv_image(s)) + self.monitor_callback(value=_PV_VALUE[self.pvname]) self.connection_callback(conn=False) def _which(): if "ArrayData" in self.pvname: return _image - if "Acquire" in self.pvname: - return _acquire + if "Acquire" in self.pvname or "TGT_STS" in self.pvname: + self._monitor_queue = queue.Queue() + if (v := _PV_VALUE.get(self.pvname)) is not None: + self._monitor_queue.put_nowait(v) + return _simple raise ValueError(f"cannot monitor pv={self.pvname}") if "IMAGE" in self.pvname: @@ -80,14 +101,21 @@ def _which(): return self._auto_monitor = value if value: + # we don't care the thread gets killed since this is a mock for unit threading.Thread(target=_which()).start() def get(self, timeout=0): # TOOD(robnagler) need to be more sophisticated - return _PV.get(self.pvname, None) + return _PV_VALUE.get(self.pvname, None) def put(self, value): - _PV[self.pvname] = value + n = self.pvname + if "PNEUMATIC" in self.pvname: + # TODO(robnagler) set to MOVING (0?) for a few ms + n = self.pvname.replace("PNEUMATIC", "TGT_STS") + value = 1 if value == 0 else 2 + self = _PV[n] + _PV_VALUE[n] = value if self._monitor_queue: self._monitor_queue.put_nowait(value) return 1 @@ -99,16 +127,21 @@ def remove_callback(self, index): def reset_state(): - global _PV + global _PV_VALUE, _PV - _PV = PKDict( + _PV_VALUE = PKDict( { "13SIM1:cam1:Acquire": 0, "13SIM1:cam1:N_OF_BITS": 8, "YAGS:IN20:211:N_OF_COL": 100, "YAGS:IN20:211:N_OF_ROW": 100, + "YAGS:IN20:351:TGT_STS": 1, + "YAGS:IN20:241:TGT_STS": 1, + "YAGS:IN20:211:TGT_STS": 2, + "OTRS:DIAG0:525:Acquire": 0, } ).pkupdate(_pv_image(_X_SIZE)) + _PV = PKDict() def _gaussian(x_size): diff --git a/slicops/package_data/device_db.sqlite3 b/slicops/package_data/device_db.sqlite3 index 2a2db81..21564c9 100644 Binary files a/slicops/package_data/device_db.sqlite3 and b/slicops/package_data/device_db.sqlite3 differ diff --git a/slicops/pkcli/device_db.py b/slicops/pkcli/device_db.py index 375a4d5..e696cfa 100644 --- a/slicops/pkcli/device_db.py +++ b/slicops/pkcli/device_db.py @@ -44,7 +44,7 @@ _KNOWN_KEYS = PKDict( - controls_information=frozenset(("PVs", "control_name")), + controls_information=frozenset(("PVs", "control_name", "pv_cache")), metadata=frozenset( ( "area", @@ -53,6 +53,7 @@ "bpms_before_wire", "l_eff", "lblms", + "hardware", "rf_freq", "sum_l_meters", "type", @@ -72,11 +73,6 @@ ) -def yaml_to_sql(): - """Convert device yaml file to db""" - return slicops.device_sql_db.recreate(_Parser()) - - def parse(): from pykern import pkjson @@ -92,6 +88,25 @@ def parse(): ) +def query(func_name, *args): + """Call func_name in `slicops.device_db` + + Args: + func_name (str): valid function + args (str): passed verbatim to function + Returns: + object: result of function + """ + from slicops import device_db + + return getattr(device_db, func_name)(*args) + + +def yaml_to_sql(): + """Convert device yaml file to db""" + return slicops.device_sql_db.recreate(_Parser()) + + class _Ignore(Exception): pass @@ -161,12 +176,15 @@ def _input_fixups(name, rec): rec.controls_information.PVs.pksetdefault( "acquire", f"{rec.controls_information.control_name}:Acquire" ) + # TODO(robnagler) parse pv_cache return rec def _meta(name, raw): # TODO validation c = raw.controls_information m = raw.metadata + # TODO ignore for now + raw.metadata.pkdel("hardware") self.meta_keys.update(m.keys()) self.ctl_keys.update(c.keys()) rv = PKDict( @@ -194,7 +212,7 @@ def _validate(name, kind, raw): raise AssertionError(f"unknown type={raw.metadata.type} expect={t}") if x := set(raw.keys()) - _TOP_LEVEL_KEYS: raise AssertionError(f"unknown top level keys={s}") - for x in ("controls_information", "metadata"): + for x in _TOP_LEVEL_KEYS: if y := set(raw[x].keys()) - _KNOWN_KEYS[x]: raise AssertionError(f"unknown {x} keys={y}") if not raw.controls_information.PVs: diff --git a/slicops/pkcli/ioc.py b/slicops/pkcli/ioc.py new file mode 100644 index 0000000..4b9dd48 --- /dev/null +++ b/slicops/pkcli/ioc.py @@ -0,0 +1,90 @@ +"""IOC configured from a YAML file + +:copyright: Copyright (c) 2024 The Board of Trustees of the Leland Stanford Junior University, through SLAC National Accelerator Laboratory (subject to receipt of any required approvals from the U.S. Dept. of Energy). All Rights Reserved. +:license: http://github.com/slaclab/slicops/LICENSE +""" + +from pykern.pkcollections import PKDict +from pykern.pkdebug import pkdc, pkdlog, pkdp, pkdexc +import asyncio +import caproto.server +import pykern.pkio +import pykern.pkyaml +import numpy + + +def run(init_yaml, db_yaml=None): + def _normalize(raw): + for k, v in raw.items(): + if not isinstance(v, dict): + v = PKDict(value=v) + v.pksetdefault(delay=1, value=None, dispatch=PKDict) + yield k, v + + def _pvgroup(config): + return PKDict( + # Need to hardwire the defaults, because ioc_arg_parser uses + # argparse globally which causes a mess with argh (which uses argparse) + pvdb=_PVGroup(config, db_yaml, macros={}, prefix="").pvdb, + interfaces=["0.0.0.0"], + module_name="caproto.asyncio.server", + log_pv_names=False, + ) + + caproto.server.run( + **_pvgroup( + PKDict(_normalize(pykern.pkyaml.load_file(init_yaml))), + ), + ) + + +class _PVGroup(caproto.server.PVGroup): + + def __init__(self, config, db_yaml, *args, **kwargs): + self.__config = config + self.__db_yaml = pykern.pkio.py_path(db_yaml) if db_yaml else None + self.__db = PKDict() + for k, v in self.__config.items(): + self._pvs_[k] = p = caproto.server.pvproperty(value=v.value) + self.__db[k] = v.value + p.__set_name__(self, k) + self.__write_db() + super().__init__(*args, **kwargs) + + async def group_write(self, instance, value, **kwargs): + async def _dispatch(todo): + for k, v in todo.items(): + u = _un_numpy(v[value], k) + if self.__db[k] != u: + self.__db[k] = u + await self.pvdb[k].write(u) + + def _un_numpy(v, name): + if not isinstance(v, numpy.generic): + return v + if isinstance(v, numpy.integer): + return int(v) + if isinstance(v, numpy.floating): + return float(v) + if isinstance(v, numpy.bool_): + return bool(v) + raise AssertionError(f"unhandled type={type(v)} pv={name}") + + try: + if self.__db[instance.pvname] != value: + self.__db[instance.pvname] = _un_numpy(value, instance.pvname) + await _dispatch(self.__config[instance.pvname].dispatch) + self.__write_db() + return await super().group_write(instance, value, **kwargs) + except Exception as e: + pkdlog( + "error={} pv={} value={} stack={}", e, instance.name, value, pkdexc() + ) + raise + + def __write_db(self): + if self.__db_yaml: + pykern.pkio.atomic_write( + self.__db_yaml, + writer=lambda p: pykern.pkyaml.dump_pretty(self.__db, filename=p), + ) diff --git a/slicops/pkcli/yaml_db.py b/slicops/pkcli/yaml_db.py index d3fa393..82e4364 100644 --- a/slicops/pkcli/yaml_db.py +++ b/slicops/pkcli/yaml_db.py @@ -28,10 +28,7 @@ def read(base): Returns: PKDict: values in the db """ - rv = _read(path(base)) - if rv is None: - return PKDict() - return rv + return _read(path(base)) def write(base, *key_value_pairs): @@ -50,26 +47,17 @@ def _pairs(): return key_value_pairs[0].items() return key_value_pairs[0] - def _read_or_new(old, new): - # Atomic read/write - if old.exists(): - old.copy(new) - return _read(new) - return PKDict() - def _validate(): ctx = slicops.ctx.Ctx(base, base) for k, v in _pairs(): yield k, ctx.fields[k].value_set(v) - o = path(base) - n = o.new(ext="tmp" + pykern.util.random_base62()) - try: - rv = _read_or_new(o, n).pkupdate(_validate()) - pykern.pkyaml.dump_pretty(rv, n) - n.rename(o) - finally: - pykern.pkio.unchecked_remove(n) + p = path(base) + rv = _read(p).pkupdate(_validate()) + pykern.pkio.atomic_write( + p, + writer=lambda x: pykern.pkyaml.dump_pretty(rv, filename=x), + ) return rv @@ -80,5 +68,5 @@ def _read(path): except Exception as e: if pykern.pkio.exception_is_not_found(e): pkdlog("ignoring not found path={}", path) - return None + return PKDict() raise diff --git a/slicops/sliclet/screen.py b/slicops/sliclet/screen.py index 38ae155..f8e7cd5 100644 --- a/slicops/sliclet/screen.py +++ b/slicops/sliclet/screen.py @@ -6,7 +6,6 @@ from pykern.pkcollections import PKDict from pykern.pkdebug import pkdc, pkdexc, pkdlog, pkdp -import enum import pykern.pkconfig import pykern.util import queue @@ -90,8 +89,6 @@ def handle_init(self, txn): txn.multi_set(("beam_path.constraints.choices", slicops.device_db.beam_paths())) self.__beam_path_change(txn, None) self.__device_change(txn, None) - - def handle_start(self, txn): b = c = None if pykern.pkconfig.in_dev_mode(): b = _cfg.dev.beam_path @@ -101,13 +98,15 @@ def handle_start(self, txn): txn.field_set("beam_path", b) self.__beam_path_change(txn, b) txn.field_set("camera", c) - self.__device_change(txn, c) + + def handle_start(self, txn): + self.__device_setup(txn, txn.field_get("camera")) def __beam_path_change(self, txn, value): def _choices(): if value is None: return () - return slicops.device_db.device_names(value, _DEVICE_TYPE) + return slicops.device_db.device_names(_DEVICE_TYPE, value) txn.multi_set( ("camera.constraints.choices", _choices()), @@ -132,36 +131,12 @@ def _choices(): self.__device_change(txn, None) def __device_change(self, txn, camera): - def _monitors(): - for n, h in ( - ("image", self.__handle_image), - ("acquire", self.__handle_acquire), - ): - a = self.__device.accessor(n) - m = self.__monitors[n] = _Monitor(a, h) - a.monitor(m) - - def _setup(): - try: - self.__device = self.__device = slicops.device.Device(camera) - _monitors() - except slicops.device.DeviceError as e: - pkdlog( - "error={} setting up {}, clearing; stack={}", e, camera, pkdexc() - ) - self.__device_destroy() - # TODO(robnagler) not clear this is right - raise pykern.util.APIError(e) - - self.__device_destroy() + self.__device_destroy(txn) txn.multi_set(_DEVICE_DISABLE) if camera: - _setup() - txn.multi_set( - _DEVICE_ENABLE + (("pv.value", self.__device.meta.pv_prefix),) - ) + self.__device_setup(txn, camera) - def __device_destroy(self): + def __device_destroy(self, txn=None): if not self.__device: return self.__single_button = False @@ -178,6 +153,27 @@ def __device_destroy(self): pkdlog("destroy device={} error={}", n, e) self.__device = None + def __device_setup(self, txn, camera): + def _monitors(): + for n, h in ( + ("image", self.__handle_image), + ("acquire", self.__handle_acquire), + ): + a = self.__device.accessor(n) + m = self.__monitors[n] = _Monitor(a, h) + a.monitor(m) + + try: + # If there's an epics issues, we have to clear the device + self.__device = self.__device = slicops.device.Device(camera) + _monitors() + except slicops.device.DeviceError as e: + pkdlog("error={} setting up {}, clearing; stack={}", e, camera, pkdexc()) + self.__device_destroy(txn) + self.__user_alert(txn, "unable to connect to camera={} error={}", camera, e) + return + txn.multi_set(_DEVICE_ENABLE + (("pv.value", self.__device.meta.pv_prefix),)) + def __handle_acquire(self, acquire): with self.lock_for_update() as txn: n = not acquire @@ -222,6 +218,14 @@ def __set_acquire(self, txn, acquire): ) raise pykern.util.APIError(e) + # def __target_moved(self, status): + # if status is failed: + # display error + # if status is out: + # disable buttons + # if status is in: + # enable buttons + # def __update_plot(self, txn): if not self.__device: return False @@ -235,6 +239,9 @@ def __update_plot(self, txn): ) return True + def __user_alert(self, txn, fmt, *args): + pkdlog("TODO: USER ALERT: " + fmt, *args) + CLASS = Screen diff --git a/slicops/sliclet/yaml_db.py b/slicops/sliclet/yaml_db.py index 3089778..4ae90c7 100644 --- a/slicops/sliclet/yaml_db.py +++ b/slicops/sliclet/yaml_db.py @@ -34,9 +34,6 @@ def handle_destroy(self): def handle_init(self, txn): self.__db_watcher = None - - def handle_start(self, txn): - # TODO(robnagler) need a separate init for the instance before start self.__db_cache = PKDict() if not self.__read_db(txn): self.__write(txn) diff --git a/slicops/unit_util.py b/slicops/unit_util.py index 78dbd08..2efc58a 100644 --- a/slicops/unit_util.py +++ b/slicops/unit_util.py @@ -7,19 +7,26 @@ # limit imports that might touch config import asyncio import pykern.api.unit_util +import contextlib +# TODO(robnagler) configurable? +_TIMEOUT = 2 -class Setup(pykern.api.unit_util.Setup): + +class SlicletSetup(pykern.api.unit_util.Setup): def __init__(self, sliclet, *args, **kwargs): - super().__init__(*args, **kwargs) self.__sliclet = sliclet + if c := kwargs.get("caproto"): + del kwargs["caproto"] + self.__caproto = c + super().__init__(*args, **kwargs) self.__update_q = asyncio.Queue() async def ctx_update(self): from pykern import pkunit self.__caller() - r = await self.__update_q.get() + r = await asyncio.wait_for(self.__update_q.get(), timeout=_TIMEOUT) self.__update_q.task_done() if r is None: pkunit.pkfail("subscription ended unexpectedly") @@ -51,10 +58,13 @@ def _http_config(self, *args, **kwargs): return config.cfg().ui_api.copy() def _server_config(self, *args, **kwargs): - from slicops import mock_epics - - mock_epics.reset_state() + if self.__caproto: + self.__start_caproto() + else: + from slicops import mock_epics + from pykern import pkdebug + mock_epics.reset_state() return super()._server_config(*args, **kwargs) def _server_start(self, *args, **kwargs): @@ -71,6 +81,9 @@ def __caller(self): c = m.group(1) pkdebug.pkdlog("{} op={}", c, pkinspect.caller_func_name()) + def __start_caproto(self): + pass + async def __subscribe(self): from pykern import pkdebug from pykern.pkcollections import PKDict @@ -81,7 +94,7 @@ async def __subscribe(self): "ui_ctx_update", PKDict(sliclet=self.__sliclet) ) as s: while True: - r = await s.result_get() + r = await asyncio.wait_for(s.result_get(), timeout=_TIMEOUT) self.__update_q.put_nowait(r) if r is None: return @@ -92,3 +105,108 @@ async def __subscribe(self): raise finally: pkdebug.pkdlog("ui_ctx_update subscription ended normally") + + +@contextlib.contextmanager +def random_epics_ports(): + """Open the IOC""" + + from pykern import util, pkconst + import os + import socket + + r = str(util.unbound_localhost_udp_port()) + s = str(util.unbound_localhost_udp_port()) + os.environ.update( + dict( + EPICS_CAS_AUTO_BEACON_ADDR_LIST="no", + EPICS_CAS_BEACON_ADDR_LIST=pkconst.LOCALHOST_IP, + EPICS_CAS_BEACON_PORT=r, + EPICS_CAS_INTF_ADDR_LIST=pkconst.LOCALHOST_IP, + EPICS_CAS_SERVER_PORT=s, + EPICS_CA_ADDR_LIST=pkconst.LOCALHOST_IP, + EPICS_CA_AUTO_ADDR_LIST="no", + EPICS_CA_REPEATER_PORT=r, + EPICS_CA_SERVER_PORT=s, + ) + ) + with socket.socket(socket.AF_INET, socket.SOCK_DGRAM) as s: + s.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) + s.bind(("127.0.0.1", int(r))) + yield None + + +@contextlib.contextmanager +def setup_screen(beam_path, device_name): + with start_ioc("ioc.yaml"): + from slicops.device import screen + from pykern.pkcollections import PKDict + + rv = PKDict(handler=_screen_handler()) + rv.device = screen.Screen(beam_path, device_name, rv.handler) + try: + yield rv + finally: + rv.device.destroy() + + +@contextlib.contextmanager +def start_ioc(init_yaml, db_yaml=None): + import os, signal, time + import socket + + with random_epics_ports(): + p = os.fork() + if p == 0: + from pykern import pkdebug + + try: + from slicops.pkcli import ioc + from pykern import pkunit + + ioc.run( + pkunit.data_dir().join(init_yaml), + db_yaml=(pkunit.work_dir().join(db_yaml) if db_yaml else None), + ) + except Exception as e: + pkdebug.pkdlog("server exception={} stack={}", e, pkdebug.pkdexc()) + finally: + os._exit(0) + try: + time.sleep(1) + yield None + finally: + os.kill(p, signal.SIGKILL) + + +def _screen_handler(): + from pykern import pkunit + from pykern.pkcollections import PKDict + import queue + from slicops.device import screen + + class _Handler(screen.EventHandler): + def __init__(self, *args, **kwargs): + super().__init__(*args, **kwargs) + self.event_q = PKDict( + { + k: queue.Queue() + for k in ("acquire", "image", "target_status", "error") + } + ) + + def on_screen_device_error(self, **kwargs): + self.event_q.error.put_nowait(PKDict(kwargs)) + + def on_screen_device_update(self, **kwargs): + self.event_q[kwargs["accessor_name"]].put_nowait(PKDict(kwargs)) + + def test_get(self, event_name): + try: + rv = self.event_q[event_name].get(timeout=_TIMEOUT) + # Errors don't have value + return rv.get("value", rv) + except queue.Empty: + pkunit.pkfail("timeout event={}", event_name) + + return _Handler() diff --git a/tests/device/screen1_data/ioc.yaml b/tests/device/screen1_data/ioc.yaml new file mode 100644 index 0000000..2637c73 --- /dev/null +++ b/tests/device/screen1_data/ioc.yaml @@ -0,0 +1,18 @@ +--- +YAGS:IN20:351:Acquire: + value: 0 + dispatch: + YAGS:IN20:351:IMAGE: + 0: [0, 0, 0, 0] + 1: [0, 1, 2, 3] +YAGS:IN20:351:IMAGE: [0, 0, 0, 0] +YAGS:IN20:351:N_OF_ROW: 2 +YAGS:IN20:351:N_OF_COL: 2 +YAGS:IN20:351:PNEUMATIC: + dispatch: + YAGS:IN20:351:TGT_STS: + 0: 1 + 1: 2 +YAGS:IN20:351:TGT_STS: 1 +YAGS:IN20:211:TGT_STS: 1 +YAGS:IN20:241:TGT_STS: 1 diff --git a/tests/device/screen1_test.py b/tests/device/screen1_test.py new file mode 100644 index 0000000..0beee91 --- /dev/null +++ b/tests/device/screen1_test.py @@ -0,0 +1,20 @@ +"""Test slicops.device.screen + +:copyright: Copyright (c) 2025 The Board of Trustees of the Leland Stanford Junior University, through SLAC National Accelerator Laboratory (subject to receipt of any required approvals from the U.S. Dept. of Energy). All Rights Reserved. +:license: http://github.com/slaclab/slicops/LICENSE +""" + +from slicops import unit_util + + +def test_upstream_ok(): + from pykern import pkdebug, pkunit + + with unit_util.setup_screen("CU_HXR", "YAG03") as s: + s.handler.test_get("image") + pkunit.pkeq(False, s.handler.test_get("acquire")) + pkunit.pkeq(False, s.handler.test_get("target_status")) + s.device.move_target(want_in=True) + pkunit.pkeq(True, s.handler.test_get("target_status")) + s.device.move_target(want_in=False) + pkunit.pkeq(False, s.handler.test_get("target_status")) diff --git a/tests/device/screen2_data/ioc.yaml b/tests/device/screen2_data/ioc.yaml new file mode 100644 index 0000000..b3967ea --- /dev/null +++ b/tests/device/screen2_data/ioc.yaml @@ -0,0 +1,19 @@ +--- +YAGS:IN20:351:Acquire: + value: 0 + dispatch: + YAGS:IN20:351:IMAGE: + 0: [0, 0, 0, 0] + 1: [0, 1, 2, 3] +YAGS:IN20:351:IMAGE: [0, 0, 0, 0] +YAGS:IN20:351:N_OF_ROW: 2 +YAGS:IN20:351:N_OF_COL: 2 +YAGS:IN20:351:PNEUMATIC: + dispatch: + YAGS:IN20:351:TGT_STS: + 0: 1 + 1: 2 +YAGS:IN20:351:TGT_STS: 1 +YAGS:IN20:211:TGT_STS: 1 +# Upstream device is in +YAGS:IN20:241:TGT_STS: 2 diff --git a/tests/device/screen2_test.py b/tests/device/screen2_test.py new file mode 100644 index 0000000..eddf68e --- /dev/null +++ b/tests/device/screen2_test.py @@ -0,0 +1,16 @@ +"""Test slicops.device.screen + +:copyright: Copyright (c) 2025 The Board of Trustees of the Leland Stanford Junior University, through SLAC National Accelerator Laboratory (subject to receipt of any required approvals from the U.S. Dept. of Energy). All Rights Reserved. +:license: http://github.com/slaclab/slicops/LICENSE +""" + + +def test_upstream_blocked(): + from pykern import pkdebug, pkunit + from slicops import unit_util + + with unit_util.setup_screen("CU_HXR", "YAG03") as s: + s.device.move_target(want_in=True) + e = s.handler.test_get("error") + pkunit.pkeq("upstream", e.error_kind) + pkunit.pkeq("upstream target is in", e.error_msg.YAG02) diff --git a/tests/device_db_test.py b/tests/device_db_test.py index 5e578b9..11b85b5 100644 --- a/tests/device_db_test.py +++ b/tests/device_db_test.py @@ -15,14 +15,14 @@ def test_basic(): pkunit.pkeq("SC_SXR", a[-1]) pkunit.pkeq(18, len(a)) - a = device_db.device_names("SC_SXR", "PROF") + a = device_db.device_names("PROF", "SC_SXR") pkunit.pkeq("BOD10", a[0]) pkunit.pkeq("YAGH2", a[-1]) pkunit.pkeq(11, len(a)) with pkunit.pkexcept("XYZZY"): - device_db.device_names("XYZZY", "PROF") + device_db.device_names("PROF", "XYZZY") with pkunit.pkexcept("xyzzy"): - device_db.device_names("SC_SXR", "xyzzy") + device_db.device_names("xyzzy", "SC_SXR") a = device_db.meta_for_device("VCCB") pkunit.pkeq("CAMR:LGUN:950:Image:ArrayData", a.accessor.image.pv_name) @@ -32,3 +32,13 @@ def test_basic(): # YAG01B does not have any PVs so not in db with pkunit.pkexcept("NoRows"): device_db.meta_for_device("YAG01B") + + +def test_upstream(): + from pykern import pkdebug, pkunit + from slicops import device_db + + a = device_db.upstream_devices("PROF", "target_control", "CU_HXR", "OTR11") + pkunit.pkeq(9, len(a)) + pkunit.pkeq("YAG01", a[0], "Lowest Z Prof") + pkunit.pkeq("OTR4", a[-1], "Closest Z Prof") diff --git a/tests/pkcli/ioc_data/init.yaml b/tests/pkcli/ioc_data/init.yaml new file mode 100644 index 0000000..5feecec --- /dev/null +++ b/tests/pkcli/ioc_data/init.yaml @@ -0,0 +1,7 @@ +WIRE:HTR:340:MOTR_ENABLED_STS: 33 +WIRE:HTR:340:STARTSCAN: + value: 0 + dispatch: + WIRE:HTR:340:MOTR_ENABLED_STS: + 0: 33 + 1: 66 diff --git a/tests/pkcli/ioc_test.py b/tests/pkcli/ioc_test.py new file mode 100644 index 0000000..0295cd5 --- /dev/null +++ b/tests/pkcli/ioc_test.py @@ -0,0 +1,31 @@ +"""Test ioc + +:copyright: Copyright (c) 2025 The Board of Trustees of the Leland Stanford Junior University, through SLAC National Accelerator Laboratory (subject to receipt of any required approvals from the U.S. Dept. of Energy). All Rights Reserved. +:license: http://github.com/slaclab/slicops/LICENSE +""" + +_DB = """WIRE:HTR:340:MOTR_ENABLED_STS: {} +WIRE:HTR:340:STARTSCAN: {} +""" + + +def test_db_yaml(): + from slicops import unit_util + + with unit_util.start_ioc("init.yaml", db_yaml="db.yaml"): + from slicops import device + from pykern import pkdebug, pkunit + import time + + b = pkunit.work_dir().join("db.yaml") + d = device.Device("WS0H04") + try: + pkunit.pkeq(33, d.get("enabled")) + pkunit.pkeq(_DB.format(33, 0), b.read("rt")) + d.put("start_scan", 1) + # Allow the dispatch to come back + time.sleep(0.1) + pkunit.pkeq(66, d.get("enabled")) + pkunit.pkeq(_DB.format(66, 1), b.read("rt")) + finally: + d.destroy() diff --git a/tests/sliclet/screen_test.py b/tests/sliclet/screen_test.py index f463a5d..f4cb07f 100644 --- a/tests/sliclet/screen_test.py +++ b/tests/sliclet/screen_test.py @@ -6,6 +6,8 @@ import pytest +_BUTTONS = tuple(f"{k}_button.ui.enabled" for k in ("start", "stop", "single")) + @pytest.mark.asyncio(loop_scope="module") async def test_basic(): @@ -13,24 +15,30 @@ async def test_basic(): async def _buttons(s, expect, msg): from pykern import pkunit, pkdebug + from asyncio.exceptions import CancelledError - rv = await s.ctx_update() - for k, a in zip(("start", "stop", "single"), expect): - x = f"{k}_button.ui.enabled" - pkunit.pkeq( - rv.fields.pkunchecked_nested_get(x), a, "field={} expect={}", x, msg - ) + # Wait for buttons to "settle" on expect. The updates + # are async so we can't control which update returns what. + while True: + try: + rv = await s.ctx_update() + except CancelledError: + # timed out so now report mismatch via pkunit + pkunit.pkeq(expect, v, msg) + v = tuple(rv.fields.pknested_get(k) for k in _BUTTONS) + if v == expect: + break return rv - async with unit_util.Setup("screen") as s: + async with unit_util.SlicletSetup("screen") as s: from pykern import pkunit, pkdebug from pykern.pkdebug import pkdc, pkdexc, pkdlog, pkdp import asyncio - r = await s.ctx_update() r = await s.ctx_update() pkunit.pkeq("DEV_BEAM_PATH", r.fields.beam_path.value) pkunit.pkeq("DEV_CAMERA", r.fields.camera.value) + await _buttons(s, (True, False, True), "start/single enabled") await s.ctx_field_set(start_button=None) await _buttons(s, (False, False, False), "all disabled after start") await _buttons(s, (False, True, False), "acquire should fire") diff --git a/tests/sliclet/yaml_db_test.py b/tests/sliclet/yaml_db_test.py index 9d1f865..09ed445 100644 --- a/tests/sliclet/yaml_db_test.py +++ b/tests/sliclet/yaml_db_test.py @@ -11,7 +11,7 @@ async def test_basic(): from slicops import unit_util - async with unit_util.Setup("yaml_db") as s: + async with unit_util.SlicletSetup("yaml_db") as s: from pykern import pkunit, pkdebug from slicops.pkcli import yaml_db import asyncio