Bernina/eco
0
0 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

tweak: stacked keys, Qt/ipywidgets panels, detector recording

Browse Source 
- 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.
This commit is contained in:
Henrik Lemke committed 2026-10-08 08:38:57 +02:00
1 parent d64dffe14f
commit 9bc4f8d01e
9 files changed
+2298 -628
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

eco/devices_general/motors.py
+63 -205
View File
@@ -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"<b>Current position:</b> {_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"<b>Step size:</b> {_format_value(step_value)}")
help_label = widgets.HTML(
value=(
"<b>Controls:</b> 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"<b>Current position:</b> {_format_value(value)}"
status_label.value = f"<b>Step size:</b> {_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"<b>Step size:</b> {_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 = "<b>Command:</b> left"
def _click_right(_):
pvf.put(1)
status_label.value = "<b>Command:</b> right"
def _click_go(_):
try:
self.set_target_value(float(go_input.value), check=True).wait()
_refresh_current()
status_label.value = f"<b>Moved to:</b> {_format_value(go_input.value)}"
except Exception as exc:
status_label.value = f"<b>Error:</b> {exc}"
def _click_set(_):
try:
self.reset_current_value_to(float(set_input.value))
_refresh_current()
status_label.value = (
f"<b>Offset reset to:</b> {_format_value(set_input.value)}"
)
except Exception as exc:
status_label.value = f"<b>Error:</b> {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 = "<b>Tweak UI closed.</b>"
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 = {
eco/elements/adjustable.py
+188 -423
View File
@@ -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", "<tweak>.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"<b>Current position:</b> {_format_value(current_value)}"
)
status_label = widgets.HTML(
value=f"<b>Step size:</b> {_format_value(step_value)}"
)
help_label = widgets.HTML(
value=(
"<b>Controls:</b> 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"<b>Current position:</b> {_format_value(value)}"
status_label.value = f"<b>Step size:</b> {_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"<b>Step size:</b> {_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"<b>Error:</b> {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"<b>Error:</b> {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 = "<b>Tweak UI closed.</b>"
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"<b>x ({x_adj.name}):</b> {_format_value(x_current)}"
y_label.value = f"<b>y ({y_adj.name}):</b> {_format_value(y_current)}"
status_label.value = (
f"<b>Step sizes:</b> 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"<b>x ({x_adj.name}):</b> {_format_value(x_adj.get_current_value())}"
)
y_label = widgets.HTML(
value=f"<b>y ({y_adj.name}):</b> {_format_value(y_adj.get_current_value())}"
)
status_label = widgets.HTML(
value=(
f"<b>Step sizes:</b> x={_format_value(x_step_input.value)}, y={_format_value(y_step_input.value)}"
)
)
help_label = widgets.HTML(
value=(
"<b>Controls:</b> 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"<b>Step sizes:</b> 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 = "<b>Tweak UI closed.</b>"
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])
eco/widgets/console_kernel.py
+3
View File
@@ -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
eco/widgets/tweak_panel.py
+445
View File
@@ -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)
eco/widgets/tweak_panel_ipy.py
+317
View File
@@ -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"<tr><td style='padding-right:12px'><b><code>{keys}</code></b></td><td>{meaning}</td></tr>"
for keys, meaning in rows
)
return (
"<table>" + body + "</table>"
"<i>keys act while the mouse pointer is over this box</i>"
)
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"<b>{title or 'tweak ' + ', '.join(controller.names)}</b>"
)
]
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"<b>{axis.name}</b>", layout=widgets.Layout(width="160px")),
value_label,
*buttons,
widgets.HTML("&nbsp;step"),
step_input,
widgets.HTML("&nbsp;go to"),
go_input,
go_btn,
widgets.HTML("&nbsp;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 += (
"<br><b>keypress control needs the ipyevents package, "
"which is not available - use the buttons</b>"
)
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"<b>{text}</b>"
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
eco/widgets/tweak_panel_qt.py
+427
View File
@@ -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"<b>{text}</b>"), 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"<tr><td style='padding-right:12px'><b><tt>{keys}</tt></b></td><td>{meaning}</td></tr>"
for keys, meaning in rows
)
return (
"<table>" + body + "</table>"
"<i>keys act while this panel has focus (green frame) - click it to give it focus</i>"
)
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
eco/widgets/tweak_recorder.py
+286
View File
@@ -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)"
)
tests/test_tweak_panel.py
+329
View File
@@ -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()
tests/test_tweak_recorder.py
+240
View File
@@ -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()