bec/superxas_bec
0
0
Fork 0
Code Issues 2 Pull Requests 1 Actions Packages Projects Releases Wiki Activity

update widget based on debye_bec
CI for superxas_bec / test (push) Failing after 1m16s
Details
CI for superxas_bec / test (pull_request) Successful in 1m18s
Details

Browse Source
This commit is contained in:
x10da
2026-08-04 12:40:42 +02:00
parent e4ca6a6880
commit 622e7635a4
4 changed files with 270 additions and 230 deletions
Download Patch File Download Diff File Expand all files Collapse all files
superxas_bec/bec_widgets/widgets/data_viewer/data_viewer.py
+254 -222
View File
@@ -1,222 +1,254 @@
"""
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, 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, *arg, parent=None, **kwargs):
super().__init__(parent=parent, *arg, **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)
self.input.unload_button.clicked_connect(self.unload_all_scans)
self.input.open_fm_button.clicked_connect(self.open_in_file_manager)
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(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]
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
scan_number = scan_data["scan_number"]
scan_name = scan_data["scan_name"]
if "metadata" in scan_data:
comment = scan_data["metadata"]["user_metadata"]["comment"]
sample_name = scan_data["metadata"]["user_metadata"]["sample_name"]
else:
comment, 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_())
"""
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, 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, *arg, parent=None, **kwargs):
super().__init__(parent=parent, *arg, **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)
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"
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_())
superxas_bec/bec_widgets/widgets/data_viewer/panels/input_panel.py
+11 -3
View File
@@ -1,5 +1,5 @@
# pylint: disable=E0611
from qtpy.QtWidgets import QHBoxLayout, QVBoxLayout, QWidget
from qtpy.QtWidgets import QFrame, QHBoxLayout, QVBoxLayout, QWidget
# pylint: disable=E0402
from ..widgets.qt_widgets import Button, Group, ListWidget
@@ -15,13 +15,21 @@ class InputPanel(QWidget):
# Scan selection
self.scan_sel = ListWidget("scan_sel")
self.load_button = Button(label_button="Load Dataset", enabled=True)
self.unload_button = Button(label_button="Unload all", 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.unload_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
superxas_bec/bec_widgets/widgets/data_viewer/panels/scan_view.py
+3 -1
View File
@@ -27,7 +27,7 @@ from qtpy.QtWidgets import (
# pylint: disable=E0402
from ..loaders import BaseFileLoader, registry
from ..widgets.qt_widgets import Group
from ..widgets.qt_widgets import Button, Group
from .data_view import DataView
logger = bec_logger.logger
@@ -59,6 +59,7 @@ class ScanViewer(QMainWindow):
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"])
@@ -68,6 +69,7 @@ class ScanViewer(QMainWindow):
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()
superxas_bec/bec_widgets/widgets/data_viewer/widgets/qt_widgets.py
+2 -4
View File
@@ -45,15 +45,13 @@ class Button(QWidget):
def __init__(self, label=None, label_button: str = "", enabled=False):
super().__init__()
layout = QHBoxLayout(self)
layout.setContentsMargins(10, 0, 0, 0)
layout.setContentsMargins(0, 0, 0, 0)
layout.setSpacing(0)
if label is not None:
self.label = QLabel(label)
self.label.setFixedWidth(140)
self.label.setFixedWidth(160)
layout.addWidget(self.label)
self.button = QPushButton(label_button)
if label is not None:
self.button.setFixedWidth(160)
self.enable_button(enabled)
layout.addWidget(self.button)