diff --git a/superxas_bec/bec_widgets/widgets/data_viewer/data_viewer.py b/superxas_bec/bec_widgets/widgets/data_viewer/data_viewer.py index 316d7cb..9dc185e 100644 --- a/superxas_bec/bec_widgets/widgets/data_viewer/data_viewer.py +++ b/superxas_bec/bec_widgets/widgets/data_viewer/data_viewer.py @@ -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_()) diff --git a/superxas_bec/bec_widgets/widgets/data_viewer/panels/input_panel.py b/superxas_bec/bec_widgets/widgets/data_viewer/panels/input_panel.py index f249b94..3d2088e 100644 --- a/superxas_bec/bec_widgets/widgets/data_viewer/panels/input_panel.py +++ b/superxas_bec/bec_widgets/widgets/data_viewer/panels/input_panel.py @@ -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 diff --git a/superxas_bec/bec_widgets/widgets/data_viewer/panels/scan_view.py b/superxas_bec/bec_widgets/widgets/data_viewer/panels/scan_view.py index aa3ebe7..9c730b1 100644 --- a/superxas_bec/bec_widgets/widgets/data_viewer/panels/scan_view.py +++ b/superxas_bec/bec_widgets/widgets/data_viewer/panels/scan_view.py @@ -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() diff --git a/superxas_bec/bec_widgets/widgets/data_viewer/widgets/qt_widgets.py b/superxas_bec/bec_widgets/widgets/data_viewer/widgets/qt_widgets.py index 187157e..7242c1e 100644 --- a/superxas_bec/bec_widgets/widgets/data_viewer/widgets/qt_widgets.py +++ b/superxas_bec/bec_widgets/widgets/data_viewer/widgets/qt_widgets.py @@ -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)