feat(bec_widgets): add data_viewer widget; rename OMNY widget classes #298

Merged
holler merged 1 commits from feat/add-data-viewer-widget into main 2026-08-17 17:25:58 +02:00
43 changed files with 1675 additions and 415 deletions
@@ -89,7 +89,7 @@ class LamniGuiTools:
if self._lamnigui_is_missing("xeyegui"):
self.lamnigui_remove_all_docks()
self.xeyegui = self.gui.lamni.new(
"XRayEye", object_name="xrayeye", timeout=self.GUI_RPC_TIMEOUT
"OMNY_XRayEye", object_name="xrayeye", timeout=self.GUI_RPC_TIMEOUT
)
# An existing xeyegui widget is reused across calls, so a Ctrl-C
# that interrupted a prior run mid-RPC (e.g. mid set_live_view_signal)
@@ -110,7 +110,7 @@ class LamniGuiTools:
if self._lamnigui_is_missing("xeyegui"):
self.lamnigui_remove_all_docks()
self.xeyegui = self.gui.lamni.new(
"XRayEye", object_name="xrayeye", timeout=self.GUI_RPC_TIMEOUT
"OMNY_XRayEye", object_name="xrayeye", timeout=self.GUI_RPC_TIMEOUT
)
self.xeyegui.switch_tab("fit")
@@ -108,7 +108,7 @@ class OMNYTools:
this method does not create it):
gui.new("test", timeout=20)
console = gui.test.new("ConsoleButtonsWidget", object_name="console", timeout=20)
console = gui.test.new("z_ConsoleButtonsWidget", object_name="console", timeout=20)
Then call, e.g. from an IPython session:
@@ -66,7 +66,7 @@ class flomniGuiTools:
self.flomnigui_show_gui()
if self._flomnigui_is_missing("xeyegui"):
self.flomnigui_remove_all_docks()
self.xeyegui = self.gui.flomni.new("XRayEye", object_name="xrayeye")
self.xeyegui = self.gui.flomni.new("OMNY_XRayEye", object_name="xrayeye")
# start live
if not dev.cam_xeye.live_mode_enabled.get():
dev.cam_xeye.live_mode_enabled.put(True)
@@ -76,7 +76,7 @@ class flomniGuiTools:
self.flomnigui_show_gui()
if self._flomnigui_is_missing("xeyegui"):
self.flomnigui_remove_all_docks()
self.xeyegui = self.gui.flomni.new("XRayEye", object_name="xrayeye")
self.xeyegui = self.gui.flomni.new("OMNY_XRayEye", object_name="xrayeye")
self.xeyegui.switch_tab("fit")
def _flomnigui_check_attribute_not_exists(self, attribute_name):
@@ -141,7 +141,7 @@ class flomniGuiTools:
# board) so it needs its own stop_all_axes() call -- also
# explained in ConsoleButtonsWidget's docstring.
self.console = self.gui.flomni.new(
"ConsoleButtonsWidget",
"z_ConsoleButtonsWidget",
object_name="console",
where="bottom",
hard_stop_device_name="ftransy",
+183 -158
View File
@@ -13,139 +13,41 @@ logger = bec_logger.logger
_Widgets = {
"ConsoleButtonsWidget": "ConsoleButtonsWidget",
"DataViewer": "DataViewer",
"OMNY_SampleStorage": "OMNY_SampleStorage",
"OMNY_TomoParams": "OMNY_TomoParams",
"OMNY_XRayEye": "OMNY_XRayEye",
"SAXSWidget": "SAXSWidget",
"SampleStorageWidget": "SampleStorageWidget",
"SlitControlWidget": "SlitControlWidget",
"TomoParamsWidget": "TomoParamsWidget",
"XRayEye": "XRayEye",
"SlitControl": "SlitControl",
"z_ConsoleButtonsWidget": "z_ConsoleButtonsWidget",
}
class ConsoleButtonsWidget(RPCBase):
"""Small Yes / No / Abort control widget, intended as a GUI replacement for"""
class DataViewer(RPCBase):
"""Main widget of Data Viewer"""
_IMPORT_MODULE = "csaxs_bec.bec_widgets.widgets.console_buttons.console_buttons"
_IMPORT_MODULE = "csaxs_bec.bec_widgets.widgets.data_viewer.data_viewer"
@property
@rpc_call
def message(self):
def remove(self):
"""
None
Cleanup the BECConnector
"""
@message.setter
@rpc_call
def message(self):
def attach(self):
"""
None
"""
@rpc_call
def response(self) -> "str":
def detach(self):
"""
Current button response: "yes", "no", or "" if no button has been
pressed (or since the last clear_response()).
Note: intentionally a plain method, not @SafeProperty. SafeProperty
only becomes a real Qt property descriptor once a setter is chained
in the class body (see `message` above); a getter-only SafeProperty
is left as an unconverted internal wrapper object, which the RPC
generator then exposes as a callable stub anyway. Since `response`
has no RPC-facing setter, a plain method avoids that trap entirely -
same pattern XRayEye uses for its own read-only `active_roi`.
"""
@rpc_call
def clear_response(self):
"""
None
Detach the widget from its parent dock widget (if widget is in the dock), making it a floating widget.
"""
class SAXSWidget(RPCBase):
"""Widget for preparing SAXS measurement tables from an image ROI."""
_IMPORT_MODULE = "csaxs_bec.bec_widgets.widgets.saxs_widget.saxs_widget"
@rpc_call
def add_scanning_row(
self,
sample_name: "str | None" = None,
exposure_time: "float | None" = None,
fast_axis: "str | None" = None,
stepper_axis: "str | None" = None,
fast_step: "float | None" = None,
stepper_step: "float | None" = None,
comment: "str" = "",
) -> "SAXSRow | None":
"""
None
"""
@rpc_call
def add_tensor_rows(
self,
sample_name: "str | None" = None,
exposure_time: "float | None" = None,
fast_axis: "str | None" = None,
stepper_axis: "str | None" = None,
fast_step: "float | None" = None,
stepper_step: "float | None" = None,
comment: "str" = "",
roty: "float | None" = None,
rotx_values: "list[float] | None" = None,
roty_values: "list[float] | None" = None,
angle_pairs: "list[tuple[float, float]] | None" = None,
) -> "list[SAXSRow]":
"""
None
"""
@rpc_call
def clear_rows(self) -> "None":
"""
None
"""
@rpc_call
def delete_selected_rows(self) -> "None":
"""
None
"""
@rpc_call
def export_csv(self, file_name: "str | None" = None, selected: "bool" = False) -> "str | None":
"""
None
"""
@rpc_call
def rows_as_dicts(self) -> "list[dict]":
"""
None
"""
@rpc_call
def set_positioners(self, x_positioner: "str", y_positioner: "str") -> "None":
"""
None
"""
@rpc_call
def mark_scan_result(self, sample_id_or_row, scan_number, success: "bool") -> "None":
"""
None
"""
@rpc_call
def move_to_selected_row(self, index: "int") -> "None":
"""
None
"""
class SampleStorageWidget(RPCBase):
class OMNY_SampleStorage(RPCBase):
"""View and correct the FlOMNI sample-storage records."""
_IMPORT_MODULE = "csaxs_bec.bec_widgets.widgets.sample_storage.sample_storage"
@@ -164,50 +66,7 @@ class SampleStorageWidget(RPCBase):
"""
class SlitControlWidget(RPCBase):
"""Interactive GUI for cSAXS slit center and size control."""
_IMPORT_MODULE = "csaxs_bec.bec_widgets.widgets.slit_control.slit_control"
@rpc_timeout(20)
@rpc_call
def set_slit(self, slit: "int"):
"""
Switch the widget to control a different slit (1–6).
"""
@rpc_call
def move_center(self, direction: "str"):
"""
Move the slit center. direction: up / down / left / right.
"""
@rpc_call
def move_size(self, direction: "str"):
"""
Adjust slit size. up/right = grow, down/left = shrink.
"""
@rpc_call
def set_step(self, value: "float"):
"""
Set step size in mm (clamped to 0.005–2.0 mm).
"""
@rpc_call
def double_step(self):
"""
None
"""
@rpc_call
def halve_step(self):
"""
None
"""
class TomoParamsWidget(RPCBase):
class OMNY_TomoParams(RPCBase):
"""Interactive GUI for tomo scan parameter editing and queue management, for"""
_IMPORT_MODULE = "csaxs_bec.bec_widgets.widgets.tomo_params.tomo_params"
@@ -266,7 +125,7 @@ class TomoParamsWidget(RPCBase):
"""
class XRayEye(RPCBase):
class OMNY_XRayEye(RPCBase):
_IMPORT_MODULE = "csaxs_bec.bec_widgets.widgets.xray_eye.x_ray_eye"
@rpc_call
@@ -447,6 +306,131 @@ class XRayEye(RPCBase):
"""
class SAXSWidget(RPCBase):
"""Widget for preparing SAXS measurement tables from an image ROI."""
_IMPORT_MODULE = "csaxs_bec.bec_widgets.widgets.saxs_widget.saxs_widget"
@rpc_call
def add_scanning_row(
self,
sample_name: "str | None" = None,
exposure_time: "float | None" = None,
fast_axis: "str | None" = None,
stepper_axis: "str | None" = None,
fast_step: "float | None" = None,
stepper_step: "float | None" = None,
comment: "str" = "",
) -> "SAXSRow | None":
"""
None
"""
@rpc_call
def add_tensor_rows(
self,
sample_name: "str | None" = None,
exposure_time: "float | None" = None,
fast_axis: "str | None" = None,
stepper_axis: "str | None" = None,
fast_step: "float | None" = None,
stepper_step: "float | None" = None,
comment: "str" = "",
roty: "float | None" = None,
rotx_values: "list[float] | None" = None,
roty_values: "list[float] | None" = None,
angle_pairs: "list[tuple[float, float]] | None" = None,
) -> "list[SAXSRow]":
"""
None
"""
@rpc_call
def clear_rows(self) -> "None":
"""
None
"""
@rpc_call
def delete_selected_rows(self) -> "None":
"""
None
"""
@rpc_call
def export_csv(self, file_name: "str | None" = None, selected: "bool" = False) -> "str | None":
"""
None
"""
@rpc_call
def rows_as_dicts(self) -> "list[dict]":
"""
None
"""
@rpc_call
def set_positioners(self, x_positioner: "str", y_positioner: "str") -> "None":
"""
None
"""
@rpc_call
def mark_scan_result(self, sample_id_or_row, scan_number, success: "bool") -> "None":
"""
None
"""
@rpc_call
def move_to_selected_row(self, index: "int") -> "None":
"""
None
"""
class SlitControl(RPCBase):
"""Interactive GUI for cSAXS slit center and size control."""
_IMPORT_MODULE = "csaxs_bec.bec_widgets.widgets.slit_control.slit_control"
@rpc_timeout(20)
@rpc_call
def set_slit(self, slit: "int"):
"""
Switch the widget to control a different slit (1–6).
"""
@rpc_call
def move_center(self, direction: "str"):
"""
Move the slit center. direction: up / down / left / right.
"""
@rpc_call
def move_size(self, direction: "str"):
"""
Adjust slit size. up/right = grow, down/left = shrink.
"""
@rpc_call
def set_step(self, value: "float"):
"""
Set step size in mm (clamped to 0.005–2.0 mm).
"""
@rpc_call
def double_step(self):
"""
None
"""
@rpc_call
def halve_step(self):
"""
None
"""
class XRayEye2DControl(RPCBase):
_IMPORT_MODULE = "csaxs_bec.bec_widgets.widgets.xray_eye.x_ray_eye"
@@ -467,3 +451,44 @@ class XRayEye2DControl(RPCBase):
"""
Detach the widget from its parent dock widget (if widget is in the dock), making it a floating widget.
"""
class z_ConsoleButtonsWidget(RPCBase):
"""Small Yes / No / Abort control widget, intended as a GUI replacement for"""
_IMPORT_MODULE = "csaxs_bec.bec_widgets.widgets.console_buttons.console_buttons"
@property
@rpc_call
def message(self):
"""
None
"""
@message.setter
@rpc_call
def message(self):
"""
None
"""
@rpc_call
def response(self) -> "str":
"""
Current button response: "yes", "no", or "" if no button has been
pressed (or since the last clear_response()).
Note: intentionally a plain method, not @SafeProperty. SafeProperty
only becomes a real Qt property descriptor once a setter is chained
in the class body (see `message` above); a getter-only SafeProperty
is left as an unconverted internal wrapper object, which the RPC
generator then exposes as a callable stub anyway. Since `response`
has no RPC-facing setter, a plain method avoids that trap entirely -
same pattern OMNY_XRayEye uses for its own read-only `active_roi`.
"""
@rpc_call
def clear_response(self):
"""
None
"""
@@ -11,7 +11,7 @@ from qtpy.QtWidgets import QHBoxLayout, QLabel, QPushButton, QVBoxLayout, QWidge
logger = bec_logger.logger
class ConsoleButtonsWidget(BECWidget, QWidget):
class z_ConsoleButtonsWidget(BECWidget, QWidget):
"""
Small Yes / No / Abort control widget, intended as a GUI replacement for
console prompts (e.g. ``OMNYTools.yesno()``) and as a hard motion-stop
@@ -177,7 +177,7 @@ class ConsoleButtonsWidget(BECWidget, QWidget):
try:
self.queue.request_scan_abortion()
except Exception:
logger.exception("ConsoleButtonsWidget: request_scan_abortion() failed")
logger.exception("z_ConsoleButtonsWidget: request_scan_abortion() failed")
# 2) SIGINT before the hard stop -- gives a client-side polling loop
# (if this is a self-abort) a head start to notice the interrupt
@@ -187,7 +187,7 @@ class ConsoleButtonsWidget(BECWidget, QWidget):
# looks like a BEC client -- it does not when the widget is
# opened standalone from the launcher menu.
if self._client_is_bec_process():
logger.warning(f"ConsoleButtonsWidget: sending SIGINT to client pid {self._client_pid}")
logger.warning(f"z_ConsoleButtonsWidget: sending SIGINT to client pid {self._client_pid}")
os.kill(self._client_pid, signal.SIGINT)
# 3) Hard motion stop(s) -- not delayed by anything above: sending
@@ -213,11 +213,11 @@ class ConsoleButtonsWidget(BECWidget, QWidget):
device = None
if device is None:
return
logger.warning(f"ConsoleButtonsWidget: hard-stopping {device_name} ({method_name})")
logger.warning(f"z_ConsoleButtonsWidget: hard-stopping {device_name} ({method_name})")
try:
getattr(device.controller, method_name)()
except Exception:
logger.exception(f"ConsoleButtonsWidget: hard motion stop of {device_name} failed")
logger.exception(f"z_ConsoleButtonsWidget: hard motion stop of {device_name} failed")
@SafeProperty(str)
def message(self):
@@ -242,7 +242,7 @@ class ConsoleButtonsWidget(BECWidget, QWidget):
is left as an unconverted internal wrapper object, which the RPC
generator then exposes as a callable stub anyway. Since `response`
has no RPC-facing setter, a plain method avoids that trap entirely -
same pattern XRayEye uses for its own read-only `active_roi`.
same pattern OMNY_XRayEye uses for its own read-only `active_roi`.
"""
return self._response
@@ -6,9 +6,9 @@ def main(): # pragma: no cover
return
from PySide6.QtDesigner import QPyDesignerCustomWidgetCollection
from csaxs_bec.bec_widgets.widgets.console_buttons.console_buttons_widget_plugin import ConsoleButtonsWidgetPlugin
from csaxs_bec.bec_widgets.widgets.console_buttons.z_console_buttons_widget_plugin import z_ConsoleButtonsWidgetPlugin
QPyDesignerCustomWidgetCollection.addCustomWidget(ConsoleButtonsWidgetPlugin())
QPyDesignerCustomWidgetCollection.addCustomWidget(z_ConsoleButtonsWidgetPlugin())
if __name__ == "__main__": # pragma: no cover
@@ -5,17 +5,17 @@ from qtpy.QtDesigner import QDesignerCustomWidgetInterface
from qtpy.QtWidgets import QWidget
from bec_widgets.utils.bec_designer import designer_material_icon
from csaxs_bec.bec_widgets.widgets.console_buttons.console_buttons import ConsoleButtonsWidget
from csaxs_bec.bec_widgets.widgets.console_buttons.console_buttons import z_ConsoleButtonsWidget
DOM_XML = """
<ui language='c++'>
<widget class='ConsoleButtonsWidget' name='console_buttons_widget'>
<widget class='z_ConsoleButtonsWidget' name='z_console_buttons_widget'>
</widget>
</ui>
"""
class ConsoleButtonsWidgetPlugin(QDesignerCustomWidgetInterface): # pragma: no cover
class z_ConsoleButtonsWidgetPlugin(QDesignerCustomWidgetInterface): # pragma: no cover
def __init__(self):
super().__init__()
self._form_editor = None
@@ -23,7 +23,7 @@ class ConsoleButtonsWidgetPlugin(QDesignerCustomWidgetInterface): # pragma: no
def createWidget(self, parent):
if parent is None:
return QWidget()
t = ConsoleButtonsWidget(parent)
t = z_ConsoleButtonsWidget(parent)
return t
def domXml(self):
@@ -33,10 +33,10 @@ class ConsoleButtonsWidgetPlugin(QDesignerCustomWidgetInterface): # pragma: no
return ""
def icon(self):
return designer_material_icon(ConsoleButtonsWidget.ICON_NAME)
return designer_material_icon(z_ConsoleButtonsWidget.ICON_NAME)
def includeFile(self):
return "console_buttons_widget"
return "z_console_buttons_widget"
def initialize(self, form_editor):
self._form_editor = form_editor
@@ -48,7 +48,7 @@ class ConsoleButtonsWidgetPlugin(QDesignerCustomWidgetInterface): # pragma: no
return self._form_editor is not None
def name(self):
return "ConsoleButtonsWidget"
return "z_ConsoleButtonsWidget"
def toolTip(self):
return ""
@@ -0,0 +1,266 @@
"""
Data Viewer: Custom BEC widget to view data from scans.
"""
import os
import subprocess
import sys
from datetime import datetime
from typing import Literal
from bec_lib import bec_logger
from bec_lib.endpoints import MessageEndpoints
from bec_widgets.utils.bec_dispatcher import BECDispatcher
from bec_widgets.utils.bec_widget import BECWidget
from bec_widgets.utils.colors import apply_theme, get_accent_colors
from bec_widgets.utils.error_popups import SafeSlot
# pylint: disable=E0611
from qtpy.QtWidgets import QApplication, QFileDialog, QMessageBox, QVBoxLayout, QWidget
from .panels.input_panel import InputPanel
from .panels.scan_view import ScanViewer
logger = bec_logger.logger
MAX_HIST_LEN = 100
class DataViewer(BECWidget, QWidget):
"""
Main widget of Data Viewer
"""
PLUGIN = True
ICON_NAME = "find_in_page"
def __init__(self, parent=None, **kwargs):
super().__init__(parent=parent, **kwargs)
self.get_bec_shortcuts()
central = QWidget()
self.root_layout = QVBoxLayout(central)
self.input = InputPanel()
self.viewer = ScanViewer()
self.root_layout.addWidget(self.input, 0)
self.root_layout.addWidget(self.viewer, 1)
self.setLayout(self.root_layout)
self.setWindowTitle("Data Viewer")
self.history = []
self.bec_dispatcher.connect_slot(self.on_history_update, MessageEndpoints.scan_history())
self.on_history_update()
self.current_row = 0
self.input.scan_sel.currentItemChanged_connect(self.scan_sel_changed)
self.input.load_button.clicked_connect(self.load_scan_from_history)
self.input.load_from_folder_button.clicked_connect(self.load_scan_from_folder)
self.viewer.unload_button.clicked_connect(self.unload_all_scans)
self.input.open_fm_button.clicked_connect(self.open_in_file_manager)
self.show_open_from_folder_infobox = True
def apply_theme(self, theme: Literal["dark", "light"]):
"""
Apply the theme
Args:
theme (str): Theme, either "dark" or "light"
"""
self.viewer.apply_theme(theme)
self.on_history_update()
@SafeSlot()
def scan_sel_changed(self, *_, **kwargs):
"""Updates the current row value of the scan selection list"""
self.current_row = kwargs["value"]().row()
@SafeSlot()
def open_in_file_manager(self, *_):
"""Open the scan folder in the systems default file manager"""
if len(self.history) > 0:
scan = self.history[self.current_row]
filepath = scan["file_components"][0].decode().rsplit("/", 1)[0]
subprocess.Popen(
["xdg-open", filepath],
stdout=subprocess.DEVNULL,
stderr=subprocess.DEVNULL,
stdin=subprocess.DEVNULL,
start_new_session=True,
)
@SafeSlot()
def load_scan_from_history(self, *_):
"""
Loads a scan. Find all files within the scan folder, sort them and
then load the files in the scan view
"""
if len(self.history) > 0:
scan = self.history[self.current_row]
base_filepath = scan["file_components"][0].decode().rsplit("/", 1)[0]
self.load_scan(base_filepath)
@SafeSlot()
def load_scan_from_folder(self, *_):
"""
Loads a scan from a folder. Find all files within the scan folder, sort them and
then load the files in the scan view
"""
hostname = self.client._hostname
start = hostname.find("x")
if start != -1:
beamline = hostname[start : start + 5]
active_account = self.client.active_account
start_folder = f"/sls/{beamline}/data/{active_account}/raw"
else:
start_folder = "/sls"
if self.show_open_from_folder_infobox:
self.show_open_from_folder_infobox = False
info_box = QMessageBox(self)
info_box.setIcon(QMessageBox.Icon.Information)
info_box.setWindowTitle("Select Scan Folder")
info_box.setText("Navigate to the scan <b>folder</b> and click 'Open' to load it.")
info_box.addButton("Got it!", QMessageBox.ButtonRole.AcceptRole)
info_box.exec_()
base_filepath = QFileDialog.getExistingDirectory(self, "Open Scan Folder", start_folder)
if base_filepath != "":
self.load_scan(base_filepath)
def load_scan(self, base_filepath):
"""
Loads a scan from a base_filepath. Find all files within the scan folder, sort them and
then load the files in the scan view
"""
filenames = [
f for f in os.listdir(base_filepath) if os.path.isfile(os.path.join(base_filepath, f))
]
def sort_priority(name):
if "master" in name:
return 0
if name.endswith(".h5"):
return 1
return 2
sorted_files = [f"{base_filepath}/{name}" for name in sorted(filenames, key=sort_priority)]
self.viewer.load_files(sorted_files)
@SafeSlot()
def unload_all_scans(self, *_):
"""Removes all scans from the scan view"""
self.viewer.clear_files()
def duration_formatted(self, start: str, end: str) -> str:
"""
Calculates the duration of a scan based on start end end time and
formats it as an easy readable string.
Args:
start(str): start time in iso-format
end(str): end time in iso-format
Returns:
str: Formatted duration, e.g. '1min 10s' or '1h 13min'
"""
start_dt = datetime.fromisoformat(start)
end_dt = datetime.fromisoformat(end)
seconds = abs(int((end_dt - start_dt).total_seconds()))
days, remainder = divmod(seconds, 86400)
hours, remainder = divmod(remainder, 3600)
minutes, seconds = divmod(remainder, 60)
parts = []
if days:
parts.append(f"{days}d")
if hours:
parts.append(f"{hours}h")
if minutes:
parts.append(f"{minutes}min")
if not days and not hours and minutes < 10:
parts.append(f"{seconds}s")
return " ".join(parts) if parts else "<1min"
def time_formatted(self, iso_time: str) -> str:
"""
Formates a time as an easy readable string.
Args:
iso_time(str): Time in iso-format
Returns:
str: Time formatted with format '%d.%m.%Y %H:%M', e.g. '14.01.1995 08:12'
"""
dt = datetime.fromisoformat(iso_time)
return dt.strftime("%d.%m.%Y %H:%M")
@SafeSlot()
def on_history_update(self, *_):
"""Updates the scan list based on the bec scan history."""
self.history = []
self.input.scan_sel.clear()
if self.client.history is None:
return
max_scans = min(len(self.client.history), MAX_HIST_LEN)
for n in range(1, max_scans): # last scans, limited by MAX_HIST_LEN
start_time = self.client.history[-n].metadata["start_time"] # type: ignore
end_time = self.client.history[-n].metadata["end_time"] # type: ignore
scan_data = self.client.history[-n].metadata["bec"] # type: ignore
# logger.info(f"scan_data: {scan_data}")
scan_number = scan_data["scan_number"]
scan_name = scan_data["scan_name"]
comment, sample_name = "", ""
user_metadata = scan_data.get("user_metadata", {})
comment = user_metadata.get("comment", comment)
sample_name = user_metadata.get("sample_name", sample_name)
if comment == "":
metadata_user = scan_data.get("metadata", {}).get("user_metadata", {})
comment = metadata_user.get("comment", comment)
if sample_name == "":
metadata_user = scan_data.get("metadata", {}).get("user_metadata", {})
sample_name = metadata_user.get("sample_name", sample_name)
status = scan_data["status"]
self.history.append(
{
"scan_number": scan_number,
"scan_name": scan_name,
"comment": comment,
"sample_name": sample_name,
"file_components": scan_data["file_components"],
"start_time": start_time,
"end_time": end_time,
"status": status,
}
)
tags = []
tags.append((scan_name, get_accent_colors().default.name()))
if sample_name != "":
tags.append((sample_name, get_accent_colors().highlight.name()))
if comment != "":
tags.append((comment, get_accent_colors().warning.name()))
if status == "closed":
tags.append((status, get_accent_colors().success.name()))
elif status == "halted" or status == "aborted":
tags.append((status, get_accent_colors().emergency.name()))
else:
tags.append((status, "#656365"))
tags.append((self.duration_formatted(start_time, end_time), "#656365"))
tags.append((self.time_formatted(start_time), "#656365"))
self.input.scan_sel.addTaggedItem(label=str(scan_number), tags=tags)
if __name__ == "__main__":
app = QApplication(sys.argv)
apply_theme("light")
dispatcher = BECDispatcher(gui_id="data_viewer")
win = DataViewer()
win.show()
sys.exit(app.exec_())
@@ -0,0 +1 @@
{'files': ['data_viewer.py']}
@@ -5,17 +5,17 @@ from qtpy.QtDesigner import QDesignerCustomWidgetInterface
from qtpy.QtWidgets import QWidget
from bec_widgets.utils.bec_designer import designer_material_icon
from csaxs_bec.bec_widgets.widgets.slit_control.slit_control import SlitControlWidget
from csaxs_bec.bec_widgets.widgets.data_viewer.data_viewer import DataViewer
DOM_XML = """
<ui language='c++'>
<widget class='SlitControlWidget' name='slit_control_widget'>
<widget class='DataViewer' name='data_viewer'>
</widget>
</ui>
"""
class SlitControlWidgetPlugin(QDesignerCustomWidgetInterface): # pragma: no cover
class DataViewerPlugin(QDesignerCustomWidgetInterface): # pragma: no cover
def __init__(self):
super().__init__()
self._form_editor = None
@@ -23,7 +23,7 @@ class SlitControlWidgetPlugin(QDesignerCustomWidgetInterface): # pragma: no cov
def createWidget(self, parent):
if parent is None:
return QWidget()
t = SlitControlWidget(parent)
t = DataViewer(parent)
return t
def domXml(self):
@@ -33,10 +33,10 @@ class SlitControlWidgetPlugin(QDesignerCustomWidgetInterface): # pragma: no cov
return ""
def icon(self):
return designer_material_icon(SlitControlWidget.ICON_NAME)
return designer_material_icon(DataViewer.ICON_NAME)
def includeFile(self):
return "slit_control_widget"
return "data_viewer"
def initialize(self, form_editor):
self._form_editor = form_editor
@@ -48,7 +48,7 @@ class SlitControlWidgetPlugin(QDesignerCustomWidgetInterface): # pragma: no cov
return self._form_editor is not None
def name(self):
return "SlitControlWidget"
return "DataViewer"
def toolTip(self):
return ""
@@ -0,0 +1,229 @@
"""
File-format loader (HDF5, images).
"""
import os
from abc import ABC, abstractmethod
from typing import Iterator, Literal, Optional
import h5py
import numpy as np
class NodeInfo:
"""Describes a single node (group or dataset) inside a loaded file."""
__slots__ = ("name", "path", "kind", "dtype", "shape")
def __init__(
self,
name: str,
path: str,
kind: Literal["group", "dataset"],
dtype: str = "",
shape: tuple[int, ...] = (),
):
self.name = name
self.path = path
self.kind = kind
self.dtype = dtype # e.g. "float", "int", "str", …
self.shape = shape # empty tuple for scalars / groups
class BaseFileLoader(ABC):
"""
Abstract base class for file-format loaders.
Subclass this to add support for a new format. Three things are required:
1. ``EXTENSIONS`` — tuple of lowercase extensions this loader handles,
e.g. ``(".h5", ".hdf5")``.
2. ``open(filepath)`` — open the file and keep any handles alive.
3. ``iter_nodes(path)`` — yield ``NodeInfo`` objects for the direct
children of *path* (depth-1 walk; the tree widget calls this
recursively as the user expands nodes).
4. ``read_dataset(path)`` — return the dataset at *path* as a
``numpy.ndarray``.
5. ``close()`` — release any open file handles.
6. ``child_count(path)`` — return the number of direct children of a
group node (used for the status label; override if cheap to compute).
"""
EXTENSIONS: tuple[str, ...] = ()
@abstractmethod
def open(self, filepath: str) -> None: ...
@abstractmethod
def iter_nodes(self, path: str) -> Iterator[NodeInfo]: ...
@abstractmethod
def read_dataset(self, path: str) -> np.ndarray: ...
@abstractmethod
def close(self) -> None: ...
def child_count(self, path: str) -> int:
return sum(1 for _ in self.iter_nodes(path))
class HDF5Loader(BaseFileLoader):
"""Loader for HDF5 / NeXus files (.h5, .hdf5, .nxs, .nx)."""
EXTENSIONS = (".h5", ".hdf5", ".hdf", ".nxs", ".nx")
def __init__(self):
self._file: Optional[h5py.File] = None
def open(self, filepath: str) -> None:
self._file = h5py.File(filepath, "r")
def close(self) -> None:
if self._file is not None:
self._file.close()
self._file = None
def iter_nodes(self, path: str) -> Iterator[NodeInfo]:
assert self._file is not None, "File not open"
obj = self._file[path] if path != "/" else self._file
if not isinstance(obj, h5py.Group):
return
for key in obj.keys():
try:
child = obj[key]
except Exception:
continue
child_path = child.name # h5py always gives the absolute path
if isinstance(child, h5py.Group):
yield NodeInfo(name=key, path=child_path, kind="group")
elif isinstance(child, h5py.Dataset):
shape_tuple = child.shape
d = child.dtype
dtype = "unknown"
if np.issubdtype(d, np.integer):
dtype = "int"
if np.issubdtype(d, np.floating):
dtype = "float"
if np.issubdtype(d, np.complexfloating):
dtype = "complex"
if d.kind in ("S", "U", "O"):
dtype = "str"
yield NodeInfo(
name=key, path=child_path, kind="dataset", dtype=dtype, shape=shape_tuple
)
def read_dataset(self, path: str) -> np.ndarray:
assert self._file is not None, "File not open"
return self._file[path][()]
def child_count(self, path: str) -> int:
assert self._file is not None, "File not open"
obj = self._file[path] if path != "/" else self._file
return len(obj) if isinstance(obj, h5py.Group) else 0
class ImageLoader(BaseFileLoader):
"""
Loader for raster image files.
The file is treated as a single, flat dataset. ``iter_nodes`` yields one
leaf node.
Requires: Pillow (``pip install Pillow``)
"""
EXTENSIONS = (
".jpg",
".jpeg",
".png",
".gif",
".tiff",
".tif",
".bmp",
".webp",
".ico",
".ppm",
".pgm",
".pbm",
)
def __init__(self):
self._filepath: Optional[str] = None
def open(self, filepath: str) -> None:
# Validate that Pillow can open it; keep only the path.
try:
from PIL import Image as _PILImage # noqa: F401 — existence check
_PILImage.open(filepath).verify()
except Exception as exc:
raise OSError(f"Cannot open image {filepath!r}: {exc}") from exc
self._filepath = filepath
def close(self) -> None:
self._filepath = None
def iter_nodes(self, path: str) -> Iterator[NodeInfo]:
"""Images have no internal hierarchy — yield a single leaf node."""
if path != "/" or self._filepath is None:
return
from PIL import Image as _PILImage
with _PILImage.open(self._filepath) as img:
w, h = img.size
mode = img.mode # e.g. "RGB", "RGBA", "L", …
name = os.path.basename(self._filepath)
yield NodeInfo(name=name, path="/image", kind="dataset", dtype=mode, shape=(h, w))
def read_dataset(self, path: str) -> np.ndarray:
"""Return the image as a uint8 numpy array (H x W x C or H x W)."""
from PIL import Image as _PILImage
with _PILImage.open(self._filepath) as img: # type: ignore[arg-type]
# Animated GIF → first frame only
if hasattr(img, "n_frames") and img.n_frames > 1:
img.seek(0)
return np.asarray(img)
def child_count(self, path: str) -> int:
return 1 if path == "/" else 0
class LoaderRegistry:
"""Maps file extensions to loader classes."""
def __init__(self):
self._registry: dict[str, type[BaseFileLoader]] = {}
def register(self, loader_cls: type[BaseFileLoader]) -> None:
"""Register a loader class for all extensions it declares."""
for ext in loader_cls.EXTENSIONS:
self._registry[ext.lower()] = loader_cls
def get_loader(self, filepath: str) -> Optional[BaseFileLoader]:
"""Return a fresh loader instance for *filepath*, or None if unsupported."""
ext = os.path.splitext(filepath)[1].lower()
cls = self._registry.get(ext)
return cls() if cls is not None else None
@property
def supported_extensions(self) -> list[str]:
return sorted(self._registry)
# Default global registry — pre-populated with built-in loaders.
registry = LoaderRegistry()
registry.register(HDF5Loader)
registry.register(ImageLoader)
@@ -0,0 +1,393 @@
"""
Data viewer displaying the data
"""
from typing import Literal, Optional
import numpy as np
import pyqtgraph as pg
from bec_lib import bec_logger
from bec_widgets.utils.colors import Colors
from qtpy.QtCore import Qt
# pylint: disable=E0611
from qtpy.QtGui import QFont, QPixmap
# pylint: disable=E0611
from qtpy.QtWidgets import (
QAbstractItemView,
QApplication,
QGroupBox,
QHBoxLayout,
QHeaderView,
QLabel,
QRadioButton,
QSizePolicy,
QSlider,
QTableWidget,
QTableWidgetItem,
QVBoxLayout,
QWidget,
)
logger = bec_logger.logger
MAX_ROWS = 2000
MAX_COLS = 500
class DataView(QWidget):
def __init__(self):
super().__init__()
self._layout = QVBoxLayout(self)
header = QHBoxLayout()
self.path_label = QLabel("")
self.path_label.setObjectName("path_label")
self.info_label = QLabel("")
self.info_label.setObjectName("info_label")
header.addWidget(self.path_label, 1)
header.addWidget(self.info_label)
self._layout.addLayout(header)
mode_box = QGroupBox("View mode")
mode_layout = QHBoxLayout(mode_box)
mode_layout.setContentsMargins(8, 4, 8, 4)
self.rb_auto = QRadioButton("Auto")
self.rb_plot = QRadioButton("Plot")
self.rb_image = QRadioButton("Image")
self.rb_table = QRadioButton("Table")
self.rb_auto.setChecked(True)
for rb in (self.rb_auto, self.rb_image, self.rb_plot, self.rb_table):
mode_layout.addWidget(rb)
rb.toggled.connect(self._on_mode_change)
mode_layout.addStretch()
self._layout.addWidget(mode_box)
self.content = QWidget()
self.content_layout = QVBoxLayout(self.content)
self.content_layout.setContentsMargins(0, 0, 0, 0)
self._layout.addWidget(self.content, 1)
self.plot_widget = None
self.image_widget = None
self._current_data = None
self.show_empty()
def apply_theme(self, theme: Optional[Literal["dark", "light"]] = None):
"""
Apply the theme
Args:
theme (Optional[str]): Theme, either "dark", "light", or None. Defaults to None.
"""
if theme is None:
app = QApplication.instance()
theme = app.theme.theme # type: ignore
bg_color = pg.getConfigOption("background")
fg_color = pg.getConfigOption("foreground")
if self.plot_widget is not None:
n_curves = len(self.plot_widget.listDataItems())
colors = Colors.golden_angle_color(
colormap="plasma", num=max(10, n_curves + 1), format="HEX"
)
for idx, curve in enumerate(self.plot_widget.listDataItems()):
curve.setPen(pg.mkPen(color=colors[idx]))
# Background
self.plot_widget.setBackground(bg_color)
# Axes (tick marks, tick labels, axis line)
for axis in ["left", "bottom", "right", "top"]:
ax = self.plot_widget.getAxis(axis)
ax.setPen(pg.mkPen(color=fg_color))
ax.setTextPen(pg.mkPen(color=fg_color))
if self.image_widget is not None:
self.image_widget.getView().setBackgroundColor(bg_color)
self.image_widget.ui.histogram.setBackground(bg_color)
def _clear_stack(self):
while self.content_layout.count():
item = self.content_layout.takeAt(0)
if item.widget():
item.widget().deleteLater()
self.plot_widget = None
self.image_widget = None
def show_empty(self):
"""Empties the content area."""
self._clear_stack()
empty_label = QLabel("No data selected")
empty_label.setObjectName("info_label")
empty_label.setAlignment(Qt.AlignmentFlag.AlignCenter)
self.path_label.setText("")
self.info_label.setText("")
self.content_layout.addWidget(empty_label)
def show_unsupported(self, path: str = "") -> None:
"""Display a friendly 'not implemented' message for unknown file types."""
self._clear_stack()
self._current_data = None
self.path_label.setText(path)
self.info_label.setText("")
lbl = QLabel("File type not supported")
lbl.setObjectName("info_label")
lbl.setAlignment(Qt.AlignmentFlag.AlignCenter)
self.content_layout.addWidget(lbl)
def _on_mode_change(self):
if self._current_data is not None:
self.display(self._current_data, self.path_label.text())
def _active_mode(self):
if self.rb_plot.isChecked():
return "plot"
if self.rb_image.isChecked():
return "image"
if self.rb_table.isChecked():
return "table"
return "auto"
def display(self, data, path: str = "") -> None:
"""
Render *data* in the panel.
Parameters
----------
data:
A ``numpy.ndarray`` (or anything convertible to one).
path:
Human-readable label shown in the header.
"""
self._current_data = data
self.path_label.setText(path)
if not isinstance(data, np.ndarray):
data = np.array(data)
self.info_label.setText(f"shape {data.shape}, dtype {data.dtype}, {data.size} elements")
mode = self._active_mode()
if mode == "auto":
if data.ndim <= 1 and data.size > 1:
mode = "plot"
elif data.ndim <= 4 and min(data.shape, default=0) > 1:
mode = "image"
else:
mode = "table"
if mode == "plot":
self._show_plot_1d(data)
elif mode == "image":
self._show_image_2d(data)
else:
self._show_table(data)
def _show_plot_1d(self, data):
self._clear_stack()
is_2d = data.ndim == 2
if is_2d:
n_rows, _ = data.shape
row_data = data[0].astype(np.float32)
else:
row_data = data.reshape(-1).astype(np.float32)
x = np.arange(row_data.size, dtype=np.float32)
self.plot_widget = pg.PlotWidget()
plot_item = self.plot_widget.getPlotItem()
assert plot_item is not None, "PlotWidget has no PlotItem"
plot_item.showGrid(x=True, y=True, alpha=0.25)
self.plot_widget.setSizePolicy(QSizePolicy.Policy.Expanding, QSizePolicy.Policy.Expanding)
plot_item.setAutoVisible(y=False) # type: ignore[attr-defined]
curve = pg.PlotDataItem(
x, row_data, pen=pg.mkPen(color="#2980b9", width=1.6), antialias=False
)
self.plot_widget.addItem(curve)
curve.setDownsampling(auto=True, method="peak")
curve.setClipToView(True)
curve.setSkipFiniteCheck(True)
plot_item.enableAutoRange() # type: ignore[attr-defined]
if is_2d:
slider = QSlider(Qt.Orientation.Vertical)
slider.setMinimum(0)
slider.setMaximum(n_rows - 1)
slider.setValue(0)
slider.setFixedWidth(32)
slider.setPageStep(1)
current_row_label = QLabel("0")
current_row_label.setAlignment(Qt.AlignmentFlag.AlignCenter)
current_row_label.setFixedWidth(32)
max_row_label = QLabel(f"0:{n_rows-1}")
max_row_label.setAlignment(Qt.AlignmentFlag.AlignCenter)
max_row_label.setFixedWidth(32)
def on_row_changed(row):
current_row_label.setText(str(row))
new_data = data[row].astype(np.float32)
new_x = np.arange(new_data.size, dtype=np.float32)
curve.setData(new_x, new_data)
plot_item.enableAutoRange() # type: ignore[attr-defined]
slider.valueChanged.connect(on_row_changed)
slider_col = QWidget()
slider_col.setFixedWidth(36)
col_layout = QVBoxLayout(slider_col)
col_layout.setContentsMargins(0, 0, 0, 0)
col_layout.setSpacing(2)
col_layout.addWidget(max_row_label)
col_layout.addWidget(slider)
col_layout.addWidget(current_row_label)
container = QWidget()
h_layout = QHBoxLayout(container)
h_layout.setContentsMargins(0, 0, 0, 0)
h_layout.setSpacing(4)
h_layout.addWidget(slider_col)
h_layout.addWidget(self.plot_widget)
self.content_layout.addWidget(container)
else:
self.content_layout.addWidget(self.plot_widget)
self.apply_theme()
def _show_image_2d(self, data):
self._clear_stack()
stacked = False
n_images = 0
rgb = False
img = data
if data.ndim == 3:
if data.shape[-1] in (3, 4):
rgb = True
else:
stacked = True
n_images = data.shape[0]
img = data[0, :]
elif data.ndim == 4:
if data.shape[-1] in (3, 4):
rgb = True
stacked = True
n_images = data.shape[0]
img = data[0, :]
self.image_widget = pg.ImageView()
self.image_widget.ui.roiBtn.hide()
self.image_widget.ui.menuBtn.hide()
if not rgb:
self.image_widget.setColorMap(pg.colormap.get("inferno", source="matplotlib"))
def set_image(img, autoLevels=True, autoHistogramRange=True):
if rgb:
self.image_widget.imageItem.setOpts(axisOrder="row-major")
self.image_widget.setImage(img)
else:
self.image_widget.setImage(
img.T, autoLevels=autoLevels, autoHistogramRange=autoHistogramRange
)
set_image(img)
self.image_widget.setSizePolicy(QSizePolicy.Policy.Expanding, QSizePolicy.Policy.Expanding)
if stacked:
slider = QSlider(Qt.Orientation.Vertical)
slider.setMinimum(0)
slider.setMaximum(n_images - 1)
slider.setValue(0)
slider.setFixedWidth(32)
slider.setPageStep(1)
current_image_label = QLabel("0")
current_image_label.setAlignment(Qt.AlignmentFlag.AlignCenter)
current_image_label.setFixedWidth(32)
max_image_label = QLabel(f"0:{n_images-1}")
max_image_label.setAlignment(Qt.AlignmentFlag.AlignCenter)
max_image_label.setFixedWidth(32)
def on_image_changed(row):
current_image_label.setText(str(row))
set_image(data[row, :], autoLevels=False, autoHistogramRange=False)
slider.valueChanged.connect(on_image_changed)
slider_col = QWidget()
slider_col.setFixedWidth(36)
col_layout = QVBoxLayout(slider_col)
col_layout.setContentsMargins(0, 0, 0, 0)
col_layout.setSpacing(2)
col_layout.addWidget(max_image_label)
col_layout.addWidget(slider)
col_layout.addWidget(current_image_label)
container = QWidget()
h_layout = QHBoxLayout(container)
h_layout.setContentsMargins(0, 0, 0, 0)
h_layout.setSpacing(4)
h_layout.addWidget(slider_col)
h_layout.addWidget(self.image_widget)
self.content_layout.addWidget(container)
else:
self.content_layout.addWidget(self.image_widget)
self.apply_theme()
def _show_table(self, data):
self._clear_stack()
if data.ndim == 0:
flat = data.reshape(1, 1)
elif data.ndim == 1:
flat = data.reshape(-1, 1)
elif data.ndim == 2:
flat = data
else:
flat = data.reshape(-1, data.shape[-1])
rows, cols = flat.shape
show_rows = min(rows, MAX_ROWS)
show_cols = min(cols, MAX_COLS)
if rows > MAX_ROWS or cols > MAX_COLS:
note = QLabel(f"⚠ Showing {show_rows}/{rows} rows x {show_cols}/{cols} columns")
note.setObjectName("info_label")
note.setAlignment(Qt.AlignmentFlag.AlignCenter)
self.content_layout.addWidget(note)
table = QTableWidget(show_rows, show_cols)
table.setEditTriggers(QAbstractItemView.EditTrigger.NoEditTriggers)
table.setSelectionMode(QAbstractItemView.SelectionMode.ContiguousSelection)
table.horizontalHeader().setSectionResizeMode(QHeaderView.ResizeMode.Interactive)
is_float = np.issubdtype(flat.dtype, np.floating)
is_complex = np.iscomplexobj(flat)
is_bytes = flat.dtype.kind == "S"
for r in range(show_rows):
for c in range(show_cols):
val = flat[r, c]
txt = (
f"{val:.6g}"
if is_float
else f"{val:.4g}" if is_complex else str(val.decode()) if is_bytes else str(val)
)
cell = QTableWidgetItem(txt)
cell.setTextAlignment(Qt.AlignmentFlag.AlignRight | Qt.AlignmentFlag.AlignVCenter)
table.setItem(r, c, cell)
self.content_layout.addWidget(table)
@@ -0,0 +1,40 @@
# pylint: disable=E0611
from qtpy.QtWidgets import QFrame, QHBoxLayout, QVBoxLayout, QWidget
# pylint: disable=E0402
from ..widgets.qt_widgets import Button, Group, ListWidget
class InputPanel(QWidget):
"""Panel for scan selection of the data viewer widget"""
def __init__(self, parent=None):
super().__init__(parent)
self._layout = QHBoxLayout(self)
# Scan selection
self.scan_sel = ListWidget("scan_sel")
self.load_button = Button(label_button="Load Dataset", enabled=True)
self.open_fm_button = Button(label_button="Open in File Manager", enabled=True)
line = QFrame()
line.setFrameShape(QFrame.HLine)
line.setFrameShadow(QFrame.Sunken)
self.load_from_folder_button = Button(
label_button="Load Dataset from File Manager", enabled=True
)
self._button_layout = QVBoxLayout()
self._button_layout.addWidget(self.load_button)
self._button_layout.addWidget(self.open_fm_button)
self._button_layout.addWidget(line)
self._button_layout.addWidget(self.load_from_folder_button)
self._button_layout.addStretch()
# Assemble complete scan selection group
self.input_group = Group(
"Scan selection", [self._button_layout, self.scan_sel], orientation="horizontal"
)
self._layout.addWidget(self.input_group)
@@ -0,0 +1,215 @@
"""
Scan viewer. Displays files of one or more scans in a tree view
"""
import os
from typing import Literal, Optional
from bec_lib import bec_logger
from bec_qthemes import material_icon
from bec_widgets.utils.colors import get_accent_colors
from qtpy.QtCore import Qt
# pylint: disable=E0611
from qtpy.QtGui import QBrush, QColor
# pylint: disable=E0611
from qtpy.QtWidgets import (
QHBoxLayout,
QHeaderView,
QMainWindow,
QSplitter,
QTreeWidget,
QTreeWidgetItem,
QVBoxLayout,
QWidget,
)
# pylint: disable=E0402
from ..loaders import BaseFileLoader, registry
from ..widgets.qt_widgets import Button, Group
from .data_view import DataView
logger = bec_logger.logger
ICON_SIZE = 20
class ScanViewer(QMainWindow):
"""
Generic scan viewer. Supports any format registered in *registry*.
Args:
filepath(str): Optional path to open on startup.
"""
def __init__(self, filepath: Optional[str] = None):
super().__init__()
self.registry = registry
self._open_files: dict[str, tuple[BaseFileLoader, str]] = {}
central = QWidget()
self.setCentralWidget(central)
root_layout = QHBoxLayout(central)
splitter = QSplitter(Qt.Orientation.Horizontal)
splitter.setChildrenCollapsible(False)
left_pane = QWidget()
left_layout = QVBoxLayout(left_pane)
self.unload_button = Button(label_button="Unload all", enabled=True)
self.tree = QTreeWidget()
self.tree.setMinimumWidth(250)
self.tree.setHeaderLabels(["Name", "Type", "Shape"])
self.tree.header().setStretchLastSection(False)
self.tree.header().setSectionResizeMode(0, QHeaderView.ResizeMode.Stretch)
self.tree.header().setSectionResizeMode(1, QHeaderView.ResizeMode.ResizeToContents)
self.tree.header().setSectionResizeMode(2, QHeaderView.ResizeMode.ResizeToContents)
self.tree.itemClicked.connect(self._on_item_clicked)
left_layout.addWidget(self.unload_button)
left_layout.addWidget(self.tree, 1)
self.data_panel = DataView()
splitter.addWidget(left_pane)
splitter.addWidget(self.data_panel)
splitter.setStretchFactor(0, 1)
splitter.setStretchFactor(1, 3)
splitter.setSizes([300, 900])
splitter.setHandleWidth(6)
splitter.setChildrenCollapsible(False)
self.scan_view_group = Group("Scan view", [splitter])
root_layout.addWidget(self.scan_view_group)
if filepath:
self.load_files([filepath])
def apply_theme(self, theme: Literal["dark", "light"]):
"""
Apply the theme
Args:
theme (str): Theme, either "dark" or "light"
"""
self.data_panel.apply_theme(theme)
def load_files(self, filepaths: list[str]) -> None:
"""Open one or more files and add each as a top-level tree node."""
for fp in filepaths:
if fp in self._open_files:
continue # already loaded
loader = self.registry.get_loader(fp)
if loader is None:
supported = ", ".join(self.registry.supported_extensions)
logger.warning("No loader found for %r. Supported extensions: %s", fp, supported)
continue
try:
loader.open(fp)
except Exception as exc:
logger.error("Failed to open %r: %s", fp, exc)
continue
display_name = os.path.basename(fp)
self._open_files[fp] = (loader, display_name)
self._add_file_to_tree(fp, loader, display_name)
def clear_files(self) -> None:
"""Close all open files and reset the tree."""
for loader, _ in self._open_files.values():
loader.close()
self._open_files.clear()
self.tree.clear()
self.data_panel.show_empty()
def closeEvent(self, event):
"""Close all"""
for loader, _ in self._open_files.values():
loader.close()
self._open_files.clear()
super().closeEvent(event)
def _add_file_to_tree(self, filepath: str, loader: BaseFileLoader, display_name: str) -> None:
"""Add a single file as a new top-level node in the tree."""
dataset_icon = material_icon(
"dataset", size=(ICON_SIZE, ICON_SIZE), color=get_accent_colors().default.name()
)
root_item = QTreeWidgetItem(self.tree, [display_name, "Group", ""])
root_item.setIcon(0, dataset_icon)
root_item.setData(0, Qt.ItemDataRole.UserRole, "/") # path
root_item.setData(0, Qt.ItemDataRole.UserRole + 1, "group") # kind
root_item.setData(0, Qt.ItemDataRole.UserRole + 2, filepath) # file key
self._populate_tree(root_item, loader, "/")
self.tree.addTopLevelItem(root_item)
# Expand first 2 levels by default
self.tree.expandItem(root_item)
for i in range(root_item.childCount()):
self.tree.expandItem(root_item.child(i))
self.tree.setCurrentItem(root_item)
def _populate_tree(
self, parent_item: QTreeWidgetItem, loader: BaseFileLoader, path: str
) -> None:
folder_icon = material_icon("folder", size=(ICON_SIZE, ICON_SIZE), color="#2980b9")
vector_icon = material_icon("show_chart", size=(ICON_SIZE, ICON_SIZE), color="#2980b9")
array_icon = material_icon(
"stacked_line_chart", size=(ICON_SIZE, ICON_SIZE), color="#2980b9"
)
scalar_icon = material_icon("point_scan", size=(ICON_SIZE, ICON_SIZE), color="#2980b9")
str_icon = material_icon("text_snippet", size=(ICON_SIZE, ICON_SIZE), color="#2980b9")
for node in loader.iter_nodes(path):
shape_str = "x".join(str(s) for s in node.shape) if node.shape else "scalar"
if node.kind == "group":
item = QTreeWidgetItem(parent_item, [node.name, "Group", ""])
item.setIcon(0, folder_icon)
item.setData(0, Qt.ItemDataRole.UserRole, node.path)
item.setData(0, Qt.ItemDataRole.UserRole + 1, "group")
self._populate_tree(item, loader, node.path)
else: # dataset
if shape_str == "scalar":
icon = str_icon if node.dtype == "str" else scalar_icon
elif "x" in shape_str:
icon = array_icon
else:
icon = vector_icon
item = QTreeWidgetItem(parent_item, [node.name, node.dtype, shape_str])
item.setIcon(0, icon)
item.setData(0, Qt.ItemDataRole.UserRole, node.path)
item.setData(0, Qt.ItemDataRole.UserRole + 1, "dataset")
item.setForeground(1, QBrush(QColor(get_accent_colors().success.name())))
item.setForeground(2, QBrush(QColor("#656365")))
def _get_filepath_for_item(self, item: QTreeWidgetItem) -> str:
"""Walk up the tree to find the filepath stored on the root node."""
node = item
while node.parent():
node = node.parent()
return node.data(0, Qt.ItemDataRole.UserRole + 2)
def _on_item_clicked(self, item: QTreeWidgetItem, _col: int) -> None:
path = item.data(0, Qt.ItemDataRole.UserRole)
kind = item.data(0, Qt.ItemDataRole.UserRole + 1)
filepath = self._get_filepath_for_item(item)
if not path or not filepath:
return
loader, _ = self._open_files[filepath]
if kind == "dataset":
data = loader.read_dataset(path)
self.data_panel.display(data, path)
@@ -6,9 +6,9 @@ def main(): # pragma: no cover
return
from PySide6.QtDesigner import QPyDesignerCustomWidgetCollection
from csaxs_bec.bec_widgets.widgets.tomo_params.tomo_params_plugin import TomoParamsPlugin
from csaxs_bec.bec_widgets.widgets.data_viewer.data_viewer_plugin import DataViewerPlugin
QPyDesignerCustomWidgetCollection.addCustomWidget(TomoParamsPlugin())
QPyDesignerCustomWidgetCollection.addCustomWidget(DataViewerPlugin())
if __name__ == "__main__": # pragma: no cover
@@ -0,0 +1,233 @@
"""
Universal Qt widgets
"""
from functools import partial
from bec_widgets.utils.colors import get_accent_colors
# pylint: disable=E0611
from qtpy.QtCore import QPoint, QRect, QSize, Qt
from qtpy.QtGui import QColor, QFont, QFontMetrics, QPainter, QPen
from qtpy.QtWidgets import (
QGroupBox,
QHBoxLayout,
QLabel,
QLayout,
QListWidget,
QListWidgetItem,
QPushButton,
QStyle,
QStyledItemDelegate,
QStyleOptionViewItem,
QVBoxLayout,
QWidget,
)
class Group(QGroupBox):
def __init__(self, label, objs, orientation="vertical"):
super().__init__(label)
if orientation == "vertical":
self._layout = QVBoxLayout(self)
elif orientation == "horizontal":
self._layout = QHBoxLayout(self)
else:
raise ValueError(f"Orientation {orientation} is not supported!")
for obj in objs:
if isinstance(obj, QWidget):
self._layout.addWidget(obj)
elif isinstance(obj, QLayout):
self._layout.addLayout(obj)
class Button(QWidget):
def __init__(self, label=None, label_button: str = "", enabled=False):
super().__init__()
layout = QHBoxLayout(self)
layout.setContentsMargins(0, 0, 0, 0)
layout.setSpacing(0)
if label is not None:
self.label = QLabel(label)
self.label.setFixedWidth(160)
layout.addWidget(self.label)
self.button = QPushButton(label_button)
self.enable_button(enabled)
layout.addWidget(self.button)
def clicked_connect(self, func):
"""Connect a function to the button press."""
self.button.clicked.connect(func)
def enable_button(self, enable: bool = False):
if enable:
self.button.setStyleSheet(
f"QPushButton {{background-color: {get_accent_colors().default.name()}; color: white;}}"
)
self.button.setEnabled(True)
else: # disabled
self.button.setStyleSheet(
"QPushButton {{background-color: rgb(120, 120, 120); color: white;}}"
)
self.button.setDisabled(True)
def setText(self, text):
self.button.setText(text)
class ListWidget(QWidget):
def __init__(self, identifier=""):
super().__init__()
layout = QHBoxLayout(self)
layout.setContentsMargins(10, 0, 0, 0)
layout.setSpacing(0)
self.identifier = identifier
self.value = TaggedListWidget()
layout.addWidget(self.value)
def clear(self):
self.value.clear()
def addTaggedItem(self, label, tags):
self.value.addTaggedItem(label, tags)
def setCurrentIndex(self, text):
self.value.setCurrentIndex(text)
def currentItemChanged_connect(self, func):
"""Connect a function to the Enter/Return key press."""
self.value.currentItemChanged.connect(
partial(
func,
identifier=self.identifier,
value_obj=self.value,
value=lambda: self.value.currentIndex(),
)
)
def setDisabled(self, disable):
self.value.setDisabled(disable)
class TaggedListWidget(QListWidget):
"""QListWidget with label + coloured tag pills per row."""
def __init__(self, parent=None) -> None:
super().__init__(parent)
self.setItemDelegate(TaggedDelegate(self))
self.setMouseTracking(True) # enables hover highlight
def addTaggedItem(self, label: str, tags: list | None = None) -> QListWidgetItem:
"""
Add a row. tags is a list of (text, hex_color) pairs, e.g.
[("v1.26", "#2563EB"), ("stable", "#16A34A")]
"""
item = QListWidgetItem(str(label))
if tags:
item.setData(_UserRole, list(tags))
self.addItem(item)
return item
def currentTags(self) -> list:
item = self.currentItem()
return item.data(_UserRole) or [] if item else []
ITEM_HEIGHT = 30
H_PAD = 12
TAG_H_PAD = 7
TAG_V_PAD = 3
TAG_GAP = 5
LABEL_TAG_GAP = 12
CORNER_RADIUS = 4
TAG_TEXT_COLOR = "#FFFFFF"
_DEMIBOLD = QFont.Weight.DemiBold
_MEDIUM = QFont.Weight.Medium
_AlignCenter = (
Qt.AlignmentFlag.AlignCenter if hasattr(Qt, "AlignmentFlag") else Qt.AlignmentFlag.AlignCenter
)
_UserRole = Qt.ItemDataRole.UserRole if hasattr(Qt, "ItemDataRole") else Qt.ItemDataRole.UserRole
_NoPen = Qt.PenStyle.NoPen if hasattr(Qt, "PenStyle") else Qt.PenStyle.NoPen
_AA = (
QPainter.RenderHint.Antialiasing
if hasattr(QPainter, "RenderHint")
else QPainter.RenderHint.Antialiasing
)
_State_Selected = QStyle.StateFlag.State_Selected
_State_MouseOver = QStyle.StateFlag.State_MouseOver
class TaggedDelegate(QStyledItemDelegate):
def sizeHint(self, option: QStyleOptionViewItem, index) -> QSize:
return QSize(option.rect.width() or 300, ITEM_HEIGHT)
def paint(self, painter: QPainter, option: QStyleOptionViewItem, index) -> None:
painter.save()
is_selected = bool(option.state & _State_Selected)
is_hover = bool(option.state & _State_MouseOver)
# Background
if is_selected:
painter.fillRect(option.rect, option.palette.highlight())
elif is_hover:
painter.fillRect(option.rect, QColor("#F1F5F9"))
else:
painter.fillRect(option.rect, option.palette.base())
label_text = (
index.data(
Qt.ItemDataRole.DisplayRole
if hasattr(Qt, "ItemDataRole")
else Qt.ItemDataRole.DisplayRole
)
or ""
)
tags: list = index.data(_UserRole) or []
# Label
label_font = painter.font() # inherit default font
label_font.setWeight(_DEMIBOLD)
painter.setFont(label_font)
label_color = (
option.palette.highlightedText().color()
if is_selected
else option.palette.text().color()
)
painter.setPen(QPen(label_color))
fm = QFontMetrics(label_font)
label_w = fm.horizontalAdvance(label_text)
label_y = option.rect.top() + (ITEM_HEIGHT - fm.height()) // 2 + fm.ascent()
painter.drawText(QPoint(option.rect.left() + H_PAD, label_y), label_text)
# Tag pills
tag_font = painter.font() # inherit default font
tag_font.setWeight(_MEDIUM)
fm_tag = QFontMetrics(tag_font)
x = option.rect.left() + H_PAD + label_w + LABEL_TAG_GAP
for tag_text, hex_color in tags:
tag_text = str(tag_text)
tw = fm_tag.horizontalAdvance(tag_text) + 2 * TAG_H_PAD
th = fm_tag.height() + 2 * TAG_V_PAD
ty = option.rect.top() + (ITEM_HEIGHT - th) // 2
pill = QRect(x, ty, tw, th)
fill = QColor(hex_color)
if is_selected:
fill = fill.lighter(140)
painter.setRenderHint(_AA)
painter.setPen(_NoPen)
painter.setBrush(fill)
painter.drawRoundedRect(pill, CORNER_RADIUS, CORNER_RADIUS)
painter.setFont(tag_font)
painter.setPen(QPen(QColor(TAG_TEXT_COLOR)))
painter.drawText(pill, _AlignCenter, tag_text)
x += tw + TAG_GAP
painter.restore()
@@ -5,31 +5,27 @@ from __future__ import annotations
# pylint: skip-file
designer_plugins = {
"ConsoleButtonsWidget": (
"csaxs_bec.bec_widgets.widgets.console_buttons.console_buttons",
"ConsoleButtonsWidget",
),
"SAXSWidget": ("csaxs_bec.bec_widgets.widgets.saxs_widget.saxs_widget", "SAXSWidget"),
"SampleStorageWidget": (
"DataViewer": ("csaxs_bec.bec_widgets.widgets.data_viewer.data_viewer", "DataViewer"),
"OMNY_SampleStorage": (
"csaxs_bec.bec_widgets.widgets.sample_storage.sample_storage",
"SampleStorageWidget",
"OMNY_SampleStorage",
),
"SlitControlWidget": (
"csaxs_bec.bec_widgets.widgets.slit_control.slit_control",
"SlitControlWidget",
"OMNY_TomoParams": ("csaxs_bec.bec_widgets.widgets.tomo_params.tomo_params", "OMNY_TomoParams"),
"OMNY_XRayEye": ("csaxs_bec.bec_widgets.widgets.xray_eye.x_ray_eye", "OMNY_XRayEye"),
"SAXSWidget": ("csaxs_bec.bec_widgets.widgets.saxs_widget.saxs_widget", "SAXSWidget"),
"SlitControl": ("csaxs_bec.bec_widgets.widgets.slit_control.slit_control", "SlitControl"),
"z_ConsoleButtonsWidget": (
"csaxs_bec.bec_widgets.widgets.console_buttons.console_buttons",
"z_ConsoleButtonsWidget",
),
"TomoParamsWidget": (
"csaxs_bec.bec_widgets.widgets.tomo_params.tomo_params",
"TomoParamsWidget",
),
"XRayEye": ("csaxs_bec.bec_widgets.widgets.xray_eye.x_ray_eye", "XRayEye"),
}
widget_icons = {
"ConsoleButtonsWidget": "widgets",
"DataViewer": "find_in_page",
"OMNY_SampleStorage": "widgets",
"OMNY_TomoParams": "widgets",
"OMNY_XRayEye": "widgets",
"SAXSWidget": "table_chart",
"SampleStorageWidget": "widgets",
"SlitControlWidget": "widgets",
"TomoParamsWidget": "widgets",
"XRayEye": "widgets",
"SlitControl": "widgets",
"z_ConsoleButtonsWidget": "widgets",
}
@@ -1,3 +1,3 @@
from .sample_storage import SampleStorageWidget
from .sample_storage import OMNY_SampleStorage
__all__ = ["SampleStorageWidget"]
__all__ = ["OMNY_SampleStorage"]
@@ -5,17 +5,17 @@ from qtpy.QtDesigner import QDesignerCustomWidgetInterface
from qtpy.QtWidgets import QWidget
from bec_widgets.utils.bec_designer import designer_material_icon
from csaxs_bec.bec_widgets.widgets.sample_storage.sample_storage import SampleStorageWidget
from csaxs_bec.bec_widgets.widgets.sample_storage.sample_storage import OMNY_SampleStorage
DOM_XML = """
<ui language='c++'>
<widget class='SampleStorageWidget' name='sample_storage_widget'>
<widget class='OMNY_SampleStorage' name='omny_sample_storage'>
</widget>
</ui>
"""
class SampleStorageWidgetPlugin(QDesignerCustomWidgetInterface): # pragma: no cover
class OMNY_SampleStoragePlugin(QDesignerCustomWidgetInterface): # pragma: no cover
def __init__(self):
super().__init__()
self._form_editor = None
@@ -23,7 +23,7 @@ class SampleStorageWidgetPlugin(QDesignerCustomWidgetInterface): # pragma: no c
def createWidget(self, parent):
if parent is None:
return QWidget()
t = SampleStorageWidget(parent)
t = OMNY_SampleStorage(parent)
return t
def domXml(self):
@@ -33,10 +33,10 @@ class SampleStorageWidgetPlugin(QDesignerCustomWidgetInterface): # pragma: no c
return ""
def icon(self):
return designer_material_icon(SampleStorageWidget.ICON_NAME)
return designer_material_icon(OMNY_SampleStorage.ICON_NAME)
def includeFile(self):
return "sample_storage_widget"
return "omny_sample_storage"
def initialize(self, form_editor):
self._form_editor = form_editor
@@ -48,7 +48,7 @@ class SampleStorageWidgetPlugin(QDesignerCustomWidgetInterface): # pragma: no c
return self._form_editor is not None
def name(self):
return "SampleStorageWidget"
return "OMNY_SampleStorage"
def toolTip(self):
return ""
@@ -6,9 +6,9 @@ def main(): # pragma: no cover
return
from PySide6.QtDesigner import QPyDesignerCustomWidgetCollection
from csaxs_bec.bec_widgets.widgets.slit_control.slit_control_widget_plugin import SlitControlWidgetPlugin
from csaxs_bec.bec_widgets.widgets.sample_storage.omny_sample_storage_plugin import OMNY_SampleStoragePlugin
QPyDesignerCustomWidgetCollection.addCustomWidget(SlitControlWidgetPlugin())
QPyDesignerCustomWidgetCollection.addCustomWidget(OMNY_SampleStoragePlugin())
if __name__ == "__main__": # pragma: no cover
@@ -1,18 +0,0 @@
def main(): # pragma: no cover
"""Register SampleStorageWidget with the Qt Designer.
Mirrors register_tomo_params.py — invoked by the Qt Designer's plugin
discovery mechanism.
"""
from qtpy import QtDesigner
from csaxs_bec.bec_widgets.widgets.sample_storage.sample_storage_plugin import (
SampleStorageWidgetPlugin,
)
QPyDesignerCustomWidgetCollection = QtDesigner.QPyDesignerCustomWidgetCollection
QPyDesignerCustomWidgetCollection.addCustomWidget(SampleStorageWidgetPlugin())
if __name__ == "__main__": # pragma: no cover
main()
@@ -1,15 +0,0 @@
def main(): # pragma: no cover
from qtpy import PYSIDE6
if not PYSIDE6:
print("PYSIDE6 is not available in the environment. Cannot patch designer.")
return
from PySide6.QtDesigner import QPyDesignerCustomWidgetCollection
from csaxs_bec.bec_widgets.widgets.sample_storage.sample_storage_widget_plugin import SampleStorageWidgetPlugin
QPyDesignerCustomWidgetCollection.addCustomWidget(SampleStorageWidgetPlugin())
if __name__ == "__main__": # pragma: no cover
main()
@@ -1,5 +1,5 @@
"""
SampleStorageWidget – BEC widget for viewing and correcting the FlOMNI
OMNY_SampleStorage – BEC widget for viewing and correcting the FlOMNI
sample-storage bookkeeping.
This widget is a GUI replacement for the ``ftransfer_show`` /
@@ -103,7 +103,7 @@ class _SlotCell(QFrame):
parent widget's mutation methods, which own all backend access.
"""
def __init__(self, slot: int, owner: "SampleStorageWidget", border_color: str):
def __init__(self, slot: int, owner: "OMNY_SampleStorage", border_color: str):
super().__init__()
self._slot = slot
self._owner = owner
@@ -186,7 +186,7 @@ class _SlotCell(QFrame):
# ── main widget ──────────────────────────────────────────────────────────────
class SampleStorageWidget(BECWidget, QWidget):
class OMNY_SampleStorage(BECWidget, QWidget):
"""
View and correct the FlOMNI sample-storage records.
@@ -267,7 +267,7 @@ class SampleStorageWidget(BECWidget, QWidget):
try:
data = self._samples.read(cached=True)
except Exception as exc:
logger.warning(f"SampleStorageWidget: bulk read failed: {exc}")
logger.warning(f"OMNY_SampleStorage: bulk read failed: {exc}")
# keep whatever is currently displayed rather than blanking out
return {}
@@ -309,7 +309,7 @@ class SampleStorageWidget(BECWidget, QWidget):
self._last_state.pop(slot, None)
return True
except Exception as exc:
logger.warning(f"SampleStorageWidget: write of slot {slot} failed: {exc}")
logger.warning(f"OMNY_SampleStorage: write of slot {slot} failed: {exc}")
QMessageBox.critical(
self,
"Write failed",
@@ -1,53 +0,0 @@
# pylint: disable=missing-module-docstring, missing-function-docstring
# This file is auto-generated style boilerplate for the Qt Designer plugin,
# mirroring the pattern used by the other csaxs_bec widget plugins
# (e.g. tomo_params_plugin.py). It registers SampleStorageWidget with the
# Qt Designer so it can be dropped onto a form.
from qtpy.QtDesigner import QDesignerCustomWidgetInterface
from csaxs_bec.bec_widgets.widgets.sample_storage.sample_storage import SampleStorageWidget
DOM_XML = """
<ui language='c++'>
<widget class='SampleStorageWidget' name='sample_storage'>
</widget>
</ui>
"""
class SampleStorageWidgetPlugin(QDesignerCustomWidgetInterface): # pragma: no cover
def __init__(self):
super().__init__()
self._form_editor = None
def initialize(self, form_editor):
self._form_editor = form_editor
def isInitialized(self):
return self._form_editor is not None
def createWidget(self, parent):
t = SampleStorageWidget(parent)
return t
def name(self):
return "SampleStorageWidget"
def group(self):
return "cSAXS FlOMNI"
def toolTip(self):
return "View and correct the FlOMNI sample-storage records (bookkeeping only)."
def whatsThis(self):
return self.toolTip()
def isContainer(self):
return False
def domXml(self):
return DOM_XML
def includeFile(self):
return "csaxs_bec.bec_widgets.widgets.sample_storage.sample_storage"
@@ -1,3 +1,3 @@
from csaxs_bec.bec_widgets.widgets.slit_control.slit_control import SlitControlWidget
from csaxs_bec.bec_widgets.widgets.slit_control.slit_control import SlitControl
__all__ = ["SlitControlWidget"]
__all__ = ["SlitControl"]
@@ -71,7 +71,7 @@ _KB_STEP = [
]
class SlitControlWidget(BECWidget, QWidget):
class SlitControl(BECWidget, QWidget):
"""
Interactive GUI for cSAXS slit center and size control.
@@ -505,7 +505,7 @@ class SlitControlWidget(BECWidget, QWidget):
device.move(target)
except Exception as exc:
msg = str(exc) or type(exc).__name__
logger.warning(f"SlitControlWidget: absolute move {dev_name} failed: {msg}")
logger.warning(f"SlitControl: absolute move {dev_name} failed: {msg}")
self._show_status(f"{dev_name}: {msg}")
def _stop_slit(self) -> None:
@@ -516,7 +516,7 @@ class SlitControlWidget(BECWidget, QWidget):
device.stop()
except Exception as exc:
msg = str(exc) or type(exc).__name__
logger.warning(f"SlitControlWidget: stop {self._dev_name(suffix)} failed: {msg}")
logger.warning(f"SlitControl: stop {self._dev_name(suffix)} failed: {msg}")
self._show_status(f"Stop failed: {msg}")
def _show_status(self, msg: str, error: bool = True) -> None:
@@ -658,7 +658,7 @@ class SlitControlWidget(BECWidget, QWidget):
device.move(current + delta)
except Exception as exc:
msg = str(exc) or type(exc).__name__
logger.warning(f"SlitControlWidget: failed to move {dev_name}: {msg}")
logger.warning(f"SlitControl: failed to move {dev_name}: {msg}")
self._show_status(f"{dev_name}: {msg}")
def set_step(self, value: float):
@@ -1,15 +1,15 @@
# Copyright (C) 2022 The Qt Company Ltd.
# SPDX-License-Identifier: LicenseRef-Qt-Commercial OR BSD-3-Clause
from bec_widgets.utils.bec_designer import designer_material_icon
from qtpy.QtDesigner import QDesignerCustomWidgetInterface
from qtpy.QtWidgets import QWidget
from csaxs_bec.bec_widgets.widgets.slit_control.slit_control import SlitControlWidget
from bec_widgets.utils.bec_designer import designer_material_icon
from csaxs_bec.bec_widgets.widgets.slit_control.slit_control import SlitControl
DOM_XML = """
<ui language='c++'>
<widget class='SlitControlWidget' name='slit_control'>
<widget class='SlitControl' name='slit_control'>
</widget>
</ui>
"""
@@ -23,7 +23,8 @@ class SlitControlPlugin(QDesignerCustomWidgetInterface): # pragma: no cover
def createWidget(self, parent):
if parent is None:
return QWidget()
return SlitControlWidget(parent)
t = SlitControl(parent)
return t
def domXml(self):
return DOM_XML
@@ -32,7 +33,7 @@ class SlitControlPlugin(QDesignerCustomWidgetInterface): # pragma: no cover
return ""
def icon(self):
return designer_material_icon(SlitControlWidget.ICON_NAME)
return designer_material_icon(SlitControl.ICON_NAME)
def includeFile(self):
return "slit_control"
@@ -47,10 +48,10 @@ class SlitControlPlugin(QDesignerCustomWidgetInterface): # pragma: no cover
return self._form_editor is not None
def name(self):
return "SlitControlWidget"
return "SlitControl"
def toolTip(self):
return "SlitControlWidget"
return ""
def whatsThis(self):
return self.toolTip()
@@ -1,3 +1,3 @@
from csaxs_bec.bec_widgets.widgets.tomo_params.tomo_params import TomoParamsWidget
from csaxs_bec.bec_widgets.widgets.tomo_params.tomo_params import OMNY_TomoParams
__all__ = ["TomoParamsWidget"]
__all__ = ["OMNY_TomoParams"]
@@ -5,17 +5,17 @@ from qtpy.QtDesigner import QDesignerCustomWidgetInterface
from qtpy.QtWidgets import QWidget
from bec_widgets.utils.bec_designer import designer_material_icon
from csaxs_bec.bec_widgets.widgets.tomo_params.tomo_params import TomoParamsWidget
from csaxs_bec.bec_widgets.widgets.tomo_params.tomo_params import OMNY_TomoParams
DOM_XML = """
<ui language='c++'>
<widget class='TomoParamsWidget' name='tomo_params_widget'>
<widget class='OMNY_TomoParams' name='omny_tomo_params'>
</widget>
</ui>
"""
class TomoParamsWidgetPlugin(QDesignerCustomWidgetInterface): # pragma: no cover
class OMNY_TomoParamsPlugin(QDesignerCustomWidgetInterface): # pragma: no cover
def __init__(self):
super().__init__()
self._form_editor = None
@@ -23,7 +23,7 @@ class TomoParamsWidgetPlugin(QDesignerCustomWidgetInterface): # pragma: no cove
def createWidget(self, parent):
if parent is None:
return QWidget()
t = TomoParamsWidget(parent)
t = OMNY_TomoParams(parent)
return t
def domXml(self):
@@ -33,10 +33,10 @@ class TomoParamsWidgetPlugin(QDesignerCustomWidgetInterface): # pragma: no cove
return ""
def icon(self):
return designer_material_icon(TomoParamsWidget.ICON_NAME)
return designer_material_icon(OMNY_TomoParams.ICON_NAME)
def includeFile(self):
return "tomo_params_widget"
return "omny_tomo_params"
def initialize(self, form_editor):
self._form_editor = form_editor
@@ -48,7 +48,7 @@ class TomoParamsWidgetPlugin(QDesignerCustomWidgetInterface): # pragma: no cove
return self._form_editor is not None
def name(self):
return "TomoParamsWidget"
return "OMNY_TomoParams"
def toolTip(self):
return ""
@@ -6,9 +6,9 @@ def main(): # pragma: no cover
return
from PySide6.QtDesigner import QPyDesignerCustomWidgetCollection
from csaxs_bec.bec_widgets.widgets.tomo_params.tomo_params_widget_plugin import TomoParamsWidgetPlugin
from csaxs_bec.bec_widgets.widgets.tomo_params.omny_tomo_params_plugin import OMNY_TomoParamsPlugin
QPyDesignerCustomWidgetCollection.addCustomWidget(TomoParamsWidgetPlugin())
QPyDesignerCustomWidgetCollection.addCustomWidget(OMNY_TomoParamsPlugin())
if __name__ == "__main__": # pragma: no cover
@@ -1,5 +1,5 @@
"""
TomoParamsWidget – BEC widget for editing tomo scan parameters and managing
OMNY_TomoParams – BEC widget for editing tomo scan parameters and managing
the tomo scan queue, for whichever cSAXS tomography setup (flomni, lamni, ...)
is active in the current BEC session -- see ``SETUP_PROFILES``.
@@ -239,7 +239,7 @@ def _hline() -> QFrame:
# ── main widget ──────────────────────────────────────────────────────────────
class TomoParamsWidget(BECWidget, QWidget):
class OMNY_TomoParams(BECWidget, QWidget):
"""
Interactive GUI for tomo scan parameter editing and queue management, for
whichever setup (see ``SETUP_PROFILES``) is active in this BEC session.
@@ -303,7 +303,7 @@ class TomoParamsWidget(BECWidget, QWidget):
self._on_queue_status, MessageEndpoints.scan_queue_status()
)
except Exception as exc:
logger.warning(f"TomoParamsWidget: could not subscribe to scan queue: {exc}")
logger.warning(f"OMNY_TomoParams: could not subscribe to scan queue: {exc}")
self._refresh_busy_banner()
# ── setup detection ──────────────────────────────────────────────────────
@@ -330,7 +330,7 @@ class TomoParamsWidget(BECWidget, QWidget):
try:
return self.client.get_global_var(key)
except Exception as exc:
logger.warning(f"TomoParamsWidget: get_global_var({key!r}) failed: {exc}")
logger.warning(f"OMNY_TomoParams: get_global_var({key!r}) failed: {exc}")
return None
def _gv_set(self, key: str, value: Any) -> bool:
@@ -338,7 +338,7 @@ class TomoParamsWidget(BECWidget, QWidget):
self.client.set_global_var(key, value)
return True
except Exception as exc:
logger.warning(f"TomoParamsWidget: set_global_var({key!r}) failed: {exc}")
logger.warning(f"OMNY_TomoParams: set_global_var({key!r}) failed: {exc}")
return False
def _load_params(self) -> dict[str, Any]:
@@ -1001,7 +1001,7 @@ class TomoParamsWidget(BECWidget, QWidget):
return False
return not all(item.status in _QUEUE_IDLE_STATUSES for item in queue_info)
except Exception as exc:
logger.warning(f"TomoParamsWidget: scan-queue busy check failed: {exc}")
logger.warning(f"OMNY_TomoParams: scan-queue busy check failed: {exc}")
return False
def _beamline_busy_reason(self) -> Optional[str]:
@@ -1339,7 +1339,7 @@ class TomoQueueDialog(QDialog):
"""
Non-modal popup window for managing the tomo scan queue.
Opened from the main TomoParamsWidget via the "Queue control…" button,
Opened from the main OMNY_TomoParams via the "Queue control…" button,
which passes its own detected ``profile`` (see ``SETUP_PROFILES``) down
so this dialog's "Add current params to queue" reads the right param
names/defaults for whichever setup is active -- it never detects the
@@ -1350,7 +1350,7 @@ class TomoQueueDialog(QDialog):
_POLL_MS = 2000
# Same heartbeat-staleness window as TomoParamsWidget._is_tomo_running().
# Same heartbeat-staleness window as OMNY_TomoParams._is_tomo_running().
# A "running" job whose tomo_progress heartbeat is older than this (or
# missing) means the process that was executing it died -- crash,
# Ctrl-C, kernel restart -- before it had a chance to update the job's
@@ -1712,7 +1712,7 @@ class TomoQueueDialog(QDialog):
"""True if there's no evidence a scan is genuinely active right now
-- tomo_progress's heartbeat is missing or older than
_RUNNING_STALE_AFTER_S. Same signal/window as
TomoParamsWidget._is_tomo_running()."""
OMNY_TomoParams._is_tomo_running()."""
prog = self._gv_get("tomo_progress")
if not isinstance(prog, dict):
return True
@@ -1,53 +0,0 @@
from bec_widgets.utils.bec_designer import designer_material_icon
from qtpy.QtDesigner import QDesignerCustomWidgetInterface
from qtpy.QtWidgets import QWidget
from csaxs_bec.bec_widgets.widgets.tomo_params.tomo_params import TomoParamsWidget
DOM_XML = """
<ui language='c++'>
<widget class='TomoParamsWidget' name='tomo_params'>
</widget>
</ui>
"""
class TomoParamsPlugin(QDesignerCustomWidgetInterface): # pragma: no cover
def __init__(self):
super().__init__()
self._form_editor = None
def createWidget(self, parent):
if parent is None:
return QWidget()
return TomoParamsWidget(parent)
def domXml(self):
return DOM_XML
def group(self):
return ""
def icon(self):
return designer_material_icon(TomoParamsWidget.ICON_NAME)
def includeFile(self):
return "tomo_params"
def initialize(self, form_editor):
self._form_editor = form_editor
def isContainer(self):
return False
def isInitialized(self):
return self._form_editor is not None
def name(self):
return "TomoParamsWidget"
def toolTip(self):
return "TomoParamsWidget"
def whatsThis(self):
return self.toolTip()
@@ -1,29 +1,29 @@
# Copyright (C) 2022 The Qt Company Ltd.
# SPDX-License-Identifier: LicenseRef-Qt-Commercial OR BSD-3-Clause
from bec_widgets.utils.bec_designer import designer_material_icon
from qtpy.QtDesigner import QDesignerCustomWidgetInterface
from qtpy.QtWidgets import QWidget
from csaxs_bec.bec_widgets.widgets.xray_eye.x_ray_eye import XRayEye
from bec_widgets.utils.bec_designer import designer_material_icon
from csaxs_bec.bec_widgets.widgets.xray_eye.x_ray_eye import OMNY_XRayEye
DOM_XML = """
<ui language='c++'>
<widget class='XRayEye' name='x_ray_eye'>
<widget class='OMNY_XRayEye' name='omny_x_ray_eye'>
</widget>
</ui>
"""
class XRayEyePlugin(QDesignerCustomWidgetInterface): # pragma: no cover
class OMNY_XRayEyePlugin(QDesignerCustomWidgetInterface): # pragma: no cover
def __init__(self):
super().__init__()
self._form_editor = None
def createWidget(self, parent):
if parent is None:
return QWidget()
t = XRayEye(parent)
return QWidget()
t = OMNY_XRayEye(parent)
return t
def domXml(self):
@@ -33,10 +33,10 @@ class XRayEyePlugin(QDesignerCustomWidgetInterface): # pragma: no cover
return ""
def icon(self):
return designer_material_icon(XRayEye.ICON_NAME)
return designer_material_icon(OMNY_XRayEye.ICON_NAME)
def includeFile(self):
return "x_ray_eye"
return "omny_x_ray_eye"
def initialize(self, form_editor):
self._form_editor = form_editor
@@ -48,10 +48,10 @@ class XRayEyePlugin(QDesignerCustomWidgetInterface): # pragma: no cover
return self._form_editor is not None
def name(self):
return "XRayEye"
return "OMNY_XRayEye"
def toolTip(self):
return "XRayEye"
return "OMNY_XRayEye"
def whatsThis(self):
return self.toolTip()
@@ -6,9 +6,9 @@ def main(): # pragma: no cover
return
from PySide6.QtDesigner import QPyDesignerCustomWidgetCollection
from csaxs_bec.bec_widgets.widgets.xray_eye.x_ray_eye_plugin import XRayEyePlugin
from csaxs_bec.bec_widgets.widgets.xray_eye.omny_x_ray_eye_plugin import OMNY_XRayEyePlugin
QPyDesignerCustomWidgetCollection.addCustomWidget(XRayEyePlugin())
QPyDesignerCustomWidgetCollection.addCustomWidget(OMNY_XRayEyePlugin())
if __name__ == "__main__": # pragma: no cover
@@ -46,7 +46,7 @@ class TargetCrosshair:
view while the user works through subsequent steps.
Position is expressed in the same image/data coordinate system used by
the widget's ROIs (see ``XRayEye.submit``), so values read back from
the widget's ROIs (see ``OMNY_XRayEye.submit``), so values read back from
``omny_xray_gui.xval_x_*`` / ``yval_y_*`` can be passed in directly.
"""
@@ -89,7 +89,7 @@ class ImageZoomControl(QWidget):
Mouse-wheel zoom steps can be far too coarse to use precisely over a
remote desktop connection, so this provides explicit zoom in/out buttons
plus a "fit to view" button that resets pan/zoom to frame the current
image (the same operation as ``XRayEye.reset_zoom()``).
image (the same operation as ``OMNY_XRayEye.reset_zoom()``).
"""
# scaleBy() factor for one "zoom in" click; >1 (its inverse) zooms out.
@@ -231,7 +231,7 @@ class XRayEye2DControl(BECWidget, QWidget):
logger.warning(f"Unknown direction {direction} for move command.")
class XRayEye(BECWidget, QWidget):
class OMNY_XRayEye(BECWidget, QWidget):
USER_ACCESS = [
"active_roi",
"user_message",
@@ -683,7 +683,7 @@ class XRayEye(BECWidget, QWidget):
try:
msg = self.client.connector.get(MessageEndpoints.scan_queue_status())
except Exception as exc:
logger.warning(f"Failed to fetch initial scan queue status for XRayEye: {exc}")
logger.warning(f"Failed to fetch initial scan queue status for OMNY_XRayEye: {exc}")
return
if msg is None:
@@ -957,7 +957,7 @@ class XRayEye(BECWidget, QWidget):
for signal in signals:
signal.connect(self.on_dap_params)
self._dap_params_forwarding_connected = True
logger.info("Enabled XRayEye DAP parameter forwarding.")
logger.info("Enabled OMNY_XRayEye DAP parameter forwarding.")
return
for signal in signals:
@@ -966,7 +966,7 @@ class XRayEye(BECWidget, QWidget):
except (TypeError, RuntimeError):
pass
self._dap_params_forwarding_connected = False
logger.info("Disabled XRayEye DAP parameter forwarding.")
logger.info("Disabled OMNY_XRayEye DAP parameter forwarding.")
@SafeSlot(dict, dict)
def on_dap_params(self, data, meta):
@@ -1070,7 +1070,7 @@ if __name__ == "__main__":
app = QApplication(sys.argv)
apply_theme("light")
dispatcher = BECDispatcher(gui_id="xray")
win = XRayEye()
win = OMNY_XRayEye()
win.resize(1000, 800)
win.show()