From 9bc4f8d01e040439ddbb99da6daf848b9bfb451e Mon Sep 17 00:00:00 2001 From: Henrik Lemke Date: Thu, 8 Oct 2026 08:38:57 +0200 Subject: [PATCH] tweak: stacked keys, Qt/ipywidgets panels, detector recording - Tweak.tweak(): 1 adjustable arrows, 2 x/y arrows, 3-4 stacked keyboard rows (j k l ; / u i o p / 7 8 9 0, plus m , . / for four), each row neg dir, step*2, step/2, pos dir; stacked=True forces it for 1-2. - Tweak panels (eco.widgets.tweak_panel, _qt, _ipy) with everything the terminal tweak does for 1-4 axes, plus a 'keypress control' checkbox (on by default) that takes the same keys and shows the key manual. Motor records drive their own TWV/TWF/TWR fields. Opened by ._widget_tweak() / Tweak.widget(), and by .tweak() wherever there is no terminal: notebooks get ipywidgets, eco desktop consoles get Qt (subprocess kernels are marked with ECO_QTCONSOLE_KERNEL). Replaces the three old notebook tweak UIs and their page-wide JS key handler. - detectors= on Tweak, tweak() and _widget_tweak(): values recorded per tweak step, averaged while the positions are constant, and plotted live: against the position for one adjustable, as a stack over the step number for several. Embedded in the panels, separate Qt window for the terminal tweak; data via .to_dataframe() on the recorder. - Tweak.wait() now waits for all changers, not only the last one. --- eco/devices_general/motors.py | 268 ++++----------- eco/elements/adjustable.py | 611 ++++++++++----------------------- eco/widgets/console_kernel.py | 3 + eco/widgets/tweak_panel.py | 445 ++++++++++++++++++++++++ eco/widgets/tweak_panel_ipy.py | 317 +++++++++++++++++ eco/widgets/tweak_panel_qt.py | 427 +++++++++++++++++++++++ eco/widgets/tweak_recorder.py | 286 +++++++++++++++ tests/test_tweak_panel.py | 329 ++++++++++++++++++ tests/test_tweak_recorder.py | 240 +++++++++++++ 9 files changed, 2298 insertions(+), 628 deletions(-) create mode 100644 eco/widgets/tweak_panel.py create mode 100644 eco/widgets/tweak_panel_ipy.py create mode 100644 eco/widgets/tweak_panel_qt.py create mode 100644 eco/widgets/tweak_recorder.py create mode 100644 tests/test_tweak_panel.py create mode 100644 tests/test_tweak_recorder.py diff --git a/eco/devices_general/motors.py b/eco/devices_general/motors.py index f1dee31..3788590 100755 --- a/eco/devices_general/motors.py +++ b/eco/devices_general/motors.py @@ -145,185 +145,59 @@ def _is_notebook(): return False -def _tweak_ioc_notebook(self, step_value=None): - try: - from IPython.display import display, Javascript - import ipywidgets as widgets - except Exception as exc: - raise RuntimeError( - "Notebook tweak mode requires IPython and ipywidgets." - ) from exc +def _widget_tweak_ioc( + self, step_value=None, backend=None, display=False, detectors=None +): + """Tweak panel (Qt or ipywidgets) driving the motor record's own tweak + fields (TWV/TWF/TWR), with the same keys as the terminal tweak under + "keypress control" -- see eco.widgets.tweak_panel. `detectors` are + recorded and plotted along the tweak (eco.widgets.tweak_recorder).""" + from eco.widgets.tweak_panel import IocTweakAxis, tweak_panel - if hasattr(self, "_motor") and hasattr(self._motor, "get_pv"): - pv = self._motor.get_pv("TWV") - pvf = self._motor.get_pv("TWF") - pvr = self._motor.get_pv("TWR") - else: - pv = PV(self.pvname + ":TWV") - pvf = PV(self.pvname + ":TWF.PROC") - pvr = PV(self.pvname + ":TWR.PROC") - if step_value is None: - step_value = pv.get() - try: - step_value = float(step_value) - except Exception: - step_value = 1.0 - - def _format_value(value): - try: - return f"{value:1.6g}" - except Exception: - return str(value) - - target_value = self.get_current_value() - current_label = widgets.HTML( - value=f"Current position: {_format_value(target_value)}" + return tweak_panel( + [IocTweakAxis(self, step_value, go_to_current_value_first=True)], + backend=backend, + display=display, + detectors=detectors, ) - status_label = widgets.HTML(value=f"Step size: {_format_value(step_value)}") - help_label = widgets.HTML( - value=( - "Controls: Stepsize *2, Stepsize /2, Down, Up, Go abs, Reset offset, Exit" + + +def _tweak_ioc_entry(self, *args, detectors=None, **kwargs): + """Interactive tweak through the motor record's tweak fields. + + detectors: optional Detector or list of Detectors, recorded for every + tweak step (averaged while the position stays constant) and plotted + against the position; the data stays in ._tweak_recorder.""" + if detectors is None: + return self._tweak_ioc(*args, **kwargs) + from eco.widgets.tweak_panel import IocTweakAxis, frontend + from eco.widgets.tweak_recorder import terminal_recording + + step_value = args[0] if args else kwargs.get("step_value") + if frontend() != "terminal": + self._tweak_panel = _widget_tweak_ioc( + self, step_value, display=True, detectors=detectors ) - ) - step_input = widgets.FloatText( - value=step_value, - description="Step:", - layout=widgets.Layout(width="220px"), - ) - go_input = widgets.FloatText( - value=target_value, - description="Go:", - layout=widgets.Layout(width="220px"), - ) - set_input = widgets.FloatText( - value=target_value, - description="Set:", - layout=widgets.Layout(width="220px"), - ) + self._tweak_recorder = self._tweak_panel.recorder + return + with terminal_recording( + [IocTweakAxis(self, step_value)], + detectors, + owner=self, + label=f"{self.name}._tweak_recorder", + ): + return self._tweak_ioc(*args, **kwargs) - def _refresh_current(value=None): - if value is None: - value = self.get_current_value() - current_label.value = f"Current position: {_format_value(value)}" - status_label.value = f"Step size: {_format_value(step_input.value)}" - def _update_callback(**kwargs): - if "value" in kwargs: - _refresh_current(kwargs["value"]) - else: - _refresh_current() +def _tweak_panel_instead(self, step_value): + """Where there is no terminal to read keys from (notebook, eco desktop + console), show the tweak panel instead. True if it did.""" + from eco.widgets.tweak_panel import frontend - callback_id = self.add_value_callback(_update_callback) - - def _set_step(factor): - step_input.value = float(step_input.value) * factor - pv.put(step_input.value) - status_label.value = f"Step size: {_format_value(step_input.value)}" - - def _click_up(_): - _set_step(2.0) - - def _click_down(_): - _set_step(0.5) - - def _click_left(_): - pvr.put(1) - status_label.value = "Command: left" - - def _click_right(_): - pvf.put(1) - status_label.value = "Command: right" - - def _click_go(_): - try: - self.set_target_value(float(go_input.value), check=True).wait() - _refresh_current() - status_label.value = f"Moved to: {_format_value(go_input.value)}" - except Exception as exc: - status_label.value = f"Error: {exc}" - - def _click_set(_): - try: - self.reset_current_value_to(float(set_input.value)) - _refresh_current() - status_label.value = ( - f"Offset reset to: {_format_value(set_input.value)}" - ) - except Exception as exc: - status_label.value = f"Error: {exc}" - - def _shutdown(_=None): - self.clear_value_callback(index=callback_id) - for ctl in [ - step_input, - go_input, - set_input, - btn_up, - btn_down, - btn_left, - btn_right, - btn_go, - btn_set, - btn_exit, - ]: - ctl.disabled = True - status_label.value = "Tweak UI closed." - - btn_up = widgets.Button(description="Stepsize *2", button_style="success") - btn_down = widgets.Button(description="Stepsize /2", button_style="warning") - btn_left = widgets.Button(description="Down", button_style="info") - btn_right = widgets.Button(description="Up", button_style="info") - btn_go = widgets.Button(description="Go abs", button_style="primary") - btn_set = widgets.Button(description="Reset offset", button_style="primary") - btn_exit = widgets.Button(description="Exit", button_style="danger") - - btn_up.on_click(_click_up) - btn_down.on_click(_click_down) - btn_left.on_click(_click_left) - btn_right.on_click(_click_right) - btn_go.on_click(_click_go) - btn_set.on_click(_click_set) - btn_exit.on_click(_shutdown) - - controls = widgets.HBox( - [btn_up, btn_down, btn_left, btn_right, btn_go, btn_set, btn_exit] - ) - inputs = widgets.HBox([step_input, go_input, set_input]) - ui = widgets.VBox([current_label, status_label, help_label, inputs, controls]) - - js = f""" -(function() {{ - const mapping = {{ - u: '{btn_up._model_id}', - d: '{btn_down._model_id}', - l: '{btn_left._model_id}', - r: '{btn_right._model_id}', - g: '{btn_go._model_id}', - s: '{btn_set._model_id}', - q: '{btn_exit._model_id}' - }}; - window.__eco_tweak_keys = window.__eco_tweak_keys || {{}}; - window.__eco_tweak_keys = Object.assign(window.__eco_tweak_keys, mapping); - if (!window.__eco_tweak_key_handler) {{ - window.__eco_tweak_key_handler = function(event) {{ - const key = event.key.toLowerCase(); - const targetId = window.__eco_tweak_keys[key]; - if (!targetId) return; - const root = document.querySelector('[data-widget-id="' + targetId + '"]'); - if (!root) return; - const btn = root.querySelector('button'); - if (btn && !btn.disabled) {{ - btn.click(); - event.preventDefault(); - }} - }}; - document.addEventListener('keydown', window.__eco_tweak_key_handler); - }} -}})(); -""" - display(ui) - display(Javascript(js)) - return ui + if frontend() == "terminal": + return False + self._tweak_panel = _widget_tweak_ioc(self, step_value, display=True) + return True @spec_convenience @@ -614,14 +488,8 @@ class SmaractStreamdevice(Assembly): self._currentChange = self.set_target_value(value) def _tweak_ioc(self, step_value=None): - if _is_notebook(): - try: - return _tweak_ioc_notebook(self, step_value=step_value) - except Exception as exc: - print( - "Notebook tweak UI failed; falling back to terminal mode:", - exc, - ) + if _tweak_panel_instead(self, step_value): + return pv = PV(self.pvname + ":TWV") pvf = PV(self.pvname + ":TWF.PROC") @@ -700,8 +568,8 @@ class SmaractStreamdevice(Assembly): print(f"final position: {self.get_current_value()}") print(f"final tweak step: {pv.get()}") - def tweak(self, *args, **kwargs): - return self._tweak_ioc(*args, **kwargs) + tweak = _tweak_ioc_entry + _widget_tweak = _widget_tweak_ioc def gui(self): num = "" @@ -1874,14 +1742,8 @@ class MotorRecord(Assembly): def _tweak_ioc(self, step_value=None, go_to_current_value_first=True): if go_to_current_value_first: self.set_target_value(self.get_current_value()).wait() - if _is_notebook(): - try: - return _tweak_ioc_notebook(self, step_value=step_value) - except Exception as exc: - print( - "Notebook tweak UI failed; falling back to terminal mode:", - exc, - ) + if _tweak_panel_instead(self, step_value): + return pv = self._motor.get_pv("TWV") pvf = self._motor.get_pv("TWF") @@ -1965,8 +1827,8 @@ class MotorRecord(Assembly): print(f"final tweak step: {pv.get()}") # print('\033[K',"the info",sep='',flush=True) - def tweak(self, *args, **kwargs): - return self._tweak_ioc(*args, **kwargs) + tweak = _tweak_ioc_entry + _widget_tweak = _widget_tweak_ioc MotorRecord = MotorRecord @@ -2506,14 +2368,8 @@ class SmaractRecord(Assembly): def _tweak_ioc(self, step_value=None, go_to_current_value_first=True): if go_to_current_value_first: self.set_target_value(self.get_current_value()).wait() - if _is_notebook(): - try: - return _tweak_ioc_notebook(self, step_value=step_value) - except Exception as exc: - print( - "Notebook tweak UI failed; falling back to terminal mode:", - exc, - ) + if _tweak_panel_instead(self, step_value): + return pv = self._motor.get_pv("TWV") pvf = self._motor.get_pv("TWF") @@ -2589,8 +2445,8 @@ class SmaractRecord(Assembly): print(f"final position: {self.get_current_value()}") print(f"final tweak step: {pv.get()}") - def tweak(self, *args, **kwargs): - return self._tweak_ioc(*args, **kwargs) + tweak = _tweak_ioc_entry + _widget_tweak = _widget_tweak_ioc @spec_convenience @@ -2909,6 +2765,8 @@ class SmaractRecord_old(Assembly): self._currentChange = self.set_target_value(value) def _tweak_ioc(self, step_value=None): + if _tweak_panel_instead(self, step_value): + return pv = self._motor.get_pv("TWV") pvf = self._motor.get_pv("TWF") pvr = self._motor.get_pv("TWR") @@ -2983,8 +2841,8 @@ class SmaractRecord_old(Assembly): print(f"final position: {self.get_current_value()}") print(f"final tweak step: {pv.get()}") - def tweak(self, *args, **kwargs): - return self._tweak_ioc(*args, **kwargs) + tweak = _tweak_ioc_entry + _widget_tweak = _widget_tweak_ioc flag_names_smaract_record = { diff --git a/eco/elements/adjustable.py b/eco/elements/adjustable.py index 6e816f4..4bb8d92 100644 --- a/eco/elements/adjustable.py +++ b/eco/elements/adjustable.py @@ -53,11 +53,21 @@ class AdjustableError(Exception): def tweak_option(Obj): - def tweak(self, interval, *args, **kwargs): - self._tweak_instance = Tweak((self, interval)) - self._tweak_instance.single_adjustable_tweak() + def tweak(self, interval, *args, detectors=None, **kwargs): + """Interactive tweak; `detectors` (a Detector or a list) are recorded + and plotted along the tweak, see Tweak.""" + self._tweak_instance = Tweak((self, interval), detectors=detectors) + self._tweak_instance._recorder_label = f"{self.name}._tweak_instance.recorder" + self._tweak_instance.tweak() + + def _widget_tweak(self, interval=1.0, backend=None, display=False, detectors=None): + """Tweak panel (Qt or ipywidgets) with keypress control, see + eco.widgets.tweak_panel.""" + self._tweak_instance = Tweak((self, interval), detectors=detectors) + return self._tweak_instance.widget(backend=backend, display=display) Obj.tweak = tweak + Obj._widget_tweak = _widget_tweak return Obj @@ -1190,8 +1200,16 @@ class AdjustableEnum: return self._base.set_target_value(value, hold=hold) class Tweak: - def __init__(self, *args): - """usage: Tweak((adj0,startstepsize0),(adj1,startstepsize1))""" + def __init__(self, *args, detectors=None): + """usage: Tweak((adj0,startstepsize0),(adj1,startstepsize1)) + + detectors: optional Detector or list of Detectors. Their values are + recorded for every tweak step (averaged while the positions stay + constant) and live-plotted -- against the position for one + adjustable, as a stack over the step number for several (see + eco.widgets.tweak_recorder). The data stays in .recorder.""" + self.detectors = detectors + self.recorder = None self.adjs = [] startsteps = [] for adj, startstep in args: @@ -1246,14 +1264,8 @@ class Tweak: if len(self.adjs) != 2: raise AdjustableError("xy_adjustable_tweak requires exactly two adjustables") - if _is_notebook(): - try: - return self._xy_adjustable_tweak_notebook() - except Exception as exc: - print( - "Notebook tweak UI failed; falling back to terminal mode:", - exc, - ) + if self._open_panel_instead(): + return x_adj, y_adj = self.adjs i_x, i_y = 0, 1 @@ -1333,12 +1345,169 @@ class Tweak: k.waitkey() + # Keyboard rows for stacked_adjustable_tweak, bottom to top. Within a row + # the keys mirror the arrow layout of the single tweak: + # (left = neg dir, up = step*2, down = step/2, right = pos dir). + STACKED_KEY_ROWS = ( + ("m", ",", ".", "/"), + ("j", "k", "l", ";"), + ("u", "i", "o", "p"), + ("7", "8", "9", "0"), + ) + + def tweak(self, stacked=None): + """Interactive keyboard tweak of all adjustables of this Tweak. + + 1 adjustable: arrow keys, 2: arrow keys in x/y, 3-4: stacked keyboard + rows (see stacked_adjustable_tweak). stacked=True forces the stacked + rows for 1 or 2 adjustables too. Outside a terminal (notebook, eco + desktop console) this opens the tweak panel instead.""" + n = len(self.adjs) + if stacked is None: + stacked = n > 2 + if self._open_panel_instead(stacked=stacked): + return + if not stacked and n > 2: + raise AdjustableError(f"cannot tweak {n} adjustables without stacked keys") + from eco.widgets.tweak_panel import axes_from_tweak + from eco.widgets.tweak_recorder import terminal_recording + + with terminal_recording( + axes_from_tweak(self) if self.detectors else [], + self.detectors, + label=getattr(self, "_recorder_label", ".recorder"), + ) as recorder: + if recorder is not None: + self.recorder = recorder + if stacked: + return self.stacked_adjustable_tweak() + if n == 1: + return self.single_adjustable_tweak() + return self.xy_adjustable_tweak() + + def widget(self, stacked=None, backend=None, display=False, **kwargs): + """Tweak panel for these adjustables (Qt or ipywidgets), with the same + keys as the terminal tweak under "keypress control" -- see + eco.widgets.tweak_panel.""" + from eco.widgets.tweak_panel import axes_from_tweak, tweak_panel + + panel = tweak_panel( + axes_from_tweak(self), + stacked=stacked, + backend=backend, + display=display, + detectors=self.detectors, + **kwargs, + ) + if getattr(panel, "recorder", None) is not None: + self.recorder = panel.recorder + return panel + + def _open_panel_instead(self, stacked=None): + """Where there is no terminal to read keys from (notebook, eco desktop + console), show the tweak panel instead. True if it did.""" + from eco.widgets.tweak_panel import frontend + + if frontend() == "terminal": + return False + self._panel = self.widget(stacked=stacked, display=True) + return True + + def _stacked_key_rows(self): + n = len(self.adjs) + if not 1 <= n <= len(self.STACKED_KEY_ROWS): + raise AdjustableError( + f"stacked tweak supports 1 to {len(self.STACKED_KEY_ROWS)} adjustables, got {n}" + ) + rows = self.STACKED_KEY_ROWS + return list(rows[:4] if n == 4 else rows[1 : 1 + n]) + + def stacked_adjustable_tweak(self): + """Tweak up to 4 adjustables, one keyboard row per adjustable. + + Per row: 1st key = neg dir, 2nd = step*2, 3rd = step/2, 4th = pos dir. + Rows used: 1 -> j k l ; 2 -> + u i o p 3 -> + 7 8 9 0 + 4 -> m , . / j k l ; u i o p 7 8 9 0 (first adjustable lowest).""" + rows = self._stacked_key_rows() + keymap = {} + for i_adj, (neg, dbl, half, pos) in enumerate(rows): + keymap[neg] = (i_adj, "move", -1) + keymap[dbl] = (i_adj, "step", 2.0) + keymap[half] = (i_adj, "step", 0.5) + keymap[pos] = (i_adj, "move", +1) + + names = [str(getattr(adj, "name", adj)) for adj in self.adjs] + width = max(len(name) for name in names) + print("tweaking " + ", ".join(names)) + print("q = exit; s = all axes back to start values") + for i_adj in reversed(range(len(rows))): + neg, dbl, half, pos = rows[i_adj] + print( + f" {names[i_adj]:<{width}} : {neg} = neg dir, {dbl} = step*2, " + f"{half} = step/2, {pos} = pos dir" + ) + print( + "Starting at " + + ", ".join( + f"{name}={value}" for name, value in zip(names, self.target_positions[0]) + ) + ) + k = KeyPress() + cll = colorama.ansi.clear_line() + + def _fmt(value): + try: + return f"{value:1.5g}" + except Exception: + return str(value) + + class Printer: + def __init__(self, tweak=self): + self.tweak = tweak + self.thread = None + + def print(self): + if self.thread and self.thread.is_alive(): + return + else: + self.thread = Thread(target=self.print_foo) + self.thread.daemon = True + self.thread.start() + + def _line(self, values): + return "; ".join( + f"{name}: {_fmt(value)} (step {_fmt(step)})" + for name, value, step in zip(names, values, self.tweak.step_sizes) + ) + + def print_foo(self, **kwargs): + if self.tweak._changers: + print(cll + "changing ...", end="\r") + self.tweak.wait() + print(cll + self._line(self.tweak.get_current_values()), end="\r") + + p = Printer() + print(" ") + p.print() + while k.isq() is False: + action = keymap.get(k.last_key) + if action is not None: + i_adj, kind, factor = action + if kind == "step": + self.set_step_size((i_adj, self.step_sizes[i_adj] * factor)) + else: + self.set_target_step_increment((i_adj, factor)) + p.print() + elif k.iskey("s"): + self.change_to_targets(self.startpositions) + p.print() + + k.waitkey() + print("") + def wait(self, sleeptime=0.02): if self._changers: - changing = True - while changing: - for changer in self._changers: - changing = changer.is_alive() + while any(changer.is_alive() for changer in self._changers): time.sleep(sleeptime) def set_step_size(self, *args): @@ -1356,413 +1525,9 @@ class Tweak: new_steps[index] = stepsize self.step_sizes = new_steps - def _single_adjustable_tweak_notebook(self): - try: - from IPython.display import display, Javascript - import ipywidgets as widgets - except Exception as exc: - raise RuntimeError( - "Notebook tweak mode requires IPython and ipywidgets." - ) from exc - - i_adj = 0 - adj = self.adjs[i_adj] - - def _format_value(value): - try: - return f"{value:1.6g}" - except Exception: - return str(value) - - current_value = adj.get_current_value() - step_value = self.step_sizes[i_adj] - current_label = widgets.HTML( - value=f"Current position: {_format_value(current_value)}" - ) - status_label = widgets.HTML( - value=f"Step size: {_format_value(step_value)}" - ) - help_label = widgets.HTML( - value=( - "Controls: Stepsize *2, Stepsize /2, Down, Up, Go abs, Reset offset, Exit" - ) - ) - step_input = widgets.FloatText( - value=step_value, - description="Step:", - layout=widgets.Layout(width="220px"), - ) - go_input = widgets.FloatText( - value=current_value, - description="Go:", - layout=widgets.Layout(width="220px"), - ) - set_input = widgets.FloatText( - value=current_value, - description="Set:", - layout=widgets.Layout(width="220px"), - ) - - def _refresh_current(value=None): - if value is None: - value = adj.get_current_value() - current_label.value = f"Current position: {_format_value(value)}" - status_label.value = f"Step size: {_format_value(step_input.value)}" - go_input.value = value - set_input.value = value - - def _update_callback(**kwargs): - if "value" in kwargs: - _refresh_current(kwargs["value"]) - else: - _refresh_current() - - callback_id = None - if hasattr(adj, "add_value_callback"): - try: - callback_id = adj.add_value_callback(_update_callback) - except Exception: - callback_id = None - - def _set_step(factor): - new_step = float(step_input.value) * factor - step_input.value = new_step - self.set_step_size((adj, new_step)) - status_label.value = f"Step size: {_format_value(new_step)}" - - def _click_up(_): - _set_step(2.0) - - def _click_down(_): - _set_step(0.5) - - def _click_left(_): - self.set_target_step_increment((adj, -1)) - _refresh_current() - - def _click_right(_): - self.set_target_step_increment((adj, +1)) - _refresh_current() - - def _click_go(_): - try: - value = float(go_input.value) - changer = adj.set_target_value(value) - if hasattr(changer, "wait"): - changer.wait() - _refresh_current(value) - except Exception as exc: - status_label.value = f"Error: {exc}" - - def _click_set(_): - try: - value = float(set_input.value) - if hasattr(adj, "reset_current_value_to"): - adj.reset_current_value_to(value) - else: - changer = adj.set_target_value(value) - if hasattr(changer, "wait"): - changer.wait() - _refresh_current(value) - except Exception as exc: - status_label.value = f"Error: {exc}" - - def _shutdown(_=None): - if callback_id is not None and hasattr(adj, "clear_value_callback"): - try: - adj.clear_value_callback(index=callback_id) - except TypeError: - adj.clear_value_callback() - except Exception: - pass - for ctl in [ - step_input, - go_input, - set_input, - btn_up, - btn_down, - btn_left, - btn_right, - btn_go, - btn_set, - btn_exit, - ]: - ctl.disabled = True - status_label.value = "Tweak UI closed." - - btn_up = widgets.Button(description="Stepsize *2", button_style="success") - btn_down = widgets.Button(description="Stepsize /2", button_style="warning") - btn_left = widgets.Button(description="Down", button_style="info") - btn_right = widgets.Button(description="Up", button_style="info") - btn_go = widgets.Button(description="Go abs", button_style="primary") - btn_set = widgets.Button(description="Reset offset", button_style="primary") - btn_exit = widgets.Button(description="Exit", button_style="danger") - - btn_up.on_click(_click_up) - btn_down.on_click(_click_down) - btn_left.on_click(_click_left) - btn_right.on_click(_click_right) - btn_go.on_click(_click_go) - btn_set.on_click(_click_set) - btn_exit.on_click(_shutdown) - - controls = widgets.HBox( - [btn_up, btn_down, btn_left, btn_right, btn_go, btn_set, btn_exit] - ) - inputs = widgets.HBox([step_input, go_input, set_input]) - ui = widgets.VBox([current_label, status_label, help_label, inputs, controls]) - - js = f""" -(function() {{ - const mapping = {{ - u: '{btn_up._model_id}', - d: '{btn_down._model_id}', - l: '{btn_left._model_id}', - r: '{btn_right._model_id}', - g: '{btn_go._model_id}', - s: '{btn_set._model_id}', - q: '{btn_exit._model_id}' - }}; - window.__eco_tweak_keys = window.__eco_tweak_keys || {{}}; - window.__eco_tweak_keys = Object.assign(window.__eco_tweak_keys, mapping); - if (!window.__eco_tweak_key_handler) {{ - window.__eco_tweak_key_handler = function(event) {{ - const key = event.key.toLowerCase(); - const targetId = window.__eco_tweak_keys[key]; - if (!targetId) return; - const root = document.querySelector('[data-widget-id="' + targetId + '"]'); - if (!root) return; - const btn = root.querySelector('button'); - if (btn && !btn.disabled) {{ - btn.click(); - event.preventDefault(); - }} - }}; - document.addEventListener('keydown', window.__eco_tweak_key_handler); - }} -}})(); -""" - display(ui) - display(Javascript(js)) - return ui - - def _xy_adjustable_tweak_notebook(self): - try: - from IPython.display import display, Javascript - import ipywidgets as widgets - except Exception as exc: - raise RuntimeError( - "Notebook tweak mode requires IPython and ipywidgets." - ) from exc - - x_adj, y_adj = self.adjs - i_x, i_y = 0, 1 - - def _format_value(value): - try: - return f"{value:1.6g}" - except Exception: - return str(value) - - def _refresh_current(value=None): - if value is None: - current = self.get_current_values() - else: - current = value - x_current = current[i_x] - y_current = current[i_y] - x_label.value = f"x ({x_adj.name}): {_format_value(x_current)}" - y_label.value = f"y ({y_adj.name}): {_format_value(y_current)}" - status_label.value = ( - f"Step sizes: x={_format_value(x_step_input.value)}, y={_format_value(y_step_input.value)}" - ) - - def _update_callback(**kwargs): - _refresh_current() - - callback_id_x = None - callback_id_y = None - if hasattr(x_adj, "add_value_callback"): - try: - callback_id_x = x_adj.add_value_callback(_update_callback) - except Exception: - callback_id_x = None - if hasattr(y_adj, "add_value_callback"): - try: - callback_id_y = y_adj.add_value_callback(_update_callback) - except Exception: - callback_id_y = None - - x_step_input = widgets.FloatText( - value=self.step_sizes[i_x], - description="X step:", - layout=widgets.Layout(width="220px"), - ) - y_step_input = widgets.FloatText( - value=self.step_sizes[i_y], - description="Y step:", - layout=widgets.Layout(width="220px"), - ) - - x_label = widgets.HTML( - value=f"x ({x_adj.name}): {_format_value(x_adj.get_current_value())}" - ) - y_label = widgets.HTML( - value=f"y ({y_adj.name}): {_format_value(y_adj.get_current_value())}" - ) - status_label = widgets.HTML( - value=( - f"Step sizes: x={_format_value(x_step_input.value)}, y={_format_value(y_step_input.value)}" - ) - ) - help_label = widgets.HTML( - value=( - "Controls: x-/x+, y-/y+, X step *2, X step /2, Y step *2, Y step /2, Reset origin, Exit" - ) - ) - - def _set_step(adj, index, factor, ctl): - new_step = float(ctl.value) * factor - ctl.value = new_step - self.set_step_size((adj, new_step)) - status_label.value = ( - f"Step sizes: x={_format_value(x_step_input.value)}, y={_format_value(y_step_input.value)}" - ) - - def _click_x_minus(_): - self.set_target_step_increment((x_adj, -1)) - _refresh_current() - - def _click_x_plus(_): - self.set_target_step_increment((x_adj, +1)) - _refresh_current() - - def _click_y_minus(_): - self.set_target_step_increment((y_adj, -1)) - _refresh_current() - - def _click_y_plus(_): - self.set_target_step_increment((y_adj, +1)) - _refresh_current() - - def _click_x_step_up(_): - _set_step(x_adj, i_x, 2.0, x_step_input) - - def _click_x_step_down(_): - _set_step(x_adj, i_x, 0.5, x_step_input) - - def _click_y_step_up(_): - _set_step(y_adj, i_y, 2.0, y_step_input) - - def _click_y_step_down(_): - _set_step(y_adj, i_y, 0.5, y_step_input) - - def _click_origin(_): - self.change_to_targets(self.startpositions) - _refresh_current() - - def _shutdown(_=None): - if callback_id_x is not None and hasattr(x_adj, "clear_value_callback"): - try: - x_adj.clear_value_callback(index=callback_id_x) - except TypeError: - x_adj.clear_value_callback() - except Exception: - pass - if callback_id_y is not None and hasattr(y_adj, "clear_value_callback"): - try: - y_adj.clear_value_callback(index=callback_id_y) - except TypeError: - y_adj.clear_value_callback() - except Exception: - pass - for ctl in [ - x_step_input, - y_step_input, - btn_x_minus, - btn_x_plus, - btn_y_minus, - btn_y_plus, - btn_x_step_up, - btn_x_step_down, - btn_y_step_up, - btn_y_step_down, - btn_origin, - btn_exit, - ]: - ctl.disabled = True - status_label.value = "Tweak UI closed." - - btn_x_minus = widgets.Button(description="X -", button_style="info") - btn_x_plus = widgets.Button(description="X +", button_style="info") - btn_y_minus = widgets.Button(description="Y -", button_style="info") - btn_y_plus = widgets.Button(description="Y +", button_style="info") - btn_x_step_up = widgets.Button(description="X step *2", button_style="success") - btn_x_step_down = widgets.Button(description="X step /2", button_style="warning") - btn_y_step_up = widgets.Button(description="Y step *2", button_style="success") - btn_y_step_down = widgets.Button(description="Y step /2", button_style="warning") - btn_origin = widgets.Button(description="Reset origin", button_style="primary") - btn_exit = widgets.Button(description="Exit", button_style="danger") - - btn_x_minus.on_click(_click_x_minus) - btn_x_plus.on_click(_click_x_plus) - btn_y_minus.on_click(_click_y_minus) - btn_y_plus.on_click(_click_y_plus) - btn_x_step_up.on_click(_click_x_step_up) - btn_x_step_down.on_click(_click_x_step_down) - btn_y_step_up.on_click(_click_y_step_up) - btn_y_step_down.on_click(_click_y_step_down) - btn_origin.on_click(_click_origin) - btn_exit.on_click(_shutdown) - - row1 = widgets.HBox([btn_x_minus, btn_x_plus, btn_y_minus, btn_y_plus]) - row2 = widgets.HBox([btn_x_step_up, btn_x_step_down, btn_y_step_up, btn_y_step_down]) - row3 = widgets.HBox([btn_origin, btn_exit]) - inputs = widgets.HBox([x_step_input, y_step_input]) - ui = widgets.VBox([x_label, y_label, status_label, help_label, inputs, row1, row2, row3]) - - js = f""" -(function() {{ - const mapping = {{ - l: '{btn_x_minus._model_id}', - r: '{btn_x_plus._model_id}', - u: '{btn_y_plus._model_id}', - d: '{btn_y_minus._model_id}', - s: '{btn_origin._model_id}', - q: '{btn_exit._model_id}' - }}; - window.__eco_tweak_keys = window.__eco_tweak_keys || {{}}; - window.__eco_tweak_keys = Object.assign(window.__eco_tweak_keys, mapping); - if (!window.__eco_tweak_key_handler) {{ - window.__eco_tweak_key_handler = function(event) {{ - const key = event.key.toLowerCase(); - const targetId = window.__eco_tweak_keys[key]; - if (!targetId) return; - const root = document.querySelector('[data-widget-id="' + targetId + '"]'); - if (!root) return; - const btn = root.querySelector('button'); - if (btn && !btn.disabled) {{ - btn.click(); - event.preventDefault(); - }} - }}; - document.addEventListener('keydown', window.__eco_tweak_key_handler); - }} -}})(); -""" - display(ui) - display(Javascript(js)) - return ui - def single_adjustable_tweak(self): - if _is_notebook(): - try: - return self._single_adjustable_tweak_notebook() - except Exception as exc: - print( - "Notebook tweak UI failed; falling back to terminal mode:", - exc, - ) + if self._open_panel_instead(): + return i_adj = 0 adj = self.adjs[i_adj] # step_value = float(self.step_sizes[0]) diff --git a/eco/widgets/console_kernel.py b/eco/widgets/console_kernel.py index 456f796..0a77020 100644 --- a/eco/widgets/console_kernel.py +++ b/eco/widgets/console_kernel.py @@ -141,6 +141,9 @@ def _subprocess_env(): import sys env = dict(os.environ) + # a qtconsole can't render ipywidgets: lets interactive code (e.g. + # eco.widgets.tweak_panel.frontend) pick a Qt window instead + env["ECO_QTCONSOLE_KERNEL"] = "1" existing = env.get("PYTHONPATH", "") env["PYTHONPATH"] = os.pathsep.join(sys.path) + (os.pathsep + existing if existing else "") return env diff --git a/eco/widgets/tweak_panel.py b/eco/widgets/tweak_panel.py new file mode 100644 index 0000000..fe912cf --- /dev/null +++ b/eco/widgets/tweak_panel.py @@ -0,0 +1,445 @@ +""" +Backend-independent core of the tweak panels (Qt: eco.widgets.tweak_panel_qt, +ipywidgets: eco.widgets.tweak_panel_ipy). + +The panels offer everything the terminal tweak does -- step *2 / step /2, +neg/pos step, back to start, go to an absolute value, reset the current value +-- for 1 to 4 axes, plus a "keypress control" mode that uses the same keys as +the terminal: + +- 1 axis: left = neg dir, up = step*2, down = step/2, right = pos dir +- 2 axes: arrows move x (left/right) and y (up/down), ctrl+arrows change + the step of x (ctrl+right *2, ctrl+left /2) and y (ctrl+up *2, + ctrl+down /2) -- as in Tweak.xy_adjustable_tweak +- stacked (automatic for 3-4 axes, optional for 1-2): one keyboard row per + axis, (neg dir, step*2, step/2, pos dir), rows as in + Tweak.STACKED_KEY_ROWS +- always: s = all axes back to start, Esc = stop, q = leave keypress control +- 1 axis, arrow mode only: g / r jump into the "go to" / "reset to" field + +Nothing here imports Qt or ipywidgets, so the key logic is testable on its own. + +Two kinds of axis, same interface: +- SoftTweakAxis drives one adjustable of an eco.elements.adjustable.Tweak via + set_target_value(), stepping from the last *target* like the terminal tweak + does (fast repeated keypresses accumulate instead of being lost). +- IocTweakAxis drives a motor record's own tweak fields (TWV/TWF/TWR), as the + MotorRecord/Smaract _tweak_ioc terminal tweaks do. +""" + +import threading + + +def _fmt(value): + try: + return f"{value:1.6g}" + except Exception: + return str(value) + + +def _name_of(obj): + try: + if hasattr(obj, "alias") and hasattr(obj.alias, "get_full_name"): + return obj.alias.get_full_name() + except Exception: + pass + return str(getattr(obj, "name", obj)) + + +class _AxisBase: + """Common parts; subclasses implement _do_move/_do_set_step/_do_go.""" + + def __init__(self, obj, step): + self.obj = obj + self.name = _name_of(obj) + self.step = float(step) + self.start_value = obj.get_current_value() + self._changer = None + + def get_value(self): + return self.obj.get_current_value() + + def current_step(self): + """The step in effect right now, also if changed elsewhere (terminal + keys, the IOC); used by the recorder's settle tolerance.""" + return self.step + + def scale_step(self, factor): + self.set_step(self.step * factor) + + def set_step(self, step): + self.step = float(step) + self._do_set_step(self.step) + + def _do_set_step(self, step): + pass + + def back_to_start(self): + self.go(self.start_value) + + def reset_current_value_to(self, value): + obj = self.obj + if hasattr(obj, "reset_current_value_to"): + obj.reset_current_value_to(value) + else: + changer = obj.set_target_value(value) + if hasattr(changer, "wait"): + changer.wait() + + def stop(self): + for target in (self._changer, self.obj): + if target is not None and callable(getattr(target, "stop", None)): + try: + target.stop() + return + except Exception: + pass + + def add_value_callback(self, callback): + """Register callback(value) on value changes; returns an undo function, + or None if the object has no callback support (panels then poll).""" + obj = self.obj + if not callable(getattr(obj, "add_value_callback", None)): + return None + try: + index = obj.add_value_callback( + lambda **kw: callback(kw["value"]) if "value" in kw else callback(None) + ) + except Exception: + return None + + def remove(): + try: + obj.clear_value_callback(index=index) + except TypeError: + obj.clear_value_callback() + except Exception: + pass + + return remove + + +class SoftTweakAxis(_AxisBase): + """Axis `index` of an eco.elements.adjustable.Tweak.""" + + def __init__(self, tweak, index): + self.tweak = tweak + self.index = index + super().__init__(tweak.adjs[index], tweak.step_sizes[index]) + self.start_value = tweak.startpositions[index] + + def _do_set_step(self, step): + self.tweak.set_step_size((self.index, step)) + + def current_step(self): + return self.tweak.step_sizes[self.index] + + def _target(self): + return self.tweak.target_positions[-1][self.index] + + def _record_target(self, value): + targets = list(self.tweak.target_positions[-1]) + targets[self.index] = value + self.tweak.target_positions.append(targets) + + def move(self, sign): + self.go(self._target() + sign * self.step) + + def go(self, value): + # only the touched axis is commanded (the terminal tweak re-sends all + # targets; for a panel that would yank back an axis moved elsewhere) + self._record_target(value) + self._changer = self.obj.set_target_value(value) + + def reset_current_value_to(self, value): + super().reset_current_value_to(value) + self._record_target(value) + + +class IocTweakAxis(_AxisBase): + """A motor record (MotorRecord, SmaractRecord, SmaractStreamdevice, ...) + tweaked through its own TWV/TWF/TWR fields.""" + + def __init__(self, motor, step=None, go_to_current_value_first=False): + if hasattr(motor, "_motor") and hasattr(motor._motor, "get_pv"): + self._pv_step = motor._motor.get_pv("TWV") + self._pv_fwd = motor._motor.get_pv("TWF") + self._pv_rev = motor._motor.get_pv("TWR") + else: + from epics import PV + + self._pv_step = PV(motor.pvname + ":TWV") + self._pv_fwd = PV(motor.pvname + ":TWF.PROC") + self._pv_rev = PV(motor.pvname + ":TWR.PROC") + if not step: + step = self._pv_step.get() + try: + step = float(step) + except Exception: + step = 1.0 + super().__init__(motor, step) + self._pv_step.put(self.step) + if go_to_current_value_first: + # as _tweak_ioc does: sync VAL to RBV before the first TWF/TWR, + # off the GUI thread since it waits for the motor + def _sync(): + try: + motor.set_target_value(motor.get_current_value()).wait() + except Exception: + pass + + threading.Thread(target=_sync, daemon=True).start() + + def _do_set_step(self, step): + self._pv_step.put(step) + + def current_step(self): + try: + value = self._pv_step.get() + if value is not None: + return float(value) + except Exception: + pass + return self.step + + def move(self, sign): + (self._pv_fwd if sign > 0 else self._pv_rev).put(1) + + def go(self, value): + self._changer = self.obj.set_target_value(value) + + +# >>> key bindings >>> + +# Normalized key names used by both backends: "Left"/"Right"/"Up"/"Down", +# "Ctrl+Left" etc., "Escape", and plain lower-case characters. +ARROWS = ("Left", "Up", "Down", "Right") + + +def _stacked_rows(n): + from eco.elements.adjustable import Tweak + + rows = Tweak.STACKED_KEY_ROWS + if not 1 <= n <= len(rows): + raise ValueError(f"stacked tweak supports 1 to {len(rows)} axes, got {n}") + return list(rows[:4] if n == 4 else rows[1 : 1 + n]) + + +def use_stacked(n, stacked=None): + if n > 4: + raise ValueError(f"tweak panels support 1 to 4 axes, got {n}") + return n > 2 if stacked is None else bool(stacked) + + +def key_bindings(n, stacked=None): + """{normalized key: action} for n axes. Actions: + ("move", i, +1/-1), ("step", i, factor), ("start",), ("stop",), + ("release",), ("focus_go", i), ("focus_reset", i).""" + b = {"s": ("start",), "q": ("release",), "Escape": ("stop",)} + if use_stacked(n, stacked): + for i, (neg, dbl, half, pos) in enumerate(_stacked_rows(n)): + b[neg] = ("move", i, -1) + b[dbl] = ("step", i, 2.0) + b[half] = ("step", i, 0.5) + b[pos] = ("move", i, +1) + elif n == 1: + b.update( + { + "Left": ("move", 0, -1), + "Up": ("step", 0, 2.0), + "Down": ("step", 0, 0.5), + "Right": ("move", 0, +1), + "g": ("focus_go", 0), + "r": ("focus_reset", 0), + } + ) + elif n == 2: + b.update( + { + "Left": ("move", 0, -1), + "Right": ("move", 0, +1), + "Down": ("move", 1, -1), + "Up": ("move", 1, +1), + "Ctrl+Right": ("step", 0, 2.0), + "Ctrl+Left": ("step", 0, 0.5), + "Ctrl+Up": ("step", 1, 2.0), + "Ctrl+Down": ("step", 1, 0.5), + } + ) + else: + raise ValueError(f"{n} axes need stacked keys") + return b + + +def key_manual(names, stacked=None): + """The key manual as a list of (keys, meaning) rows.""" + n = len(names) + rows = [] + if use_stacked(n, stacked): + key_rows = _stacked_rows(n) + for i in reversed(range(n)): # top keyboard row first + neg, dbl, half, pos = key_rows[i] + rows.append( + ( + f"{neg} {dbl} {half} {pos}", + f"{names[i]}: {neg} neg dir, {dbl} step*2, {half} step/2, {pos} pos dir", + ) + ) + elif n == 1: + rows += [ + ("← →", f"{names[0]}: neg dir / pos dir"), + ("↑ ↓", "step*2 / step/2"), + ("g / r", "type a go-to / reset-to value (Enter applies)"), + ] + else: + rows += [ + ("← →", f"x = {names[0]}: neg dir / pos dir"), + ("↓ ↑", f"y = {names[1]}: neg dir / pos dir"), + ("ctrl+→ ctrl+←", "x step*2 / step/2"), + ("ctrl+↑ ctrl+↓", "y step*2 / step/2"), + ] + rows += [ + ("s", "all axes back to start values"), + ("Esc", "stop"), + ("q", "leave keypress control"), + ] + return rows + + +class TweakController: + """The axes of one panel, and what a key/button does to them. + + `notify(text)` is how errors and the like are reported back to the + panel; panels replace it.""" + + def __init__(self, axes, stacked=None): + self.axes = list(axes) + self.stacked = use_stacked(len(self.axes), stacked) + self.bindings = key_bindings(len(self.axes), self.stacked) + self.notify = lambda text: None + + @property + def names(self): + return [axis.name for axis in self.axes] + + def manual(self): + return key_manual(self.names, self.stacked) + + def _guard(self, what, func, *args): + try: + func(*args) + return True + except Exception as exc: + self.notify(f"{what} failed: {exc}") + return False + + def move(self, i, sign): + return self._guard(f"{self.axes[i].name} move", self.axes[i].move, sign) + + def scale_step(self, i, factor): + return self._guard( + f"{self.axes[i].name} step", self.axes[i].scale_step, factor + ) + + def set_step(self, i, step): + return self._guard(f"{self.axes[i].name} step", self.axes[i].set_step, step) + + def go(self, i, value): + return self._guard(f"{self.axes[i].name} go", self.axes[i].go, value) + + def reset_current_value_to(self, i, value): + # may wait for a set; run off the calling (GUI) thread + def _run(): + self._guard( + f"{self.axes[i].name} reset", + self.axes[i].reset_current_value_to, + value, + ) + + threading.Thread(target=_run, daemon=True).start() + + def back_to_start(self): + for axis in self.axes: + self._guard(f"{axis.name} back to start", axis.back_to_start) + + def stop(self): + for axis in self.axes: + axis.stop() + + def handle_key(self, key): + """Apply the action bound to `key`. Returns the action tuple (so the + panel can handle the UI-only ones: release, focus_go, focus_reset), + or None if the key isn't bound.""" + action = self.bindings.get(key) + if action is None: + return None + kind = action[0] + if kind == "move": + self.move(action[1], action[2]) + elif kind == "step": + self.scale_step(action[1], action[2]) + elif kind == "start": + self.back_to_start() + elif kind == "stop": + self.stop() + return action + + +def axes_from_tweak(tweak): + return [SoftTweakAxis(tweak, i) for i in range(len(tweak.adjs))] + + +# >>> front-end selection >>> + + +def frontend(): + """'qt', 'ipy' or 'terminal': where an interactive tweak should show up. + + - eco desktop console, in-process kernel (InProcessInteractiveShell; the + terminal KeyPress loop has no tty there) -> 'qt' + - eco desktop console, subprocess kernel (marked by ECO_QTCONSOLE_KERNEL, + set in eco.widgets.console_kernel; qtconsole can't render ipywidgets) + -> 'qt' + - any other Jupyter kernel (notebook, lab, voila) -> 'ipy' + - terminal IPython / plain python -> 'terminal'""" + import os + + try: + from IPython import get_ipython + + shell = get_ipython() + except Exception: + shell = None + cls = shell.__class__.__name__ if shell is not None else "" + if cls == "InProcessInteractiveShell": + return "qt" + if cls == "ZMQInteractiveShell": + return "qt" if os.environ.get("ECO_QTCONSOLE_KERNEL") else "ipy" + return "terminal" + + +def tweak_panel(axes, stacked=None, backend=None, display=False, detectors=None, **kwargs): + """A tweak panel for `axes` (SoftTweakAxis/IocTweakAxis objects). + + backend: 'qt', 'ipy' or None (ipywidgets in a Jupyter kernel that isn't + an eco desktop console, Qt otherwise). A Qt panel always opens its own + window; an ipywidgets panel is returned for the cell to show, or shown + right away with display=True. + + detectors: a Detector or a list of them, recorded along the tweak and + plotted in the panel (see eco.widgets.tweak_recorder); the recorder is + the panel's .recorder.""" + controller = TweakController(axes, stacked=stacked) + if detectors is not None: + from eco.widgets.tweak_recorder import TweakRecorder + + kwargs["recorder"] = TweakRecorder(controller.axes, detectors) + if backend is None: + backend = "ipy" if frontend() == "ipy" else "qt" + if backend == "ipy": + from eco.widgets.tweak_panel_ipy import tweak_panel_ipy + + return tweak_panel_ipy(controller, display=display, **kwargs) + from eco.widgets.tweak_panel_qt import show_tweak_panel_qt + + return show_tweak_panel_qt(controller, **kwargs) diff --git a/eco/widgets/tweak_panel_ipy.py b/eco/widgets/tweak_panel_ipy.py new file mode 100644 index 0000000..c7803c3 --- /dev/null +++ b/eco/widgets/tweak_panel_ipy.py @@ -0,0 +1,317 @@ +""" +ipywidgets tweak panel, the notebook/lab/voila counterpart of +eco.widgets.tweak_panel_qt: same rows, buttons and key bindings (logic in +eco.widgets.tweak_panel). + +Keypress control uses ipyevents on the box that shows the key manual: keys act +while the mouse pointer is over that box (ipyevents' rule), so typing anywhere +else in the notebook never moves anything. Without ipyevents installed the +panel still works by buttons, and says so. +""" + +import threading + +import ipywidgets as widgets + +from eco.widgets.tweak_panel import _fmt + +_ARROW_KEYS = { + "ArrowLeft": "Left", + "ArrowUp": "Up", + "ArrowDown": "Down", + "ArrowRight": "Right", +} + + +def normalize_key_event(event): + """ipyevents keydown dict -> eco.widgets.tweak_panel normalized key.""" + key = event.get("key", "") + if key in _ARROW_KEYS: + name = _ARROW_KEYS[key] + return "Ctrl+" + name if event.get("ctrlKey") else name + if key == "Escape": + return "Escape" + if len(key) == 1: + return key.lower() + return None + + +def _manual_html(rows): + body = "".join( + f"{keys}{meaning}" + for keys, meaning in rows + ) + return ( + "" + body + "
" + "keys act while the mouse pointer is over this box" + ) + + +def render_png(recorder): + """The recorder's plot as PNG bytes (Agg, no pyplot: safe off the main + thread).""" + import io + + from matplotlib.backends.backend_agg import FigureCanvasAgg + from matplotlib.figure import Figure + + from eco.widgets.tweak_recorder import draw_tweak_plot, figure_size + + fig = Figure(figsize=figure_size(recorder), dpi=90) + FigureCanvasAgg(fig) + draw_tweak_plot(fig, recorder) + buf = io.BytesIO() + fig.savefig(buf, format="png") + return buf.getvalue() + + +class TweakPanelIpy: + def __init__( + self, controller, title=None, poll_interval=0.3, keypress=True, recorder=None + ): + self.controller = controller + self.recorder = recorder + self._stop_event = threading.Event() + self.status = widgets.HTML("") + controller.notify = self._message + + key_of = {action: key for key, action in controller.bindings.items()} + btn_layout = widgets.Layout(width="40px") + num_layout = widgets.Layout(width="100px") + + self._value_labels = [] + self._step_inputs = [] + self._go_inputs = [] + self._reset_inputs = [] + rows = [ + widgets.HTML( + f"{title or 'tweak ' + ', '.join(controller.names)}" + ) + ] + for i, axis in enumerate(controller.axes): + value_label = widgets.HTML("…", layout=widgets.Layout(width="110px")) + self._value_labels.append(value_label) + + buttons = [] + for text, tip, action in ( + ("◀", "neg dir", ("move", i, -1)), + ("×2", "step*2", ("step", i, 2.0)), + ("÷2", "step/2", ("step", i, 0.5)), + ("▶", "pos dir", ("move", i, +1)), + ): + key = key_of.get(action) + btn = widgets.Button( + description=text, + tooltip=f"{tip} (key: {key})" if key else tip, + layout=btn_layout, + ) + btn.on_click(lambda _b, a=action: self._do(a)) + buttons.append(btn) + + step_input = widgets.FloatText( + value=axis.step, layout=num_layout, continuous_update=False + ) + step_input.observe( + lambda change, i=i: self._set_step(i, change["new"]), names="value" + ) + self._step_inputs.append(step_input) + go_input = widgets.FloatText(layout=num_layout) + go_btn = widgets.Button(description="Go", layout=widgets.Layout(width="45px")) + go_btn.on_click( + lambda _b, i=i: self.controller.go(i, self._go_inputs[i].value) + ) + self._go_inputs.append(go_input) + reset_input = widgets.FloatText(layout=num_layout) + reset_btn = widgets.Button( + description="Set", + tooltip="redefine the current position as this value", + layout=widgets.Layout(width="45px"), + ) + reset_btn.on_click( + lambda _b, i=i: self.controller.reset_current_value_to( + i, self._reset_inputs[i].value + ) + ) + self._reset_inputs.append(reset_input) + + rows.append( + widgets.HBox( + [ + widgets.HTML(f"{axis.name}", layout=widgets.Layout(width="160px")), + value_label, + *buttons, + widgets.HTML(" step"), + step_input, + widgets.HTML(" go to"), + go_input, + go_btn, + widgets.HTML(" reset to"), + reset_input, + reset_btn, + ] + ) + ) + + start_btn = widgets.Button( + description="Back to start", + tooltip="all axes back to " + + ", ".join(_fmt(a.start_value) for a in controller.axes) + + " (key: s)", + ) + start_btn.on_click(lambda _b: self.controller.back_to_start()) + stop_btn = widgets.Button( + description="Stop", button_style="danger", tooltip="stop all axes (key: Esc)" + ) + stop_btn.on_click(lambda _b: self.controller.stop()) + close_btn = widgets.Button(description="Close") + close_btn.on_click(lambda _b: self.close()) + self.keypress_box = widgets.Checkbox( + value=keypress, description="keypress control", indent=False + ) + rows.append(widgets.HBox([start_btn, stop_btn, self.keypress_box, close_btn])) + + self.manual = widgets.HTML(_manual_html(controller.manual())) + self.key_area = widgets.Box( + [self.manual], + layout=widgets.Layout( + border="2px solid #2e9b45", padding="4px 8px", width="fit-content" + ), + ) + self._key_event = None + try: + from ipyevents import Event + + self._key_event = Event( + source=self.key_area, + watched_events=["keydown"], + prevent_default_action=True, + ) + self._key_event.on_dom_event(self._on_key) + except Exception: + self.manual.value += ( + "
keypress control needs the ipyevents package, " + "which is not available - use the buttons" + ) + self.keypress_box.observe( + lambda change: self._keypress_toggled(change["new"]), names="value" + ) + rows += [self.key_area, self.status] + self.plot = None + self._plotted_version = None + if recorder is not None: + recorder.start() + self.plot = widgets.Image(format="png") + self.ui = widgets.HBox([widgets.VBox(rows), self.plot]) + else: + self.ui = widgets.VBox(rows) + self._keypress_toggled(keypress) + + self._poll_interval = poll_interval + threading.Thread(target=self._poll_loop, daemon=True).start() + + def _do(self, action): + kind, i, arg = action + if kind == "move": + self.controller.move(i, arg) + elif self.controller.scale_step(i, arg): + self._show_step(i) + + def _show_step(self, i): + step_input = self._step_inputs[i] + step = self.controller.axes[i].step + if step_input.value != step: + step_input.value = step # observer: set_step with the same value + + def _set_step(self, i, value): + if value != self.controller.axes[i].step and value: + self.controller.set_step(i, value) + + def _keypress_toggled(self, on): + self.key_area.layout.display = None if on else "none" + + def _on_key(self, event): + if not self.keypress_box.value: + return + key = normalize_key_event(event) + action = self.controller.handle_key(key) if key else None + if action is None: + return + kind = action[0] + if kind == "step": + self._show_step(action[1]) + elif kind == "release": + self.keypress_box.value = False + elif kind in ("focus_go", "focus_reset"): + inputs = self._go_inputs if kind == "focus_go" else self._reset_inputs + try: + inputs[action[1]].focus() + except Exception: + pass + + def _poll_loop(self): + first = True + while not self._stop_event.is_set(): + for i, axis in enumerate(self.controller.axes): + try: + value = axis.get_value() + text = _fmt(value) + except Exception as exc: + value, text = None, f"error: {exc}" + if self._value_labels[i].value != text: + self._value_labels[i].value = text + if first and value is not None: + try: + self._go_inputs[i].value = value + self._reset_inputs[i].value = value + except Exception: + pass + first = False + self._refresh_plot() + self._stop_event.wait(self._poll_interval) + + def _refresh_plot(self): + rec = self.recorder + if rec is None or self._plotted_version == rec.version: + return + self._plotted_version = rec.version + try: + self.plot.value = render_png(rec) + except Exception as exc: + self._message(f"plot failed: {exc}") + + def _message(self, text): + self.status.value = f"{text}" + + def close(self): + self._stop_event.set() + if self.recorder is not None: + self.recorder.stop() + if self._key_event is not None: + try: + self._key_event.close() + except Exception: + pass + self.ui.close() + + def _ipython_display_(self): + from IPython.display import display + + display(self.ui) + + +def tweak_panel_ipy( + controller, title=None, poll_interval=0.3, keypress=True, display=False, recorder=None +): + """display=False returns the panel for the caller/cell to show.""" + panel = TweakPanelIpy( + controller, + title=title, + poll_interval=poll_interval, + keypress=keypress, + recorder=recorder, + ) + if display: + from IPython.display import display as _display + + _display(panel.ui) + return panel diff --git a/eco/widgets/tweak_panel_qt.py b/eco/widgets/tweak_panel_qt.py new file mode 100644 index 0000000..c657823 --- /dev/null +++ b/eco/widgets/tweak_panel_qt.py @@ -0,0 +1,427 @@ +""" +Qt tweak panel for 1-4 axes: per axis neg/pos step, step *2 / /2, editable +step, go-to and reset-to fields; back-to-start and stop for all; and a +"keypress control" checkbox (on by default) that makes the panel take the same +keys as the terminal tweak and shows the key manual. Logic and key bindings +live in eco.widgets.tweak_panel. + +Keys only reach the panel while it has keyboard focus (click on it); the +frame turns green while it does. Buttons never take focus, so a key always +means the bound action, except while typing in one of the number fields +(Enter applies the value and hands the keys back to the panel, Esc leaves the +field without applying). + +Usage: + from eco.widgets.tweak_panel import tweak_panel, IocTweakAxis + tweak_panel([IocTweakAxis(mot)], backend="qt") +or through Tweak.widget() / an adjustable's ._widget_tweak(). +""" + +import threading + +from qtpy import QtCore, QtWidgets + +from eco.widgets.tweak_panel import _fmt + +_open_panels = set() # strong refs: a parentless widget is otherwise collected +_app_ref = None + +_QT_ARROWS = { + QtCore.Qt.Key_Left: "Left", + QtCore.Qt.Key_Up: "Up", + QtCore.Qt.Key_Down: "Down", + QtCore.Qt.Key_Right: "Right", +} + +_FRAME_ACTIVE = "QFrame#tweakFrame { border: 2px solid #2e9b45; border-radius: 4px; }" +_FRAME_IDLE = "QFrame#tweakFrame { border: 2px solid transparent; border-radius: 4px; }" + + +def normalize_key_event(event): + """Qt key event -> eco.widgets.tweak_panel normalized key name, or None.""" + key = event.key() + if key in _QT_ARROWS: + name = _QT_ARROWS[key] + if event.modifiers() & QtCore.Qt.ControlModifier: + return "Ctrl+" + name + return name + if key == QtCore.Qt.Key_Escape: + return "Escape" + text = event.text() + if len(text) == 1 and text.isprintable(): + return text.lower() + return None + + +class _Bridge(QtCore.QObject): + value = QtCore.Signal(int, str) + message = QtCore.Signal(str) + + +class _ValueEdit(QtWidgets.QLineEdit): + """Number field: Enter applies (via `applied`) and returns focus to the + panel, Esc returns focus without applying.""" + + applied = QtCore.Signal(float) + + def __init__(self, panel, width=90): + super().__init__() + self._panel = panel + self.setFixedWidth(width) + self.returnPressed.connect(self._apply) + + def _apply(self): + try: + value = float(self.text()) + except ValueError: + self._panel._message(f"not a number: {self.text()!r}") + return + self.applied.emit(value) + self._panel.setFocus() + + def keyPressEvent(self, event): + if event.key() == QtCore.Qt.Key_Escape: + self._panel.setFocus() + return + super().keyPressEvent(event) + + +class TweakPlotQt(QtWidgets.QWidget): + """Live plot of a TweakRecorder (eco.widgets.tweak_recorder), with the + matplotlib toolbar; redraws only when the recorder has new data.""" + + def __init__(self, recorder, refresh_interval=0.5, parent=None): + super().__init__(parent) + from matplotlib.backends.backend_qtagg import ( + FigureCanvasQTAgg, + NavigationToolbar2QT, + ) + from matplotlib.figure import Figure + + from eco.widgets.tweak_recorder import figure_size + + self.recorder = recorder + self.figure = Figure(figsize=figure_size(recorder), dpi=90) + self.canvas = FigureCanvasQTAgg(self.figure) + self.canvas.setFocusPolicy(QtCore.Qt.NoFocus) + toolbar = NavigationToolbar2QT(self.canvas, self) + toolbar.setFocusPolicy(QtCore.Qt.NoFocus) + layout = QtWidgets.QVBoxLayout(self) + layout.setContentsMargins(0, 0, 0, 0) + layout.addWidget(toolbar) + layout.addWidget(self.canvas) + self._drawn_version = None + self._timer = QtCore.QTimer(self) + self._timer.timeout.connect(self.refresh) + self._timer.start(int(refresh_interval * 1000)) + self.refresh() + + def refresh(self, force=False): + if not force and self._drawn_version == self.recorder.version: + return + from eco.widgets.tweak_recorder import draw_tweak_plot + + self._drawn_version = self.recorder.version + draw_tweak_plot(self.figure, self.recorder) + self.canvas.draw_idle() + + def closeEvent(self, event): + self._timer.stop() + _open_panels.discard(self) + super().closeEvent(event) + + +class TweakPanelQt(QtWidgets.QFrame): + def __init__( + self, + controller, + title=None, + poll_interval=0.2, + keypress=True, + recorder=None, + parent=None, + ): + super().__init__(parent) + self.controller = controller + self.recorder = recorder + self.setObjectName("tweakFrame") + self.setWindowTitle(title or "tweak " + ", ".join(controller.names)) + self.setAttribute(QtCore.Qt.WA_QuitOnClose, False) + self.setFocusPolicy(QtCore.Qt.StrongFocus) + + self._bridge = _Bridge() + self._bridge.value.connect(self._set_value_label) + self._bridge.message.connect(self._show_message) + controller.notify = self._bridge.message.emit + + key_of = {action: key for key, action in controller.bindings.items()} + + main = QtWidgets.QHBoxLayout(self) + outer = QtWidgets.QVBoxLayout() + main.addLayout(outer) + grid = QtWidgets.QGridLayout() + for col, text in enumerate( + ("axis", "current", "", "", "", "", "step", "go to", "reset to") + ): + if text: + grid.addWidget(QtWidgets.QLabel(f"{text}"), 0, col) + + self._value_labels = [] + self._step_edits = [] + self._go_edits = [] + self._reset_edits = [] + for i, axis in enumerate(controller.axes): + r = i + 1 + grid.addWidget(QtWidgets.QLabel(axis.name), r, 0) + value_label = QtWidgets.QLabel("…") + value_label.setMinimumWidth(100) + value_label.setTextInteractionFlags(QtCore.Qt.TextSelectableByMouse) + grid.addWidget(value_label, r, 1) + self._value_labels.append(value_label) + + buttons = ( + ("◀", "neg dir", ("move", i, -1), lambda _=False, i=i: self._move(i, -1)), + ("×2", "step*2", ("step", i, 2.0), lambda _=False, i=i: self._scale(i, 2.0)), + ("÷2", "step/2", ("step", i, 0.5), lambda _=False, i=i: self._scale(i, 0.5)), + ("▶", "pos dir", ("move", i, +1), lambda _=False, i=i: self._move(i, +1)), + ) + for col, (text, tip, action, slot) in enumerate(buttons, start=2): + btn = QtWidgets.QPushButton(text) + btn.setFixedWidth(36) + btn.setFocusPolicy(QtCore.Qt.NoFocus) + key = key_of.get(action) + btn.setToolTip(f"{tip} (key: {_key_label(key)})" if key else tip) + btn.clicked.connect(slot) + grid.addWidget(btn, r, col) + + step_edit = _ValueEdit(self, width=80) + step_edit.setText(_fmt(axis.step)) + step_edit.applied.connect(lambda v, i=i: self._set_step(i, v)) + go_edit = _ValueEdit(self) + go_edit.applied.connect(lambda v, i=i: self.controller.go(i, v)) + reset_edit = _ValueEdit(self) + reset_edit.setToolTip("redefine the current position as this value") + reset_edit.applied.connect( + lambda v, i=i: self.controller.reset_current_value_to(i, v) + ) + for col, edit in ((6, step_edit), (7, go_edit), (8, reset_edit)): + grid.addWidget(edit, r, col) + self._step_edits.append(step_edit) + self._go_edits.append(go_edit) + self._reset_edits.append(reset_edit) + outer.addLayout(grid) + + bottom = QtWidgets.QHBoxLayout() + start_btn = QtWidgets.QPushButton("Back to start") + start_btn.setToolTip( + "all axes back to " + + ", ".join(_fmt(a.start_value) for a in controller.axes) + + " (key: s)" + ) + start_btn.clicked.connect(lambda _=False: self.controller.back_to_start()) + stop_btn = QtWidgets.QPushButton("Stop") + stop_btn.setToolTip("stop all axes (key: Esc)") + stop_btn.clicked.connect(lambda _=False: self.controller.stop()) + self.keypress_box = QtWidgets.QCheckBox("keypress control") + self.keypress_box.setChecked(keypress) + self.keypress_box.toggled.connect(self._keypress_toggled) + for w in (start_btn, stop_btn, self.keypress_box): + w.setFocusPolicy(QtCore.Qt.NoFocus) + bottom.addWidget(w) + bottom.addStretch(1) + outer.addLayout(bottom) + + self.manual = QtWidgets.QLabel(_manual_html(controller.manual())) + self.manual.setTextFormat(QtCore.Qt.RichText) + outer.addWidget(self.manual) + self.status = QtWidgets.QLabel("") + self.status.setWordWrap(True) + outer.addWidget(self.status) + outer.addStretch(1) + self._keypress_toggled(keypress) + + self.plot = None + if recorder is not None: + recorder.start() + self.plot = TweakPlotQt(recorder) + main.addWidget(self.plot, 1) + + self._stop_event = threading.Event() + self._poll_interval = poll_interval + self._poll_thread = threading.Thread(target=self._poll_loop, daemon=True) + self._poll_thread.start() + + # --- actions --- + def _move(self, i, sign): + self.controller.move(i, sign) + + def _scale(self, i, factor): + if self.controller.scale_step(i, factor): + self._step_edits[i].setText(_fmt(self.controller.axes[i].step)) + + def _set_step(self, i, value): + if self.controller.set_step(i, value): + self._step_edits[i].setText(_fmt(self.controller.axes[i].step)) + + # --- keys --- + def keypress_active(self): + return self.keypress_box.isChecked() + + def _keypress_toggled(self, on): + self.manual.setVisible(on) + self._update_frame() + if on: + self.setFocus() + + def _update_frame(self): + self.setStyleSheet( + _FRAME_ACTIVE if self.keypress_active() and self.hasFocus() else _FRAME_IDLE + ) + + def focusInEvent(self, event): + self._update_frame() + super().focusInEvent(event) + + def focusOutEvent(self, event): + self._update_frame() + super().focusOutEvent(event) + + def mousePressEvent(self, event): + self.setFocus() + super().mousePressEvent(event) + + def keyPressEvent(self, event): + if not self.keypress_active(): + return super().keyPressEvent(event) + key = normalize_key_event(event) + action = self.controller.handle_key(key) if key else None + if action is None: + return super().keyPressEvent(event) + kind = action[0] + if kind == "step": + self._step_edits[action[1]].setText(_fmt(self.controller.axes[action[1]].step)) + elif kind == "release": + self.keypress_box.setChecked(False) + elif kind in ("focus_go", "focus_reset"): + edits = self._go_edits if kind == "focus_go" else self._reset_edits + edit = edits[action[1]] + edit.setText(self._value_labels[action[1]].text()) + edit.setFocus() + edit.selectAll() + event.accept() + + # --- values / messages --- + def _poll_loop(self): + while not self._stop_event.is_set(): + for i, axis in enumerate(self.controller.axes): + try: + text = _fmt(axis.get_value()) + except Exception as exc: + text = f"error: {exc}" + self._bridge.value.emit(i, text) + self._stop_event.wait(self._poll_interval) + + def _set_value_label(self, i, text): + label = self._value_labels[i] + if label.text() != text: + label.setText(text) + + def _message(self, text): + self._bridge.message.emit(text) + + def _show_message(self, text): + self.status.setText(text) + QtCore.QTimer.singleShot(8000, lambda: self.status.setText("")) + + def closeEvent(self, event): + self._stop_event.set() + if self.recorder is not None: + self.recorder.stop() + _open_panels.discard(self) + super().closeEvent(event) + + +def _key_label(key): + return key.replace("Left", "←").replace("Right", "→").replace("Up", "↑").replace( + "Down", "↓" + ).replace("Ctrl+", "ctrl+") + + +def _manual_html(rows): + body = "".join( + f"{keys}{meaning}" + for keys, meaning in rows + ) + return ( + "" + body + "
" + "keys act while this panel has focus (green frame) - click it to give it focus" + ) + + +def _ensure_qt_app(): + """A QApplication pumped by IPython's event loop integration if there is + an IPython shell (enabling `%gui qt` if no loop is active yet). Returns + True if the caller must run the event loop itself (plain python).""" + global _app_ref + from eco.widgets.tweak_recorder import display_available + + if QtWidgets.QApplication.instance() is None and not display_available(): + raise RuntimeError("no display available for a Qt window") + try: + from IPython import get_ipython + + ip = get_ipython() + except Exception: + ip = None + if QtWidgets.QApplication.instance() is None: + _app_ref = QtWidgets.QApplication([]) + if ip is None: + return True + if getattr(ip, "active_eventloop", None) is None and ip.__class__.__name__ != ( + "InProcessInteractiveShell" + ): + try: + ip.enable_gui("qt") + except Exception: + pass + return False + + +def show_tweak_plot_qt(recorder, title=None): + """Standalone live-plot window (used by the terminal tweaks); never + blocks -- it is pumped by IPython's event loop integration, also while + the terminal tweak waits for keys.""" + _ensure_qt_app() + window = TweakPlotQt(recorder) + window.setAttribute(QtCore.Qt.WA_QuitOnClose, False) + window.setWindowTitle( + title + or "tweak " + + ", ".join(recorder.axis_names) + + " - " + + ", ".join(recorder.detector_names) + ) + _open_panels.add(window) + window.show() + return window + + +def show_tweak_panel_qt( + controller, title=None, poll_interval=0.2, keypress=True, recorder=None +): + blocking = _ensure_qt_app() + panel = TweakPanelQt( + controller, + title=title, + poll_interval=poll_interval, + keypress=keypress, + recorder=recorder, + ) + _open_panels.add(panel) + panel.show() + panel.raise_() + panel.activateWindow() + panel.setFocus() + if blocking: + QtWidgets.QApplication.instance().exec_() + return panel diff --git a/eco/widgets/tweak_recorder.py b/eco/widgets/tweak_recorder.py new file mode 100644 index 0000000..7e22811 --- /dev/null +++ b/eco/widgets/tweak_recorder.py @@ -0,0 +1,286 @@ +""" +Record detector values along a tweak, and plot them. + +TweakRecorder samples the tweak axes and the detectors in a background thread. +Each plateau of constant axis positions becomes one point: detector values are +averaged only while the positions stay put, so readings taken during a move +never mix in. A new point starts as soon as the positions settle somewhere +else -- whatever moved them (panel, terminal keys, IOC tweak, someone else). + +"Constant" means: all axes within `settle_fraction` x their current step size +of the previous reading, for `settle_samples` consecutive readings. + +draw_tweak_plot(fig, recorder) draws the points into a matplotlib Figure: +- 1 axis: one x/y plot per detector, detector vs axis position +- 2+ axes: a vertical stack sharing "step" as x, detectors on top, then one + plot per axis position + +Used by the Qt/ipywidgets tweak panels (embedded plot) and by the terminal +tweaks (separate Qt window, see terminal_recording). +""" + +import contextlib +import math +import threading +import time + +import numpy as np + +from eco.widgets.tweak_panel import _name_of + + +def _as_float(value): + """Scalar -> float; array-like -> mean of its elements; else NaN.""" + if value is None: + return math.nan + try: + return float(value) + except Exception: + pass + try: + arr = np.asarray(value, dtype=float) + return float(np.nanmean(arr)) if np.isfinite(arr).any() else math.nan + except Exception: + return math.nan + + +def _as_detector_list(detectors): + if detectors is None: + return [] + if isinstance(detectors, (list, tuple)): + return list(detectors) + return [detectors] + + +class TweakRecorder: + def __init__( + self, + axes, + detectors, + interval=0.1, + settle_fraction=0.1, + settle_samples=2, + ): + self.axes = list(axes) + self.detectors = _as_detector_list(detectors) + self.axis_names = [a.name for a in self.axes] + self.detector_names = [_name_of(d) for d in self.detectors] + self.interval = interval + self.settle_fraction = settle_fraction + self.settle_samples = settle_samples + self.version = 0 # bumped on every change, for cheap redraw checks + self._points = [] # dicts, see points + self._lock = threading.Lock() + self._stop_event = threading.Event() + self._thread = None + self._last_positions = None + self._stable_count = 0 + + # --- running --- + def start(self): + if self._thread is None: + self._stop_event.clear() + self._thread = threading.Thread(target=self._run, daemon=True) + self._thread.start() + return self + + def stop(self): + self._stop_event.set() + self._thread = None + + def _run(self): + while not self._stop_event.is_set(): + t0 = time.time() + try: + self.sample() + except Exception: + pass + self._stop_event.wait(max(0.0, self.interval - (time.time() - t0))) + + # --- sampling --- + def _tolerances(self): + tols = [] + for axis in self.axes: + try: + step = axis.current_step() + except Exception: + step = getattr(axis, "step", 0.0) + tols.append(self.settle_fraction * abs(step or 0.0)) + return tols + + @staticmethod + def _same(a, b, tols): + return all(abs(x - y) <= t for x, y, t in zip(a, b, tols)) + + def sample(self): + """Take one reading (called by the thread; public for tests).""" + positions = [_as_float(a.get_value()) for a in self.axes] + if any(math.isnan(p) for p in positions): + return + tols = self._tolerances() + last, self._last_positions = self._last_positions, positions + if last is not None and self._same(positions, last, tols): + self._stable_count += 1 + else: + self._stable_count = 0 + if self._stable_count < self.settle_samples - 1: + return # moving, or not yet settled + + values = [] + for det in self.detectors: + try: + values.append(_as_float(det.get_current_value())) + except Exception: + values.append(math.nan) + + with self._lock: + current = self._points[-1] if self._points else None + if current is None or not self._same( + positions, current["_position_sums"] / current["n_samples"], tols + ): + current = { + "step": len(self._points), + "time": time.time(), + "n_samples": 0, + "_position_sums": np.zeros(len(positions)), + "_values": [[] for _ in self.detectors], + } + self._points.append(current) + current["n_samples"] += 1 + current["_position_sums"] += positions + for store, value in zip(current["_values"], values): + if not math.isnan(value): + store.append(value) + self.version += 1 + + # --- results --- + @property + def points(self): + """One dict per plateau: step, time, n_samples, positions (mean per + axis), values (mean per detector), stds (per detector).""" + out = [] + with self._lock: + for p in self._points: + vals = [np.asarray(v) for v in p["_values"]] + out.append( + { + "step": p["step"], + "time": p["time"], + "n_samples": p["n_samples"], + "positions": [float(v) for v in p["_position_sums"] / p["n_samples"]], + "values": [float(v.mean()) if v.size else math.nan for v in vals], + "stds": [float(v.std()) if v.size else math.nan for v in vals], + } + ) + return out + + def arrays(self): + """(steps, positions[n_points, n_axes], values[n_points, n_detectors])""" + pts = self.points + steps = np.array([p["step"] for p in pts]) + pos = np.array([p["positions"] for p in pts]).reshape(len(pts), len(self.axes)) + val = np.array([p["values"] for p in pts]).reshape(len(pts), len(self.detectors)) + return steps, pos, val + + def to_dataframe(self): + import pandas as pd + + rows = [] + for p in self.points: + row = {"step": p["step"], "time": p["time"], "n_samples": p["n_samples"]} + row.update(zip(self.axis_names, p["positions"])) + row.update(zip(self.detector_names, p["values"])) + row.update((f"{n}_std", s) for n, s in zip(self.detector_names, p["stds"])) + rows.append(row) + return pd.DataFrame(rows) + + +def plot_rows(recorder): + return len(recorder.detectors) if len(recorder.axes) == 1 else ( + len(recorder.detectors) + len(recorder.axes) + ) + + +def draw_tweak_plot(fig, recorder): + """(Re)draw the recorder's points into matplotlib Figure `fig`.""" + fig.clear() + steps, pos, val = recorder.arrays() + n_rows = max(plot_rows(recorder), 1) + axs = fig.subplots(n_rows, 1, sharex=True, squeeze=False)[:, 0] + style = dict(marker="o", markersize=4, linewidth=0.8, color="C0") + last = dict(marker="o", markersize=7, linestyle="none", color="C3") + + if len(recorder.axes) == 1: + x = pos[:, 0] if len(steps) else [] + for ax, i in zip(axs, range(len(recorder.detectors))): + if len(steps): + ax.plot(x, val[:, i], **style) + ax.plot(x[-1:], val[-1:, i], **last) + ax.set_ylabel(recorder.detector_names[i]) + axs[-1].set_xlabel(recorder.axis_names[0]) + else: + series = [(n, val[:, i] if len(steps) else []) for i, n in enumerate(recorder.detector_names)] + series += [(n, pos[:, i] if len(steps) else []) for i, n in enumerate(recorder.axis_names)] + for ax, (name, y) in zip(axs, series): + if len(steps): + ax.plot(steps, y, **style) + ax.plot(steps[-1:], y[-1:], **last) + ax.set_ylabel(name) + axs[-1].set_xlabel("step") + for ax in axs: + ax.grid(True, alpha=0.3) + ax.yaxis.label.set_fontsize(8) + ax.tick_params(labelsize=7) + fig.tight_layout() + + +def figure_size(recorder): + return (5.0, max(2.8, 1.7 * plot_rows(recorder))) + + +def display_available(): + """False where creating a QApplication would abort the process (no X / + Wayland display, e.g. a plain ssh session).""" + import os + import sys + + if os.environ.get("QT_QPA_PLATFORM"): + return True + if not sys.platform.startswith("linux"): + return True + return bool(os.environ.get("DISPLAY") or os.environ.get("WAYLAND_DISPLAY")) + + +@contextlib.contextmanager +def terminal_recording(axes, detectors, owner=None, label=None): + """Record (and, where a display is available, live-plot in a Qt window) + `detectors` during a terminal tweak of `axes`. The recorder is left on + `owner._tweak_recorder` (if given) for later access, `label` says where + in the final message; the plot window stays open after the tweak ends.""" + detectors = _as_detector_list(detectors) + if not detectors: + yield None + return + recorder = TweakRecorder(axes, detectors).start() + if owner is not None: + try: + owner._tweak_recorder = recorder + except Exception: + pass + if display_available(): + try: + from eco.widgets.tweak_panel_qt import show_tweak_plot_qt + + recorder._plot_window = show_tweak_plot_qt(recorder) + except Exception as exc: + print(f"tweak: no live plot ({exc}); still recording") + else: + print("tweak: no display for a live plot; still recording") + try: + yield recorder + finally: + recorder.stop() + print( + f"\ntweak recorded {len(recorder.points)} points " + f"({label or 'recorder'}.to_dataframe() for the data)" + ) diff --git a/tests/test_tweak_panel.py b/tests/test_tweak_panel.py new file mode 100644 index 0000000..cd25c2b --- /dev/null +++ b/tests/test_tweak_panel.py @@ -0,0 +1,329 @@ +"""Tweak panels (eco.widgets.tweak_panel[_qt|_ipy]): key bindings, controller, +the soft (Tweak) axis, and both panels driven by key events.""" + +import os +import time + +import pytest + +from eco.widgets import tweak_panel as tp + + +class FakeAxis: + def __init__(self, name, value=0.0, step=1.0): + self.name = name + self.value = value + self.step = step + self.start_value = value + self.log = [] + + def get_value(self): + return self.value + + def move(self, sign): + self.value += sign * self.step + self.log.append(("move", sign)) + + def scale_step(self, factor): + self.step *= factor + + def set_step(self, step): + self.step = step + + def go(self, value): + self.value = value + self.log.append(("go", value)) + + def back_to_start(self): + self.go(self.start_value) + + def reset_current_value_to(self, value): + self.value = value + self.log.append(("reset", value)) + + def stop(self): + self.log.append(("stop",)) + + +def axes(n): + return [FakeAxis(f"a{i}") for i in range(n)] + + +# --- bindings --- + + +def test_single_axis_uses_arrows_like_the_terminal(): + b = tp.key_bindings(1) + assert b["Left"] == ("move", 0, -1) + assert b["Up"] == ("step", 0, 2.0) + assert b["Down"] == ("step", 0, 0.5) + assert b["Right"] == ("move", 0, +1) + assert b["s"] == ("start",) and b["q"] == ("release",) + + +def test_two_axes_use_xy_arrows_and_ctrl_arrows_for_steps(): + b = tp.key_bindings(2) + assert b["Left"] == ("move", 0, -1) and b["Right"] == ("move", 0, +1) + assert b["Down"] == ("move", 1, -1) and b["Up"] == ("move", 1, +1) + assert b["Ctrl+Right"] == ("step", 0, 2.0) and b["Ctrl+Left"] == ("step", 0, 0.5) + assert b["Ctrl+Up"] == ("step", 1, 2.0) and b["Ctrl+Down"] == ("step", 1, 0.5) + + +@pytest.mark.parametrize( + "n, rows", + [ + (1, ["jkl;"]), + (2, ["jkl;", "uiop"]), + (3, ["jkl;", "uiop", "7890"]), + (4, ["m,./", "jkl;", "uiop", "7890"]), + ], +) +def test_stacked_rows(n, rows): + b = tp.key_bindings(n, stacked=True) + for i, (neg, dbl, half, pos) in enumerate(rows): + assert b[neg] == ("move", i, -1) + assert b[dbl] == ("step", i, 2.0) + assert b[half] == ("step", i, 0.5) + assert b[pos] == ("move", i, +1) + assert "Left" not in b + + +def test_stacked_is_automatic_above_two_axes_and_limited_to_four(): + assert not tp.use_stacked(1) and not tp.use_stacked(2) + assert tp.use_stacked(3) and tp.use_stacked(4) + with pytest.raises(ValueError): + tp.key_bindings(5) + + +def test_manual_names_every_key_and_lists_the_top_row_first(): + manual = tp.key_manual(["x", "y", "z", "w"]) + keys = [k for k, _ in manual] + assert keys[:4] == ["7 8 9 0", "u i o p", "j k l ;", "m , . /"] + assert manual[0][1].startswith("w:") + assert {"s", "Esc", "q"} <= set(keys) + + +# --- controller --- + + +def test_controller_applies_keys_to_the_right_axis(): + a = axes(3) + c = tp.TweakController(a) + c.handle_key(";") + c.handle_key("i") + c.handle_key("p") + c.handle_key("7") + assert a[0].value == 1 and a[1].step == 2 and a[1].value == 2 and a[2].value == -1 + assert c.handle_key("x") is None + assert c.handle_key("q") == ("release",) + c.handle_key("s") + assert [x.value for x in a] == [0, 0, 0] + c.handle_key("Escape") + assert all(x.log[-1] == ("stop",) for x in a) + + +def test_controller_reports_errors_instead_of_raising(): + a = axes(1) + a[0].move = lambda sign: 1 / 0 + c = tp.TweakController(a) + messages = [] + c.notify = messages.append + c.handle_key("Right") + assert messages and "a0 move failed" in messages[0] + + +# --- soft axis on a real Tweak --- + + +class Changer: + def is_alive(self): + return False + + def wait(self): + pass + + +class Adj: + def __init__(self, name, value=0.0): + self.name = name + self.value = value + self.sets = [] + + def get_current_value(self): + return self.value + + def set_target_value(self, value): + self.sets.append(value) + self.value = value + return Changer() + + def reset_current_value_to(self, value): + self.value = value + + +def test_soft_axis_steps_from_the_target_and_commands_only_its_adjustable(): + from eco.elements.adjustable import Tweak + + x, y = Adj("x"), Adj("y", 5.0) + t = Tweak((x, 1.0), (y, 1.0)) + ax, ay = tp.axes_from_tweak(t) + x.value = 0.4 # still moving: the next step goes from the target + ax.move(+1) + ax.move(+1) + assert x.sets == [1.0, 2.0] and y.sets == [] + ax.scale_step(0.5) + assert t.step_sizes == [0.5, 1.0] + ax.move(-1) + assert x.sets[-1] == 1.5 + ay.go(7.0) + ax.reset_current_value_to(10.0) + ax.move(+1) + assert x.sets[-1] == 10.5 # relative to the redefined position + ax.back_to_start() + assert x.sets[-1] == 0.0 and y.sets == [7.0] + + +# --- front-end selection --- + + +def test_frontend(monkeypatch): + import IPython + + class InProcessInteractiveShell: + pass + + class ZMQInteractiveShell: + pass + + monkeypatch.delenv("ECO_QTCONSOLE_KERNEL", raising=False) + monkeypatch.setattr(IPython, "get_ipython", lambda: None) + assert tp.frontend() == "terminal" + monkeypatch.setattr(IPython, "get_ipython", lambda: InProcessInteractiveShell()) + assert tp.frontend() == "qt" + monkeypatch.setattr(IPython, "get_ipython", lambda: ZMQInteractiveShell()) + assert tp.frontend() == "ipy" + monkeypatch.setenv("ECO_QTCONSOLE_KERNEL", "1") + assert tp.frontend() == "qt" + + +# --- Qt panel --- + + +@pytest.fixture +def qt_panel(): + pytest.importorskip("qtpy") + os.environ.setdefault("QT_QPA_PLATFORM", "offscreen") + from qtpy import QtWidgets + + from eco.widgets.tweak_panel_qt import TweakPanelQt + + app = QtWidgets.QApplication.instance() or QtWidgets.QApplication([]) + made = [] + + def make(n, **kwargs): + a = axes(n) + panel = TweakPanelQt(tp.TweakController(a, **kwargs), poll_interval=0.05) + made.append(panel) + return panel, a + + yield make + for panel in made: + panel.close() + app.processEvents() + + +def test_qt_keys_act_like_the_terminal(qt_panel): + from qtpy import QtCore + from qtpy.QtTest import QTest + + panel, a = qt_panel(1) + assert panel.keypress_active() and not panel.manual.isHidden() + QTest.keyClick(panel, QtCore.Qt.Key_Right) + QTest.keyClick(panel, QtCore.Qt.Key_Up) + QTest.keyClick(panel, QtCore.Qt.Key_Right) + assert a[0].value == 3 and panel._step_edits[0].text() == "2" + QTest.keyClick(panel, QtCore.Qt.Key_Q) + assert not panel.keypress_active() and panel.manual.isHidden() + QTest.keyClick(panel, QtCore.Qt.Key_Right) + assert a[0].value == 3 # keypress control off + + +def test_qt_ctrl_arrows_and_stacked_keys(qt_panel): + from qtpy import QtCore + from qtpy.QtTest import QTest + + panel, a = qt_panel(2) + QTest.keyClick(panel, QtCore.Qt.Key_Up, QtCore.Qt.ControlModifier) + QTest.keyClick(panel, QtCore.Qt.Key_Up) + assert a[1].step == 2 and a[1].value == 2 and a[0].value == 0 + + panel, a = qt_panel(4) + for key in (QtCore.Qt.Key_Slash, QtCore.Qt.Key_0, QtCore.Qt.Key_Semicolon): + QTest.keyClick(panel, key) + assert [x.value for x in a] == [1, 1, 0, 1] + + +def test_qt_buttons_and_fields(qt_panel): + from qtpy import QtCore, QtWidgets + from qtpy.QtTest import QTest + + panel, a = qt_panel(1) + buttons = [b for b in panel.findChildren(QtWidgets.QPushButton) if b.text() == "▶"] + assert buttons[0].focusPolicy() == QtCore.Qt.NoFocus + assert "key: →" in buttons[0].toolTip() + buttons[0].click() + assert a[0].value == 1 + go = panel._go_edits[0] + go.setText("4.5") + QTest.keyClick(go, QtCore.Qt.Key_Return) + assert a[0].log[-1] == ("go", 4.5) + go.setText("nonsense") + QTest.keyClick(go, QtCore.Qt.Key_Return) + assert a[0].log[-1] == ("go", 4.5) + step = panel._step_edits[0] + step.setText("0.25") + QTest.keyClick(step, QtCore.Qt.Key_Return) + assert a[0].step == 0.25 + + +def test_qt_value_labels_are_polled(qt_panel): + from qtpy import QtWidgets + + panel, a = qt_panel(1) + a[0].value = 1.25 + app = QtWidgets.QApplication.instance() + for _ in range(50): + app.processEvents() + if panel._value_labels[0].text() == "1.25": + break + time.sleep(0.02) + assert panel._value_labels[0].text() == "1.25" + + +# --- ipywidgets panel --- + + +def test_ipy_keys_and_buttons(): + pytest.importorskip("ipywidgets") + from eco.widgets.tweak_panel_ipy import TweakPanelIpy, normalize_key_event + + assert normalize_key_event({"key": "ArrowUp", "ctrlKey": True}) == "Ctrl+Up" + assert normalize_key_event({"key": "J"}) == "j" + assert normalize_key_event({"key": "Shift"}) is None + + a = axes(3) + panel = TweakPanelIpy(tp.TweakController(a), poll_interval=0.05) + try: + panel._on_key({"key": ";"}) + panel._on_key({"key": "i"}) + assert a[0].value == 1 and a[1].step == 2 + assert panel._step_inputs[1].value == 2 + panel._step_inputs[2].value = 0.5 + assert a[2].step == 0.5 + panel._on_key({"key": "q"}) + assert panel.keypress_box.value is False + assert panel.key_area.layout.display == "none" + panel._on_key({"key": ";"}) + assert a[0].value == 1 + finally: + panel.close() diff --git a/tests/test_tweak_recorder.py b/tests/test_tweak_recorder.py new file mode 100644 index 0000000..0dc6dc6 --- /dev/null +++ b/tests/test_tweak_recorder.py @@ -0,0 +1,240 @@ +"""Detector recording along a tweak (eco.widgets.tweak_recorder) and its plots +in the Qt/ipywidgets tweak panels.""" + +import math +import os + +import pytest + +from eco.widgets import tweak_panel as tp +from eco.widgets.tweak_recorder import TweakRecorder, draw_tweak_plot, plot_rows + + +class Axis: + def __init__(self, name, value=0.0, step=1.0): + self.name = name + self.value = value + self.step = step + self.start_value = value + + def get_value(self): + return self.value + + def current_step(self): + return self.step + + def move(self, sign): + self.value += sign * self.step + + def scale_step(self, factor): + self.step *= factor + + def set_step(self, step): + self.step = step + + def go(self, value): + self.value = value + + def back_to_start(self): + self.value = self.start_value + + def reset_current_value_to(self, value): + self.value = value + + def stop(self): + pass + + +class Det: + def __init__(self, name, values): + self.name = name + self.values = list(values) + + def get_current_value(self): + return self.values.pop(0) + + +def test_values_are_averaged_per_plateau_and_moves_are_skipped(): + x = Axis("x") + det = Det("diode", [1, 3, 99, 10, 20, 30]) + rec = TweakRecorder([x], det, settle_samples=2) + rec.sample() # first reading: not settled yet, detector not read + rec.sample() # settled -> 1 + rec.sample() # -> 3 + x.value = 0.5 # moving + rec.sample() # position changed: not read + x.value = 1.0 + rec.sample() # moved again: not read + rec.sample() # settled at 1.0 -> 99 + rec.sample() # -> 10 + pts = rec.points + assert [p["positions"] for p in pts] == [[0.0], [1.0]] + assert pts[0]["values"] == [2.0] and pts[0]["n_samples"] == 2 + assert pts[1]["values"] == [54.5] + assert [p["step"] for p in pts] == [0, 1] + + +def test_jitter_within_the_settle_tolerance_stays_on_one_plateau(): + x = Axis("x", step=1.0) + rec = TweakRecorder([x], Det("d", [1, 2, 3, 4]), settle_fraction=0.1) + for v in (0.0, 0.03, -0.04, 0.02, 0.05): + x.value = v + rec.sample() + assert len(rec.points) == 1 and rec.points[0]["n_samples"] == 4 + x.step = 0.1 # smaller step -> tighter tolerance: 0.05 is now a new spot + x.value = 0.3 + rec.sample() + rec.sample() + assert len(rec.points) == 2 + + +def test_failed_and_array_detector_reads(): + class Flaky: + name = "flaky" + + def __init__(self): + self.n = 0 + + def get_current_value(self): + self.n += 1 + if self.n == 1: + raise RuntimeError("timeout") + return None if self.n == 2 else [1.0, 3.0] + + x = Axis("x") + rec = TweakRecorder([x], [Flaky()], settle_samples=1) + for _ in range(3): + rec.sample() + p = rec.points[0] + assert p["n_samples"] == 3 and p["values"] == [2.0] # array -> its mean + + +def test_multi_axis_points_and_arrays(): + x, y = Axis("x"), Axis("y", 5.0) + rec = TweakRecorder([x, y], [Det("a", range(100)), Det("b", range(100))], settle_samples=1) + rec.sample() + y.value = 6.0 + rec.sample() + steps, pos, val = rec.arrays() + assert list(steps) == [0, 1] + assert pos.tolist() == [[0.0, 5.0], [0.0, 6.0]] + assert val.shape == (2, 2) + + +def test_dataframe(): + pytest.importorskip("pandas") + x = Axis("x") + rec = TweakRecorder([x], Det("d", [1.0, 2.0]), settle_samples=1) + rec.sample() + rec.sample() + df = rec.to_dataframe() + assert list(df.columns) == ["step", "time", "n_samples", "x", "d", "d_std"] + assert df["d"].tolist() == [1.5] + + +def test_recorder_thread_starts_and_stops(): + import time + + x = Axis("x") + rec = TweakRecorder([x], Det("d", [1.0] * 1000), interval=0.01).start() + time.sleep(0.15) + rec.stop() + assert rec.points and rec.points[0]["n_samples"] > 3 + + +# --- plots --- + + +def _figure(): + pytest.importorskip("matplotlib") + from matplotlib.backends.backend_agg import FigureCanvasAgg + from matplotlib.figure import Figure + + fig = Figure() + FigureCanvasAgg(fig) + return fig + + +def test_one_axis_plots_each_detector_against_the_position(): + fig = _figure() + x = Axis("x") + rec = TweakRecorder([x], [Det("a", range(9)), Det("b", range(9))], settle_samples=1) + for v in (0.0, 1.0, 2.0): + x.value = v + rec.sample() + draw_tweak_plot(fig, rec) + axs = fig.axes + assert plot_rows(rec) == 2 and len(axs) == 2 + assert [a.get_ylabel() for a in axs] == ["a", "b"] + assert axs[-1].get_xlabel() == "x" + assert list(axs[0].lines[0].get_xdata()) == [0.0, 1.0, 2.0] + + +def test_several_axes_stack_detectors_on_top_over_the_step_number(): + fig = _figure() + axes = [Axis("x"), Axis("y"), Axis("z")] + rec = TweakRecorder(axes, [Det("a", range(9))], settle_samples=1) + for v in (0.0, 1.0): + axes[1].value = v + rec.sample() + draw_tweak_plot(fig, rec) + axs = fig.axes + assert [a.get_ylabel() for a in axs] == ["a", "x", "y", "z"] + assert axs[-1].get_xlabel() == "step" + assert list(axs[2].lines[0].get_ydata()) == [0.0, 1.0] + + +def test_empty_recorder_draws_without_points(): + fig = _figure() + draw_tweak_plot(fig, TweakRecorder([Axis("x")], Det("a", []))) + assert len(fig.axes) == 1 + + +# --- panels --- + + +def test_qt_panel_embeds_a_live_plot(): + pytest.importorskip("qtpy") + pytest.importorskip("matplotlib") + os.environ.setdefault("QT_QPA_PLATFORM", "offscreen") + from qtpy import QtWidgets + + from eco.widgets.tweak_panel_qt import TweakPanelQt + + app = QtWidgets.QApplication.instance() or QtWidgets.QApplication([]) + x = Axis("x") + rec = TweakRecorder([x], Det("d", [float(i) for i in range(10000)]), interval=0.01) + panel = TweakPanelQt(tp.TweakController([x]), recorder=rec) + try: + assert panel.plot is not None and rec._thread is not None + import time + + time.sleep(0.1) + panel.plot.refresh() + assert panel.plot._drawn_version == rec.version or rec.version > 0 + assert panel.plot.figure.axes[0].get_ylabel() == "d" + finally: + panel.close() + app.processEvents() + assert rec._stop_event.is_set() + + +def test_ipy_panel_renders_the_plot_as_png(): + pytest.importorskip("ipywidgets") + pytest.importorskip("matplotlib") + from eco.widgets.tweak_panel_ipy import TweakPanelIpy + + x = Axis("x") + rec = TweakRecorder([x], Det("d", [1.0, 2.0, 3.0]), settle_samples=1) + rec.sample() + panel = TweakPanelIpy(tp.TweakController([x]), recorder=rec, poll_interval=10) + try: + import time + + for _ in range(100): # rendered by the panel's poll thread + if len(panel.plot.value): + break + time.sleep(0.02) + assert bytes(panel.plot.value[:4]) == b"\x89PNG" + finally: + panel.close()