diff --git a/.gitignore b/.gitignore index 7ac465e..c648774 100644 --- a/.gitignore +++ b/.gitignore @@ -10,6 +10,7 @@ tox.ini \#* .idea/ .vscode/ +*~ # Byte-compiled / optimized / DLL files __pycache__/ diff --git a/slicops/ctx.py b/slicops/ctx.py index ed2c9e7..5a6e4a6 100644 --- a/slicops/ctx.py +++ b/slicops/ctx.py @@ -168,14 +168,16 @@ def group_get(self, field, group, attr=None): def multi_set(self, *args): def _args(): if len(args) > 1: + # (("a", 1), ("b", 2), ..) return args if len(args) == 0: raise AssetionError("must be at list one update") - if isinstance(args[0][0], str): - # (("a", 1)) - return args - # ((("a", 1), ("b", 2), ..)) or a dict - return args[0] + rv = args[0] + # ({"a": 1, "b": 2, ...}) + if isinstance(rv, dict): + return rv.items() + # else ((("a", 1), ("b", 2), ..)) + return args if isinstance(rv[0], str) else rv def _parse(): rv = PKDict() @@ -184,6 +186,8 @@ def _parse(): return rv for k, v in _parse().items(): + if not isinstance(v, PKDict): + v = PKDict(value=v) self.__field_update(k, self.__field(k), v) def rollback(self): diff --git a/slicops/device/screen.py b/slicops/device/screen.py index cde846b..b212a2d 100644 --- a/slicops/device/screen.py +++ b/slicops/device/screen.py @@ -5,8 +5,8 @@ """ from pykern.pkcollections import PKDict -from pykern.pkdebug import pkdc, pkdexc, pkdlog, pkdp -from slicops.device import DeviceError +from pykern.pkdebug import pkdc, pkdexc, pkdlog, pkdp, pkdformat +import pykern.pkasyncio import abc import enum import logging @@ -14,14 +14,11 @@ 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" +_BLOCKING_MSG = "upstream target is {}" _TIMEOUT_MSG = "upstream target status accessor timed out" _ERROR_PREFIX_MSG = "upstream target error: " @@ -47,9 +44,7 @@ def move_target(self, want_in): 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) - ) + self.__worker.req_action("req_move_target", PKDict(want_in=want_in)) class ErrorKind(enum.Enum): @@ -64,7 +59,7 @@ class EventHandler: """Clients of DeviceScreen must implement this""" @abc.abstractmethod - def on_screen_device_error(self, accessor_name, error_kind, error_msg): + def on_screen_device_error(self, exc): pass @abc.abstractmethod @@ -72,98 +67,6 @@ 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): - """Queue ``method`` to be called in loop thread. - - Actions are methods that (by convention) begin with - ``action_`` and are called sequentially inside `_start`. A - lock is used to prevent `destroy` being called during the action. - - Actions return ``None`` to continue on to the next - action. `_LOOP_END` should be returned to terminate `_start` - (the loop) in which case no further actions are - performed. Actions can return a callable that will be called - inside the loop and outside the lock. These returned callables - are known as external callbacks, that is, functions that may - do anything so holding the lock could be problematic. - - Args: - method (callable): see above - arg (object): passed verbatim to ``method`` - - """ - self.__actions.put_nowait((method, arg)) - - def destroy(self): - """Stops thread and calls subclass `_destroy` - - THREADING: subclasses should not call destroy directly. They should - return `_LOOP_END` instead. External callbacks may call destroy, because - _ActionLoop does not hold lock during external callbacks. - """ - 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: - with self.__lock: - if self.destroyed: - return - 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 - # Do not need to check m, because only invalid when destroyed is True - if (m := m(a)) is self._LOOP_END: - return - # Will be true if destroy called inside action (m) - if self.destroyed: - return - # Action returned an external callback, which must occur outside lock - if m: - m() - except Exception as e: - pkdlog("error={} {} stack={}", e, self, pkdexc(simplify=True)) - finally: - self.destroy() - - class _FSM: """Finite State Machine called by `_Worker` exclusively @@ -173,11 +76,11 @@ class _FSM: def __init__(self, worker): self.worker = worker self.curr = PKDict( - acquire=False, - check_upstream=False, - move_target_arg=None, - target_status=None, - upstream_problems=None, + acquire=False, # status of screen acquire + move_target_arg=None, # where do we want the target to be? + target_status=None, # where is the status right now? + await_upstream=False, # are we checking upstream? + upstream_problems=None, # are there problems upstream? ) self.prev = self.curr.copy() @@ -186,13 +89,24 @@ def event(self, name, arg): if u := getattr(self, f"_event_{name}")(arg, **self.curr): self.curr.update(u) - def _event_handle_monitor(self, arg, **kwargs): + def _event_handle_monitor( + self, + arg, + await_upstream, + move_target_arg, + target_status, + upstream_problems, + **kwargs, + ): n = arg.accessor.accessor_name if "error" in arg: self.worker.action( - self.worker.action_call_handler, - PKDict( - error_kind=ErrorKind.monitor, accessor_name=n, error_msg=arg.error + "call_handler", + ScreenError( + device=self.worker.device.device_name, + error_kind=ErrorKind.monitor, + accessor_name=n, + error_msg=arg.error, ), ) if n == "target_status": @@ -208,53 +122,81 @@ def _event_handle_monitor(self, arg, **kwargs): 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) + v = TargetStatus(arg.value) + rv = PKDict(target_status=v) + if target_status is None and move_target_arg: + rv.await_upstream = self.__move_target_upstream_check( + move_target_arg, upstream_problems, await_upstream + ) + else: + rv.move_target_arg = None else: raise AssertionError(f"unsupported accessor={n} {self}") - self.worker.action( - self.worker.action_call_handler, PKDict(accessor_name=n, value=v) - ) + self.worker.action("call_handler", PKDict(accessor_name=n, value=v)) return rv def _event_move_target( self, arg, - check_upstream, + await_upstream, move_target_arg, target_status, upstream_problems, **kwargs, ): - if move_target_arg: + # If target_status hasn't initialized, defer to monitor fire. + if target_status == None: + rv = PKDict(move_target_arg=arg) + return rv + if move_target_arg or target_status in ( + TargetStatus.MOVING, + TargetStatus.INCONSISTENT, + ): self.worker.action( - self.worker.action_call_handler, - PKDict(error_kind=ErrorKind.fsm, error_msg="target already moving"), + "call_handler", + ScreenError( + device=self.worker.device.device_name, + error_kind=ErrorKind.fsm, + error_msg="target already moving, inconsistent, or not intialized", + ), ) return - if target_status is not None and arg.want_in == target_status: + if arg.want_in == (target_status == TargetStatus.IN): # 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) + rv.await_upstream = self.__move_target_upstream_check( + arg.want_in, upstream_problems, await_upstream + ) return rv def _event_upstream_status(self, arg, move_target_arg, **kwargs): - rv = PKDict(check_upstream=False, upstream_problems=arg.problems) + rv = PKDict(await_upstream=False, upstream_problems=arg.problems) if arg.problems: self.worker.action( - self.worker.action_call_handler, - PKDict(error_kind=ErrorKind.upstream, error_msg=arg.problems), + "call_handler", + ScreenError( + device=self.worker.device.device_name, + error_kind=ErrorKind.upstream, + error_msg=arg.problems, + ), ) return rv.pkupdate(move_target_arg=None) - self.worker.action(self.worker.action_move_target, move_target_arg) + self.worker.action("move_target", move_target_arg) + return rv + + def __move_target_upstream_check(self, want_in, upstream_problems, await_upstream): + rv = False + if want_in and upstream_problems is None or upstream_problems: + # Recheck the upstream + if not await_upstream: + self.worker.action("check_upstream", None) + rv = True + else: + arg = PKDict(want_in=want_in) + self.worker.action("move_target", arg) return rv def __repr__(self): @@ -264,7 +206,27 @@ def _states(curr): return f"<_FSM {self.worker.device.device_name} {_states(self.curr)}>" -class _Upstream(_ActionLoop): +class ScreenError(Exception): + def __init__(self, **kwargs): + def _arg_str(): + return pkdformat( + " ".join(k + "={" + k + "}" for k in sorted(kwargs)), + **kwargs, + ) + + super().__init__(_arg_str()) + + +class TargetStatus(enum.Enum): + """Errors passed to on_screen_device_error""" + + INCONSISTENT = 3 + IN = 2 + OUT = 1 + MOVING = 0 + + +class _Upstream(pykern.pkasyncio.ActionLoop): """Action loop to check targets of upstream screens""" def __init__(self, worker): @@ -276,17 +238,22 @@ def _names(): self.__worker = worker self.__problems = PKDict() self.__devices = PKDict({u: slicops.device.Device(u) for u in _names()}) + if len(self.__devices) == 0: + self.__done() + self._destroy() + return self._loop_timeout_secs = _cfg.upstream_timeout_secs super().__init__() - def action_handle_status(self, arg): + def action_handle_target_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 + elif arg.value != TargetStatus.OUT.value: + s = TargetStatus(arg.value) + self.__problems[n] = _BLOCKING_MSG.format(s.name) if not self.__devices: return self.__done() return None @@ -302,26 +269,24 @@ def _destroy(self): x.destroy() def __done(self): - self.__worker.action( - self.__worker.action_upstream_status, PKDict(problems=self.__problems) - ) + self.__worker.action("upstream_status", PKDict(problems=self.__problems)) return self._LOOP_END - def __handle_status(self, kwargs): + def __handle_target_status(self, kwargs): if "connected" in kwargs: return - self.action(self.action_handle_status, kwargs) + self.action("handle_target_status", kwargs) def _start(self, *args, **kwargs): for d in self.__devices.values(): - d.accessor("target_status").monitor(self.__handle_status) + d.accessor("target_status").monitor(self.__handle_target_status) super()._start(*args, **kwargs) def _repr(self): return f"pending={sorted(self.__devices)} problems={sorted(self.__problems)}" -class _Worker(_ActionLoop): +class _Worker(pykern.pkasyncio.ActionLoop): """Action loop for Screen _Worker uses `_FSM` to translate events to actions. Monitor calls @@ -336,22 +301,31 @@ def __init__(self, beam_path, handler, device): self.__handler = handler self.__upstream = None self.__status = None - self.__fsm = _FSM(self) + # self.monitors = pkdict... + self.__fsm = _FSM(self) # self.monitors ... self.__target_control = None self._loop_timeout_secs = 0 super().__init__() + # get from ready queue with timeout + # except + # exit + def action_call_handler(self, arg): m = ( self.__handler.on_screen_device_error - if "error_kind" in arg + if isinstance(arg, Exception) else self.__handler.on_screen_device_update ) # Denormalized state so no need for lock during call - return lambda: m(**arg) + if isinstance(arg, dict): + return lambda: m(**arg) + else: + return lambda: m(arg) def action_check_upstream(self, arg): - self.__upstream = _Upstream(self) + if self.__upstream is None or self.__upstream.destroyed: + self.__upstream = _Upstream(self) return None def action_handle_monitor(self, arg): @@ -375,6 +349,7 @@ def action_upstream_status(self, arg): def req_action(self, method, arg): """Called by DeviceScreen which has separate life cycle""" + # __fsm.is_ready if self.destroyed: raise AssertionError("object is destroyed") self.action(method, arg) @@ -384,12 +359,22 @@ def _destroy(self): (u, self.__upstream) = (self.__upstream, None) u.destroy() + def _handle_exception(self, exc): + self.__handler.on_screen_device_error( + ScreenError( + device=self.device.device_name, + error=exc, + ) + ) + def __handle_monitor(self, change): - self.action(self.action_handle_monitor, change) + self.action("handle_monitor", change) def _start(self, *args, **kwargs): - for a in "acquire", "image", "target_status": + for a in "acquire", "image": # self.monitors ... self.device.accessor(a).monitor(self.__handle_monitor) + if self.device.has_accessor("target_status"): + self.device.accessor("target_status").monitor(self.__handle_monitor) super()._start(*args, **kwargs) def _repr(self): diff --git a/slicops/package_data/device_db.sqlite3 b/slicops/package_data/device_db.sqlite3 index 21564c9..ab85f97 100644 Binary files a/slicops/package_data/device_db.sqlite3 and b/slicops/package_data/device_db.sqlite3 differ diff --git a/slicops/package_data/sliclet/screen.yaml b/slicops/package_data/sliclet/screen.yaml index 7c07650..0c7e516 100644 --- a/slicops/package_data/sliclet/screen.yaml +++ b/slicops/package_data/sliclet/screen.yaml @@ -35,6 +35,12 @@ fields: label: PV writable: false widget: static + target_status: + prototype: String + ui: + label: Target Status + writable: false + widget: static single_button: prototype: Button ui: @@ -50,6 +56,16 @@ fields: ui: css_kind: danger label: Stop + target_in_button: + prototype: Button + ui: + css_kind: primary + label: In + target_out_button: + prototype: Button + ui: + css_kind: primary + label: Out ui_layout: - cols: @@ -62,6 +78,10 @@ ui_layout: - start_button - stop_button - single_button + - cell_group: + - target_in_button + - target_out_button + - target_status - css: col-sm-9 col-xxl-7 rows: - plot diff --git a/slicops/pkcli/device_db.py b/slicops/pkcli/device_db.py index e696cfa..c1c965f 100644 --- a/slicops/pkcli/device_db.py +++ b/slicops/pkcli/device_db.py @@ -26,6 +26,23 @@ _DEV_YAML = """ screens: DEV_CAMERA: + controls_information: + PVs: + acquire: 13SIM1:cam1:Acquire + image: 13SIM1:image1:ArrayData + n_col: 13SIM1:cam1:SizeX + n_row: 13SIM1:cam1:SizeY + n_bits: 13SIM1:cam1:N_OF_BITS + target_status: 13SIM1:cam1:ShutterMode + target_control: 13SIM1:cam1:TriggerMode + control_name: 13SIM1 + metadata: + area: DEV_AREA + beam_path: + - DEV_BEAM_PATH + sum_l_meters: 0.614 + type: PROF + DEV_CAMERA2: controls_information: PVs: acquire: 13SIM1:cam1:Acquire diff --git a/slicops/pkcli/epics.py b/slicops/pkcli/epics.py index 18235ce..75fb943 100644 --- a/slicops/pkcli/epics.py +++ b/slicops/pkcli/epics.py @@ -11,11 +11,15 @@ import pykern.pkcli import pykern.pkio import subprocess +import threading import time # Local so should connect quickly _SIM_DETECTOR_TIMEOUT = 5 _LOG_BASE = "sim_detector.log" +_CAM1_STATUS_PV = "13SIM1:cam1:ShutterMode" +_CAM1_CONTROL_PV = "13SIM1:cam1:TriggerMode" +_INITIAL_TARGET_STATUS = 1 def init_sim_detector(): @@ -37,6 +41,8 @@ def init_sim_detector(): "13SIM1:cam1:SizeX": 1024, "13SIM1:cam1:SizeY": 768, "13SIM1:image1:EnableCallbacks": 1, + _CAM1_STATUS_PV: _INITIAL_TARGET_STATUS, + _CAM1_CONTROL_PV: 0, }.items(): pv = epics.PV(name) v = pv.put(value, wait=True, timeout=_SIM_DETECTOR_TIMEOUT) @@ -83,6 +89,33 @@ def _st_cmd(dir_path): "rb" ) + def _watch_target(): + + def _control_monitor(pvname, value, **kwargs): + # control value: 0: move out, 1: move in + # status: 1: out, 2: in + state.new_status = value + 1 + state.trigger.set() + + def _watch_status(): + while True: + state.trigger.wait() + state.trigger.clear() + if state.status != state.new_status: + state.status_pv.put(0, wait=True, timeout=3) + time.sleep(2) + state.status = state.new_status + state.status_pv.put(state.status, wait=True, timeout=3) + + state = PKDict( + new_status=_INITIAL_TARGET_STATUS, + status=_INITIAL_TARGET_STATUS, + status_pv=epics.PV(_CAM1_STATUS_PV), + trigger=threading.Event(), + ) + epics.PV(_CAM1_CONTROL_PV).add_callback(_control_monitor) + threading.Thread(target=_watch_status, daemon=True).start() + d = _dir() with _log() as o: p = subprocess.Popen( @@ -100,8 +133,11 @@ def _st_cmd(dir_path): time.sleep(2) pkdlog("initializing sim detector") init_sim_detector() + _watch_target() pkdlog("waiting for pid={} to exit", p.pid) p.wait() + except KeyboardInterrupt: + pass finally: if p.poll() is not None: p.terminate() diff --git a/slicops/pkcli/ioc.py b/slicops/pkcli/ioc.py index 4b9dd48..de01786 100644 --- a/slicops/pkcli/ioc.py +++ b/slicops/pkcli/ioc.py @@ -8,12 +8,21 @@ from pykern.pkdebug import pkdc, pkdlog, pkdp, pkdexc import asyncio import caproto.server +import numpy +import pykern.fconf import pykern.pkio import pykern.pkyaml -import numpy def run(init_yaml, db_yaml=None): + def _fconf(): + p = pykern.pkio.py_path(init_yaml) + if p.check(dir=True): + return pykern.fconf.parse_all(p) + if p.check() and p.ext: + return pykern.fconf.Parser([p]).result + return pykern.fconf.parse_all(path=p.dirpath(), glob=p.basename + "*") + def _normalize(raw): for k, v in raw.items(): if not isinstance(v, dict): @@ -31,11 +40,7 @@ def _pvgroup(config): log_pv_names=False, ) - caproto.server.run( - **_pvgroup( - PKDict(_normalize(pykern.pkyaml.load_file(init_yaml))), - ), - ) + caproto.server.run(**_pvgroup(PKDict(_normalize(_fconf())))) class _PVGroup(caproto.server.PVGroup): diff --git a/slicops/sliclet/__init__.py b/slicops/sliclet/__init__.py index 2a6028e..06d20f0 100644 --- a/slicops/sliclet/__init__.py +++ b/slicops/sliclet/__init__.py @@ -106,12 +106,15 @@ def lock_for_update(self, log_op=None): if log_op: d += f" op={log_op}" except Exception as e2: - pkdlog("error={} during exception stack={}", e2, pkdexc()) + pkdlog("error={} during exception stack={}", e2, pkdexc(simplify=True)) if not isinstance(e, pykern.util.APIError): - pkdlog("stack={}", pkdexc()) + pkdlog("stack={}", pkdexc(simplify=True)) pkdlog("ERROR {}", d) self.__put_work(_Work.error, PKDict(desc=d)) + def put_exception(self, exc): + self.__put_work(_Work.error, exc) + def session_end(self): self.__put_work(_Work.session_end, None) diff --git a/slicops/sliclet/screen.py b/slicops/sliclet/screen.py index a0bb81c..105e265 100644 --- a/slicops/sliclet/screen.py +++ b/slicops/sliclet/screen.py @@ -6,10 +6,12 @@ from pykern.pkcollections import PKDict from pykern.pkdebug import pkdc, pkdexc, pkdlog, pkdp +from slicops.device.screen import TargetStatus import pykern.pkconfig import pykern.util import queue import slicops.device +import slicops.device.screen import slicops.device_db import slicops.plot import slicops.sliclet @@ -20,36 +22,60 @@ _cfg = None _BUTTONS_DISABLE = ( - ("single_button.ui.enabled", False), - ("stop_button.ui.enabled", False), ("start_button.ui.enabled", False), + ("stop_button.ui.enabled", False), + ("single_button.ui.enabled", False), ) -_DEVICE_DISABLE = ( - ("color_map.ui.enabled", False), - ("color_map.ui.visible", False), - ("curve_fit_method.ui.enabled", False), - ("curve_fit_method.ui.visible", False), - ("plot.ui.visible", False), - # Useful to avoid large ctx sends - ("plot.value", None), - ("pv.ui.visible", False), - ("pv.value", None), +_TARGET_DISABLE = ( + ("target_in_button.ui.enabled", False), + ("target_out_button.ui.enabled", False), +) + +_TARGET_INVISIBLE = ( + ("target_in_button.ui.visible", False), + ("target_out_button.ui.visible", False), + ("target_status.ui.visible", False), +) + +_TARGET_VISIBLE = ( + ("target_in_button.ui.visible", True), + ("target_out_button.ui.visible", True), + ("target_status.ui.visible", True), +) + +_BUTTONS_INVISIBLE = ( ("single_button.ui.visible", False), ("start_button.ui.visible", False), ("stop_button.ui.visible", False), -) + _BUTTONS_DISABLE +) -_DEVICE_ENABLE = ( - ("pv.ui.visible", True), +_BUTTONS_VISIBLE = ( ("single_button.ui.visible", True), ("start_button.ui.visible", True), ("stop_button.ui.visible", True), - ("single_button.ui.enabled", True), - ("stop_button.ui.enabled", False), - ("start_button.ui.enabled", True), ) +_DEVICE_DISABLE = ( + ( + ("color_map.ui.enabled", False), + ("color_map.ui.visible", False), + ("curve_fit_method.ui.enabled", False), + ("curve_fit_method.ui.visible", False), + ("plot.ui.visible", False), + # Useful to avoid large ctx sends + ("plot.value", None), + ("pv.ui.visible", False), + ("pv.value", None), + ) + + _BUTTONS_DISABLE + + _BUTTONS_INVISIBLE + + _TARGET_DISABLE + + _TARGET_INVISIBLE +) + +_DEVICE_ENABLE = (("pv.ui.visible", True),) + _BUTTONS_VISIBLE + _PLOT_ENABLE = ( ("color_map.ui.enabled", True), ("color_map.ui.visible", True), @@ -60,11 +86,15 @@ class Screen(slicops.sliclet.Base): + def __init__(self, *args): + self.__current_value = PKDict(acquire=None, image=None, target=None) + super().__init__(*args) + def handle_destroy(self): self.__device_destroy() def on_change_camera(self, txn, value, **kwargs): - self.__device_change(txn, value) + self.__device_change(txn, txn.field_get("beam_path"), value) def on_change_beam_path(self, txn, value, **kwargs): self.__beam_path_change(txn, value) @@ -74,21 +104,27 @@ def on_change_curve_fit_method(self, txn, **kwargs): def on_click_single_button(self, txn, **kwargs): self.__single_button = True - self.__set_acquire(txn, True) + self.__set(txn, "acquire", True, _BUTTONS_DISABLE) def on_click_start_button(self, txn, **kwargs): - self.__set_acquire(txn, True) + self.__set(txn, "acquire", True, _BUTTONS_DISABLE) def on_click_stop_button(self, txn, **kwargs): - self.__set_acquire(txn, False) + self.__set(txn, "acquire", False, _BUTTONS_DISABLE) + + def on_click_target_in_button(self, txn, **kwargs): + self.__set(txn, "target", True, _TARGET_DISABLE, method="move_target") + + def on_click_target_out_button(self, txn, **kwargs): + self.__set(txn, "target", False, _TARGET_DISABLE, method="move_target") def handle_init(self, txn): self.__device = None - self.__monitors = PKDict() + self.__handler = None self.__single_button = False txn.multi_set(("beam_path.constraints.choices", slicops.device_db.beam_paths())) self.__beam_path_change(txn, None) - self.__device_change(txn, None) + self.__device_change(txn, None, None) b = c = None if pykern.pkconfig.in_dev_mode(): b = _cfg.dev.beam_path @@ -100,7 +136,7 @@ def handle_init(self, txn): txn.field_set("camera", c) def handle_start(self, txn): - self.__device_setup(txn, txn.field_get("camera")) + self.__device_setup(txn, txn.field_get("beam_path"), txn.field_get("camera")) def __beam_path_change(self, txn, value): def _choices(): @@ -128,20 +164,19 @@ def _choices(): # Camera is the same so restore the value, no device change txn.field_set("camera", c) else: - self.__device_change(txn, None) + self.__device_change(txn, value, None) - def __device_change(self, txn, camera): + def __device_change(self, txn, beam_path, camera): self.__device_destroy(txn) txn.multi_set(_DEVICE_DISABLE) - self.__device_setup(txn, camera) + self.__device_setup(txn, beam_path, camera) def __device_destroy(self, txn=None): if not self.__device: return self.__single_button = False - for m in self.__monitors.values(): - m.destroy() - self.__monitors = PKDict() + self.__handler.destroy() + self.__handler = None try: n = self.__device.device_name except Exception: @@ -152,31 +187,38 @@ def __device_destroy(self, txn=None): 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) + def __device_setup(self, txn, beam_path, camera): + self.__handler = _Handler( + self.__handle_device_error, + PKDict( + image=self.__handle_image, + acquire=self.__handle_acquire, + target_status=self.__handle_target_status, + ), + ) if camera is None: return try: # If there's an epics issues, we have to clear the device - self.__device = self.__device = slicops.device.Device(camera) - _monitors() + self.__device = slicops.device.screen.Screen( + beam_path, + camera, + self.__handler, + ) 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),)) + s = PKDict(_DEVICE_ENABLE + (("pv.value", self.__device.meta.pv_prefix),)) + if self.__device.has_accessor("target_status"): + s.update(_TARGET_VISIBLE) + txn.multi_set(s) def __handle_acquire(self, acquire): with self.lock_for_update() as txn: + self.__current_value["acquire"] = acquire n = not acquire # Leave plot alone txn.multi_set( @@ -190,47 +232,54 @@ def __handle_acquire(self, acquire): if not acquire: self.__single_button = False + def __handle_device_error(self, exc): + self.put_exception(exc) + def __handle_image(self, image): with self.lock_for_update() as txn: + self.__current_value["image"] = image if self.__update_plot(txn) and self.__single_button: # self.__single_button = False - self.__set_acquire(txn, False) + self.__set(txn, "acquire", False, _BUTTONS_DISABLE) txn.multi_set( ("single_button.ui.enabled", True), ("start_button.ui.enabled", True), ) - def __set_acquire(self, txn, acquire): - if not self.__device: + def __handle_target_status(self, status): + with self.lock_for_update() as txn: + self.__current_value["target"] = status + txn.multi_set( + ("target_status", status.name), + ("target_in_button.ui.enabled", status == TargetStatus.OUT), + ("target_out_button.ui.enabled", status == TargetStatus.IN), + ) + + def __set(self, txn, accessor, value, txn_set, method=None): + if not self.__device or not self.__handler: # buttons already disabled return - v = self.__monitors.acquire.prev_value() - if v is not None and v == acquire: + v = self.__current_value[accessor] + if v is not None and v == value: # No button disable since nothing changed return - if txn: - # No presses until we get a response from device - txn.multi_set(_BUTTONS_DISABLE) + txn.multi_set(txn_set) try: - self.__device.put("acquire", acquire) + if method is None: + self.__device.put(accessor, value) + else: + m = getattr(self.__device, method) + m(value) except slicops.device.DeviceError as e: pkdlog( "error={} on {}, clearing camera; stack={}", e, self.__device, pkdexc() ) 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: + if not self.__device or not self.__handler: return False - if (i := self.__monitors.image.prev_value()) is None or not i.size: + if (i := self.__current_value["image"]) is None or not i.size: return False if not txn.group_get("plot", "ui", "visible"): txn.multi_set(_PLOT_ENABLE) @@ -247,80 +296,34 @@ def __user_alert(self, txn, fmt, *args): CLASS = Screen -class _Monitor: - # TODO(robnagler) handle more values besides plot - def __init__(self, accessor, handler): - self.__name = str(accessor) +class _Handler(slicops.device.screen.EventHandler): + def __init__( + self, + handle_device_error, + handle_device_update, + ): self.__destroyed = False - self.__handler = handler - self.__value = None - self.__change_q = queue.Queue(1) self.__lock = threading.Lock() - self.__thread = threading.Thread( - target=self.__dispatch, - daemon=True, - # Reduce the places where locking needs to occur - args=(self.__name, self.__change_q, self.__lock, self.__handler), - ) - self.__thread.start() + self.__handle_device_error = handle_device_error + self.__handle_device_update = handle_device_update def destroy(self): with self.__lock: if self.__destroyed: return self.__destroyed = True - try: - # if there is an exception, ignore it because the queue already has an item for wakeup dispatch - self.__change_q.put_nowait(False) - except Exception: - pass - # cause callers to crash - try: - delattr(self, "value") - except Exception: - pass - - def prev_value(self): - with self.__lock: - if self.__destroyed: - return - return self.__value - - def __call__(self, change): - with self.__lock: - if self.__destroyed: - return - if e := change.get("error"): - pkdlog("error={} on {}", e, change.get("accessor")) - return - if (v := change.get("value")) is None: - return - try: - self.__change_q.put_nowait(v) - except queue.Full: - if self.__change_q.get_nowait() is not None: - self.__change_q.task_done() - # puts are locked - self.__change_q.put_nowait(v) - - def __dispatch(self, name, change_q, lock, handler): - try: - while True: - v = change_q.get() - change_q.task_done() - with lock: - if self.__destroyed: - return - self.__value = v - try: - handler(v) - except Exception as e: - # Touches self.__name which should not be modified - pkdlog("handler error={} accessor={} stack={}", e, name, pkdexc()) - except Exception as e: - pkdlog("error={} accessor={} stack={}", e, name, pkdexc()) - finally: - self.destroy() + self.__handle_device_error = None + self.__handle_device_update = None + + def on_screen_device_error(self, exc): + self.__handle_device_error(exc) + + def on_screen_device_update(self, accessor_name, value): + # TODO move prev value to sliclet within txn + if not accessor_name in self.__handle_device_update: + raise AssertionError(f"unsupported accessor={n} {self}") + h = self.__handle_device_update[accessor_name] + h(value) def _init(): diff --git a/slicops/unit_util.py b/slicops/unit_util.py index 208c649..1671bf7 100644 --- a/slicops/unit_util.py +++ b/slicops/unit_util.py @@ -16,9 +16,6 @@ class SlicletSetup(pykern.api.unit_util.Setup): def __init__(self, sliclet, *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() @@ -58,13 +55,6 @@ def _http_config(self, *args, **kwargs): return config.cfg().ui_api.copy() def _server_config(self, *args, **kwargs): - 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): @@ -81,9 +71,6 @@ 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 @@ -176,7 +163,7 @@ def _path(path, arg): finally: os._exit(0) try: - time.sleep(1) + time.sleep(2) yield None finally: os.kill(p, signal.SIGKILL) @@ -198,8 +185,8 @@ def __init__(self, *args, **kwargs): } ) - def on_screen_device_error(self, **kwargs): - self.event_q.error.put_nowait(PKDict(kwargs)) + def on_screen_device_error(self, exc): + self.event_q.error.put_nowait(PKDict(exception=exc)) def on_screen_device_update(self, **kwargs): self.event_q[kwargs["accessor_name"]].put_nowait(PKDict(kwargs)) diff --git a/tests/device/screen1_test.py b/tests/device/screen1_test.py index 0beee91..1cbaf58 100644 --- a/tests/device/screen1_test.py +++ b/tests/device/screen1_test.py @@ -9,12 +9,13 @@ def test_upstream_ok(): from pykern import pkdebug, pkunit + from slicops.device.screen import TargetStatus 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")) + pkunit.pkeq(TargetStatus.OUT, s.handler.test_get("target_status")) s.device.move_target(want_in=True) - pkunit.pkeq(True, s.handler.test_get("target_status")) + pkunit.pkeq(TargetStatus.IN, s.handler.test_get("target_status")) s.device.move_target(want_in=False) - pkunit.pkeq(False, s.handler.test_get("target_status")) + pkunit.pkeq(TargetStatus.OUT, s.handler.test_get("target_status")) diff --git a/tests/device/screen2_test.py b/tests/device/screen2_test.py index f2912fa..b309208 100644 --- a/tests/device/screen2_test.py +++ b/tests/device/screen2_test.py @@ -8,9 +8,14 @@ def test_upstream_blocked(): from pykern import pkdebug, pkunit from slicops import unit_util + from slicops.device.screen import ScreenError, ErrorKind 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.name) - pkunit.pkeq("upstream target is in", e.error_msg.YAG02) + s = ScreenError( + device="YAG03", + error_kind=ErrorKind.upstream, + error_msg="{'YAG02': 'upstream target is IN'}", + ) + pkunit.pkeq(repr(s), repr(e.exception)) # pkunit magic? diff --git a/tests/sliclet/screen_data/ioc.py b/tests/sliclet/screen_data/ioc.py new file mode 100644 index 0000000..2900eb1 --- /dev/null +++ b/tests/sliclet/screen_data/ioc.py @@ -0,0 +1,37 @@ +import numpy + +_X = 50 +_Y_FACTOR = 1.3 + + +def empty_image(self): + return [0] * 50 * 65 + + +def gaussian_image(self): + sigma = _X // 5 + + def _dist(vec, is_y): + s = _y_adjust(sigma) if is_y else sigma + return (vec - vec.shape[0] // 2) ** 2 / (2 * (s**2)) + + def _norm(mat): + return ((mat - mat.min()) / (mat.max() - mat.min())) * 255 + + def _vec(size): + return numpy.linspace(0, size - 1, size) + + x, y = numpy.meshgrid(_vec(_X), _vec(_y_adjust(_X))) + return _norm(numpy.exp(-(_dist(x, False) + _dist(y, True)))).flatten() + + +def size_x(self): + return _X + + +def size_y(self): + return _y_adjust(_X) + + +def _y_adjust(value): + return int(value * _Y_FACTOR) diff --git a/tests/sliclet/screen_data/ioc.yaml b/tests/sliclet/screen_data/ioc.yaml new file mode 100644 index 0000000..4d47095 --- /dev/null +++ b/tests/sliclet/screen_data/ioc.yaml @@ -0,0 +1,10 @@ +--- +13SIM1:cam1:Acquire: + value: 0 + dispatch: + 13SIM1:image1:ArrayData: + 0: empty_image() + 1: gaussian_image() +13SIM1:image1:ArrayData: empty_image() +13SIM1:cam1:SizeY: size_y() +13SIM1:cam1:SizeX: size_x() diff --git a/tests/sliclet/screen_test.py b/tests/sliclet/screen_test.py index f4cb07f..c8ab803 100644 --- a/tests/sliclet/screen_test.py +++ b/tests/sliclet/screen_test.py @@ -15,6 +15,7 @@ async def test_basic(): async def _buttons(s, expect, msg): from pykern import pkunit, pkdebug + from pykern.pkdebug import pkdlog from asyncio.exceptions import CancelledError # Wait for buttons to "settle" on expect. The updates @@ -30,40 +31,38 @@ async def _buttons(s, expect, msg): break return rv - async with unit_util.SlicletSetup("screen") as s: - from pykern import pkunit, pkdebug - from pykern.pkdebug import pkdc, pkdexc, pkdlog, pkdp - import asyncio + with unit_util.start_ioc("ioc"): + async with unit_util.SlicletSetup("screen") as s: + from pykern import pkunit, pkdebug + from pykern.pkdebug import pkdc, pkdexc, pkdlog, pkdp + import asyncio + import epics - 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") - # plot comes back - r = await s.ctx_update() - # TODO(robnagler) need to test acquire is set around when the plot comes back - # mock epics should handle this - p = r.fields.plot.value - pkunit.pkeq(65, len(p.raw_pixels)) - pkunit.pkeq(50, len(p.raw_pixels[0])) - # x fit should be 10 - pkunit.pkeq(10.00, round(p.x.fit.results.sig, 2)) - pkunit.pkeq(13.00, round(p.y.fit.results.sig, 2)) - await s.ctx_field_set( - beam_path="CU_SPEC", - curve_fit_method="super_gaussian", - stop_button=None, - ) - r = await _buttons(s, (False, False, False), "all disabled after stop") - pkunit.pkeq(None, r.fields.camera.value) - # there's no device so buttons on not visible - pkunit.pkeq(False, r.fields.start_button.ui.visible) - with pkunit.pkexcept("unknown choice"): - await s.ctx_field_set(camera="DEV_CAMERA") - # TODO(robnagler) better error handling await _put(ux, "camera", "DEV_CAMERA", Exception) - await s.ctx_field_set(camera="YAG01") - r = await s.ctx_update() - pkunit.pkeq("YAGS:IN20:211", r.fields.pv.value) + 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") + p = (await s.ctx_update()).fields.plot.value + pkunit.pkeq(65, len(p.raw_pixels)) + pkunit.pkeq(50, len(p.raw_pixels[0])) + # x fit should be 10 + pkunit.pkeq(10.00, round(p.x.fit.results.sig, 2)) + pkunit.pkeq(13.00, round(p.y.fit.results.sig, 2)) + await s.ctx_field_set( + beam_path="CU_SPEC", + curve_fit_method="super_gaussian", + stop_button=None, + ) + r = await _buttons(s, (False, False, False), "all disabled after stop") + pkunit.pkeq(None, r.fields.camera.value) + # there's no device so buttons on not visible + pkunit.pkeq(False, r.fields.start_button.ui.visible) + with pkunit.pkexcept("unknown choice"): + await s.ctx_field_set(camera="DEV_CAMERA") + # TODO(robnagler) better error handling await _put(ux, "camera", "DEV_CAMERA", Exception) + await s.ctx_field_set(camera="YAG01") + r = await s.ctx_update() + pkunit.pkeq("YAGS:IN20:211", r.fields.pv.value)