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 (
+ ""
+ "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 (
+ ""
+ "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()