diff --git a/bec_widgets/applications/main_app.py b/bec_widgets/applications/main_app.py index 163132dd6..ff4924bcf 100644 --- a/bec_widgets/applications/main_app.py +++ b/bec_widgets/applications/main_app.py @@ -3,6 +3,7 @@ from bec_widgets.applications.navigation_centre.reveal_animator import ANIMATION_DURATION from bec_widgets.applications.navigation_centre.side_bar import SideBar from bec_widgets.applications.navigation_centre.side_bar_components import NavigationItem +from bec_widgets.applications.views.admin_view.admin_view import AdminView from bec_widgets.applications.views.developer_view.developer_view import DeveloperView from bec_widgets.applications.views.device_manager_view.device_manager_view import DeviceManagerView from bec_widgets.applications.views.view import ViewBase, WaveformViewInline, WaveformViewPopup @@ -54,6 +55,7 @@ def _add_views(self): self.ads.setObjectName("MainWorkspace") self.device_manager = DeviceManagerView(self) self.developer_view = DeveloperView(self) + self.admin_view = AdminView(self) self.add_view( icon="widgets", title="Dock Area", id="dock_area", widget=self.ads, mini_text="Docks" @@ -72,6 +74,13 @@ def _add_views(self): id="developer_view", exclusive=True, ) + self.add_view( + icon="admin_panel_settings", + title="Admin View", + widget=self.admin_view, + id="admin_view", + mini_text="Admin", + ) if self._show_examples: self.add_section("Examples", "examples") diff --git a/bec_widgets/applications/views/admin_view/__init__.py b/bec_widgets/applications/views/admin_view/__init__.py new file mode 100644 index 000000000..e69de29bb diff --git a/bec_widgets/applications/views/admin_view/admin_view.py b/bec_widgets/applications/views/admin_view/admin_view.py new file mode 100644 index 000000000..ec6e2f593 --- /dev/null +++ b/bec_widgets/applications/views/admin_view/admin_view.py @@ -0,0 +1,42 @@ +"""Module for Admin View.""" + +from qtpy.QtWidgets import QWidget + +from bec_widgets.applications.views.admin_view.admin_widget import AdminWidget +from bec_widgets.applications.views.view import ViewBase +from bec_widgets.utils.error_popups import SafeSlot + + +class AdminView(ViewBase): + """ + A view for administrators to change the current active experiment, manage messaging + services, and more tasks reserved for users with admin privileges. + """ + + def __init__( + self, + parent: QWidget | None = None, + content: QWidget | None = None, + *, + id: str | None = None, + title: str | None = None, + ): + super().__init__(parent=parent, content=content, id=id, title=title) + self.admin_widget = AdminWidget(parent=self) + self.set_content(self.admin_widget) + + @SafeSlot() + def on_enter(self) -> None: + """Called after the view becomes current/visible. + + Default implementation does nothing. Override in subclasses. + """ + self.admin_widget.on_enter() + + @SafeSlot() + def on_exit(self) -> None: + """Called before the view is hidden. + + Default implementation does nothing. Override in subclasses. + """ + self.admin_widget.on_exit() diff --git a/bec_widgets/applications/views/admin_view/admin_widget.py b/bec_widgets/applications/views/admin_view/admin_widget.py new file mode 100644 index 000000000..8c42db1c5 --- /dev/null +++ b/bec_widgets/applications/views/admin_view/admin_widget.py @@ -0,0 +1,48 @@ +"""Module to define a widget for the admin view.""" + +from __future__ import annotations + +from qtpy.QtWidgets import QVBoxLayout, QWidget + +from bec_widgets.utils.bec_widget import BECWidget +from bec_widgets.widgets.services.bec_atlas_admin_view.bec_atlas_admin_view import BECAtlasAdminView + + +class AdminWidget(BECWidget, QWidget): + """Widget for admin view.""" + + RPC = False + + def __init__(self, parent=None, client=None): + super().__init__(parent=parent, client=client) + # Overview widget + layout = QVBoxLayout(self) + self.admin_view_widget = BECAtlasAdminView(parent=self, client=self.client) + layout.addWidget(self.admin_view_widget) + + def on_enter(self) -> None: + """Called after the widget becomes visible.""" + # self.admin_view_widget.check_health() + + def on_exit(self) -> None: + """Called before the widget is hidden.""" + self.admin_view_widget.logout() + + +if __name__ == "__main__": + from bec_qthemes import apply_theme + from qtpy.QtWidgets import QApplication + + from bec_widgets.widgets.utility.visual.dark_mode_button.dark_mode_button import DarkModeButton + + app = QApplication([]) + + apply_theme("dark") + w = QWidget() + l = QVBoxLayout(w) + widget = AdminWidget(parent=w) + dark_mode_button = DarkModeButton(parent=w) + l.addWidget(dark_mode_button) + l.addWidget(widget) + w.show() + app.exec() diff --git a/bec_widgets/widgets/services/bec_atlas_admin_view/__init__.py b/bec_widgets/widgets/services/bec_atlas_admin_view/__init__.py new file mode 100644 index 000000000..e69de29bb diff --git a/bec_widgets/widgets/services/bec_atlas_admin_view/bec_atlas_admin_view.py b/bec_widgets/widgets/services/bec_atlas_admin_view/bec_atlas_admin_view.py new file mode 100644 index 000000000..f09764c94 --- /dev/null +++ b/bec_widgets/widgets/services/bec_atlas_admin_view/bec_atlas_admin_view.py @@ -0,0 +1,399 @@ +"""Admin View panel for setting up account and messaging services in BEC.""" + +from __future__ import annotations + +import json +from typing import TYPE_CHECKING + +from bec_lib.endpoints import MessageEndpoints +from bec_lib.logger import bec_logger +from bec_lib.messages import DeploymentInfoMessage, ExperimentInfoMessage +from qtpy.QtCore import Qt, Signal +from qtpy.QtWidgets import ( + QHBoxLayout, + QMessageBox, + QSizePolicy, + QStackedLayout, + QVBoxLayout, + QWidget, +) + +from bec_widgets.utils.bec_widget import BECWidget +from bec_widgets.utils.error_popups import SafeSlot +from bec_widgets.utils.toolbars.actions import MaterialIconAction +from bec_widgets.utils.toolbars.bundles import ToolbarBundle +from bec_widgets.utils.toolbars.toolbar import ModularToolBar +from bec_widgets.widgets.services.bec_atlas_admin_view.bec_atlas_http_service import ( + BECAtlasHTTPService, + HTTPResponse, +) +from bec_widgets.widgets.services.bec_atlas_admin_view.experiment_selection.experiment_mat_card import ( + ExperimentMatCard, +) +from bec_widgets.widgets.services.bec_atlas_admin_view.experiment_selection.experiment_selection import ( + ExperimentSelection, +) + +if TYPE_CHECKING: # pragma: no cover + from bec_lib.messages import ExperimentInfoMessage + +logger = bec_logger.logger + + +class OverviewWidget(QWidget): + """Overview Widget for the BEC Atlas Admin view""" + + change_experiment_requested = Signal() + + def __init__(self, parent=None): + super().__init__(parent=parent) + layout = QHBoxLayout(self) + self.setAutoFillBackground(True) + layout.setAlignment(Qt.AlignmentFlag.AlignCenter) + self.setSizePolicy(QSizePolicy.Policy.Expanding, QSizePolicy.Policy.Expanding) + layout.setContentsMargins(16, 16, 16, 16) + layout.setSpacing(16) + self._experiment_info: ExperimentInfoMessage | None = None + self._mat_card = ExperimentMatCard( + parent=self, + show_activate_button=True, + button_text="Change Experiment", + title="Current Experiment", + ) + layout.addWidget(self._mat_card) + self._mat_card.experiment_selected.connect(self._on_experiment_selected) + + def _on_experiment_selected(self, experiment_info: dict) -> None: + """We reuse the experiment_selected signal from the mat card to trigger the login and experiment change process.""" + self.change_experiment_requested.emit() + + @SafeSlot(dict) + def set_experiment_info(self, experiment_info: dict) -> None: + self._experiment_info = ExperimentInfoMessage.model_validate(experiment_info) + self._mat_card.set_experiment_info(self._experiment_info) + + +class BECAtlasAdminView(BECWidget, QWidget): + + authenticated = Signal(bool) + account_changed = Signal(str) + messaging_service_activated = Signal(str) + + def __init__( + self, parent=None, atlas_url: str = "https://bec-atlas-dev.psi.ch/api/v1", client=None + ): + super().__init__(parent=parent, client=client) + + # State variables + self._current_deployment_info: DeploymentInfoMessage | None = None + self._current_deployment_info = None + self._current_session_info = None + self._current_experiment_info = None + self._authenticated = False + + # Root layout + self.root_layout = QVBoxLayout(self) + self.root_layout.setContentsMargins(0, 0, 0, 0) + self.root_layout.setSpacing(0) + + # Toolbar for navigation between different views in the admin panel + self.toolbar = ModularToolBar(self) + self.init_toolbar() + self.root_layout.insertWidget(0, self.toolbar) + self.toolbar.show_bundles(["view", "auth"]) + + # Stacked layout to switch between overview, experiment selection and messaging services + # It is added below the toolbar + self.stacked_layout = QStackedLayout() + self.stacked_layout.setContentsMargins(0, 0, 0, 0) + self.stacked_layout.setSpacing(0) + self.stacked_layout.setStackingMode(QStackedLayout.StackingMode.StackAll) + self.root_layout.addLayout(self.stacked_layout) + + # Overview widget + self.overview_widget = OverviewWidget(parent=self) + self.stacked_layout.addWidget(self.overview_widget) + self.overview_widget.change_experiment_requested.connect( + self._on_change_experiment_requested + ) + + # Experiment Selection widget + self.experiment_selection = ExperimentSelection(parent=self) + self.stacked_layout.addWidget(self.experiment_selection) + self.experiment_selection.experiment_selected.connect(self._on_experiment_selected) + + # BEC Atlas HTTP Service + self.atlas_http_service = BECAtlasHTTPService( + parent=self, base_url=atlas_url, headers={"accept": "application/json"} + ) + + # Connect signals + self.atlas_http_service.http_response_received.connect(self._on_http_response_received) + self.atlas_http_service.authenticated.connect(self._on_authenticated) + self.atlas_http_service.account_changed.connect(self.account_changed.emit) + + self.bec_dispatcher.connect_slot( + slot=self._update_deployment_info, + topics=MessageEndpoints.deployment_info(), + from_start=True, + ) + + @SafeSlot() + def _on_change_experiment_requested(self) -> None: + """Handle the change experiment requested signal from the overview widget.""" + if self._authenticated: + self._on_experiment_selection_selected() + else: + self.login() + + @SafeSlot(dict) + def _on_experiment_selected(self, experiment_info: dict) -> None: + """Handle the experiment selected signal from the experiment selection widget""" + experiment_info = ExperimentInfoMessage.model_validate(experiment_info) + experiment_id = experiment_info.pgroup + deployment_id = self._current_deployment_info.deployment_id + self.set_experiment(experiment_id=experiment_id, deployment_id=deployment_id) + + @SafeSlot(dict, dict) + def _update_deployment_info(self, msg: dict, metadata: dict) -> None: + """Fetch current deployment info from the server.""" + deployment = DeploymentInfoMessage.model_validate(msg) + self._current_deployment_info = deployment + self._current_session_info = deployment.active_session + if self._current_session_info is not None: + self._current_experiment_info = self._current_session_info.experiment + + self.overview_widget.set_experiment_info( + self._current_experiment_info.model_dump() if self._current_experiment_info else {} + ) + + def init_toolbar(self): + """Initialize the toolbar for the admin view. This allows to switch between different views in the admin panel.""" + # Overview + overview = MaterialIconAction( + icon_name="home", + tooltip="Show Overview Panel", + label_text="Overview", + text_position="under", + parent=self, + filled=True, + ) + overview.action.triggered.connect(self._on_overview_selected) + self.toolbar.components.add_safe("overview", overview) + + # Experiment Selection + experiment_selection = MaterialIconAction( + icon_name="experiment", + tooltip="Show Experiment Selection Panel", + label_text="Experiment Selection", + text_position="under", + parent=self, + filled=True, + ) + experiment_selection.action.triggered.connect(self._on_experiment_selection_selected) + experiment_selection.action.setEnabled(False) # Initially disabled until authenticated + self.toolbar.components.add_safe("experiment_selection", experiment_selection) + + # Messaging Services + messaging_services = MaterialIconAction( + icon_name="chat", + tooltip="Show Messaging Services Panel", + label_text="Messaging Services", + text_position="under", + parent=self, + filled=True, + ) + messaging_services.action.triggered.connect(self._on_messaging_services_selected) + messaging_services.action.setEnabled(False) # Initially disabled until authenticated + self.toolbar.components.add_safe("messaging_services", messaging_services) + + # Login + login_action = MaterialIconAction( + icon_name="login", + tooltip="Login", + label_text="Login", + text_position="under", + parent=self, + filled=True, + ) + login_action.action.triggered.connect(self.login) + self.toolbar.components.add_safe("login", login_action) + + # Logout + logout_action = MaterialIconAction( + icon_name="logout", + tooltip="Logout", + label_text="Logout", + text_position="under", + parent=self, + filled=True, + ) + logout_action.action.triggered.connect(self.logout) + logout_action.action.setEnabled(False) # Initially disabled until authenticated + self.toolbar.components.add_safe("logout", logout_action) + + # Add view_bundle to toolbar + view_bundle = ToolbarBundle("view", self.toolbar.components) + view_bundle.add_action("overview") + view_bundle.add_action("experiment_selection") + view_bundle.add_action("messaging_services") + self.toolbar.add_bundle(view_bundle) + + # Add auth_bundle to toolbar + auth_bundle = ToolbarBundle("auth", self.toolbar.components) + auth_bundle.add_action("login") + auth_bundle.add_action("logout") + self.toolbar.add_bundle(auth_bundle) + + def _on_overview_selected(self): + """Show the overview panel.""" + self.overview_widget.setVisible(True) + self.experiment_selection.setVisible(False) + self.stacked_layout.setCurrentWidget(self.overview_widget) + + def _on_experiment_selection_selected(self): + """Show the experiment selection panel.""" + if not self._authenticated: + logger.warning("Attempted to access experiment selection without authentication.") + return + self.overview_widget.setVisible(False) + self.experiment_selection.setVisible(True) + self.stacked_layout.setCurrentWidget(self.experiment_selection) + + def _on_messaging_services_selected(self): + """Show the messaging services panel.""" + logger.info("Messaging services panel is not implemented yet.") + # TODO + return + # if not self._authenticated: + # logger.warning("Attempted to access messaging services without authentication.") + # return + # self.overview_widget.setVisible(False) + # self.experiment_selection.setVisible(False) + + def _fetch_available_experiments(self): + """Fetch the list of available experiments for the authenticated user.""" + # What if this is None, should this be an optional user input in the UI? + if self._current_experiment_info is None: + logger.error( + "No current experiment info available, cannot fetch available experiments." + ) + return + current_realm_id = self._current_experiment_info.realm_id + if current_realm_id is None: + logger.error( + "Current experiment does not have a realm_id, cannot fetch available experiments." + ) + return + self.atlas_http_service.get_experiments_for_realm(current_realm_id) + + def _on_http_response_received(self, response: dict) -> None: + """Handle the HTTP response received from the BEC Atlas API.""" + response = HTTPResponse(**response) + logger.info(f"HTTP Response received: {response.request_url} with status {response.status}") + if "realms/experiments" in response.request_url and response.status == 200: + experiments = response.data if isinstance(response.data, list) else [] + self.experiment_selection.set_experiment_infos(experiments) + self._on_experiment_selection_selected() # Switch to experiment selection once experiments are loaded + elif "deployments/experiment" in response.request_url: + if response.status == 200: + self._on_overview_selected() # Switch back to overview to show updated experiment info + else: + logger.error( + f"Failed to fetch experiments for realm. Status: {response.status}, Response: {response.data}" + ) + self._show_warning_message( + title="Failed to Fetch Experiments", + text="An error occured while fetching experiment information. Please check logs for details.", + ) + + def _show_warning_message(self, title: str, text: str): + """Show a warning message box.""" + + msg_box = QMessageBox.warning(self, title, text, QMessageBox.StandardButton.Ok) + msg_box.exec_() + + def _on_authenticated(self, authenticated: bool) -> None: + """Handle authentication state change.""" + self._authenticated = authenticated + self.authenticated.emit(authenticated) + if authenticated: + self.toolbar.components.get_action("experiment_selection").action.setEnabled(True) + self.toolbar.components.get_action("messaging_services").action.setEnabled(True) + self.toolbar.components.get_action("login").action.setEnabled(False) + self.toolbar.components.get_action("logout").action.setEnabled(True) + self._fetch_available_experiments() # Fetch experiments upon successful authentication + else: + self.toolbar.components.get_action("experiment_selection").action.setEnabled(False) + self.toolbar.components.get_action("messaging_services").action.setEnabled(False) + self.toolbar.components.get_action("login").action.setEnabled(True) + self.toolbar.components.get_action("logout").action.setEnabled(False) + # Delete data in experiment selection widget upon logout + self.experiment_selection.set_experiment_infos([]) + self._on_overview_selected() # Switch back to overview on logout + + @SafeSlot(dict) + def set_experiment(self, experiment_id: str, deployment_id: str) -> None: + """Set the experiment information for the current experiment.""" + self.atlas_http_service.set_experiment(experiment_id, deployment_id) + + def check_health(self) -> None: + """Check the health of the BEC Atlas API.""" + self.atlas_http_service.check_health() + + def login(self) -> None: + """Login to the BEC Atlas API.""" + self.atlas_http_service.login() + + def logout(self) -> None: + """Logout from the BEC Atlas API.""" + self.atlas_http_service.logout() + + def cleanup(self): + self.atlas_http_service.cleanup() + return super().cleanup() + + +if __name__ == "__main__": + import sys + + from bec_qthemes import apply_theme + from qtpy.QtWidgets import QApplication + + app = QApplication(sys.argv) + + apply_theme("light") + window = BECAtlasAdminView() + + exp_info_dict = { + "_id": "p22622", + "owner_groups": ["admin"], + "access_groups": ["unx-sls_xda_bs", "p22622"], + "realm_id": "TestBeamline", + "proposal": "12345967", + "title": "Test Experiment for Mat Card Widget", + "firstname": "John", + "lastname": "Doe", + "email": "john.doe@psi.ch", + "account": "doe_j", + "pi_firstname": "Jane", + "pi_lastname": "Smith", + "pi_email": "jane.smith@psi.ch", + "pi_account": "smith_j", + "eaccount": "e22622", + "pgroup": "p22622", + "abstract": "This is a test abstract for the experiment mat card widget. It should be long enough to test text wrapping and display in the card. The abstract provides a brief overview of the experiment, its goals, and its significance. This text is meant to simulate a real abstract that might be associated with an experiment in the BEC Atlas system. The card should be able to handle abstracts of varying lengths without any issues, ensuring that the user can read the full abstract even if it is quite long.", + "schedule": [{"start": "01/01/2025 08:00:00", "end": "03/01/2025 18:00:00"}], + "proposal_submitted": "15/12/2024", + "proposal_expire": "31/12/2025", + "proposal_status": "Scheduled", + "delta_last_schedule": 30, + "mainproposal": "", + } + from bec_lib.messages import ExperimentInfoMessage + + proposal_info = ExperimentInfoMessage(**exp_info_dict) + window.set_experiment_info(proposal_info) + window.resize(800, 600) + window.show() + sys.exit(app.exec_()) diff --git a/bec_widgets/widgets/services/bec_atlas_admin_view/bec_atlas_http_service.py b/bec_widgets/widgets/services/bec_atlas_admin_view/bec_atlas_http_service.py new file mode 100644 index 000000000..315d57888 --- /dev/null +++ b/bec_widgets/widgets/services/bec_atlas_admin_view/bec_atlas_http_service.py @@ -0,0 +1,206 @@ +import json + +from pydantic import BaseModel +from qtpy.QtCore import QUrl, QUrlQuery, Signal +from qtpy.QtNetwork import QNetworkAccessManager, QNetworkReply, QNetworkRequest +from qtpy.QtWidgets import QMessageBox, QWidget + +from bec_widgets.utils.error_popups import SafeSlot +from bec_widgets.widgets.services.bec_atlas_admin_view.login_dialog import LoginDialog + + +class HTTPResponse(BaseModel): + request_url: str + headers: dict + status: int + data: dict | list | str + + +class BECAtlasHTTPService(QWidget): + """HTTP service using the QNetworkAccessManager to interact with the BEC Atlas API.""" + + http_response_received = Signal(dict) + authenticated = Signal(bool) + account_changed = Signal(bool) + + def __init__(self, parent=None, base_url: str = "", headers: dict | None = None): + super().__init__(parent) + if headers is None: + headers = {"accept": "application/json"} + self._headers = headers + self._base_url = base_url + self.network_manager = QNetworkAccessManager(self) + self.network_manager.finished.connect(self._handle_response) + self._authenticated = False + + def closeEvent(self, event): + self.cleanup() + return super().closeEvent(event) + + def cleanup(self): + """Cleanup connection, destroy authenticate cookies.""" + + # Disconnect signals to avoid handling responses after cleanup + self.network_manager.finished.disconnect(self._handle_response) + + # Logout to invalidate session on server side + self.logout() + + # Delete all cookies related to the base URL + for cookie in self.network_manager.cookieJar().cookiesForUrl(QUrl(self._base_url)): + self.network_manager.cookieJar().deleteCookie(cookie) + + def _handle_response(self, reply: QNetworkReply): + """ + Handle the HTTP response from the server. + + Args: + reply (QNetworkReply): The network reply object containing the response. + """ + status = reply.attribute(QNetworkRequest.Attribute.HttpStatusCodeAttribute) + raw_bytes = reply.readAll().data() + request_url = reply.url().toString() + headers = dict(reply.rawHeaderPairs()) + reply.deleteLater() + + if "login" in request_url and status == 200: + self._authenticated = True + self.authenticated.emit(True) + elif "logout" in request_url and status == 200: + self._authenticated = False + self.authenticated.emit(False) + if "deployments/experiment" in request_url and status == 200: + self.account_changed.emit(True) + elif "deployments/experiment" in request_url and status != 200: + self.account_changed.emit(False) + + # TODO, should we handle failures here or rather on more high levels? + if status == 401: + if "login" in request_url: + # Failed login attempt + self._show_warning( + title="Login Failed", text="Please check your login credentials." + ) + else: + self._show_warning( + title="Unauthorized", + text="You are not authorized to request this information. Please authenticate first.", + ) + return + self._handle_raw_response(raw_bytes, status, request_url, headers) + + def _handle_raw_response(self, raw_bytes: bytes, status: int, request_url: str, headers: dict): + try: + if len(raw_bytes) > 0: + data = json.loads(raw_bytes.decode("utf-8")) + else: + data = {} + + except Exception: + data = {} # Drop this + # + # jwt.decode(token, options={"verify_signature": False}) + # dict with email and expiration time. + # from datetime import datetime + # datetime.now().timestamp() + + response = HTTPResponse(request_url=request_url, headers=headers, status=status, data=data) + self.http_response_received.emit(response.model_dump()) + + def _show_warning(self, title: str, text: str): + """Show a warning message box for unauthorized access.""" + QMessageBox.warning(self, title, text, QMessageBox.StandardButton.Ok) + + def _show_login(self): + """Show the login dialog to enter credentials.""" + dlg = LoginDialog(parent=self) + dlg.credentials_entered.connect(self._set_credentials) + dlg.exec_() # blocking here is OK for login + + def _set_credentials(self, username: str, password: str): + """Set the credentials and perform login.""" + self.post_request("/user/login", {"username": username, "password": password}) + + ################ + # HTTP Methods + ################ + + def get_request(self, endpoint: str, query_parameters: dict | None = None): + """ + GET request to the API endpoint. + + Args: + endpoint (str): The API endpoint to send the GET request to. + query_parameters (dict | None): Optional query parameters to include in the URL. + """ + url = QUrl(self._base_url + endpoint) + if query_parameters: + query = QUrlQuery() + for key, value in query_parameters.items(): + query.addQueryItem(key, value) + url.setQuery(query) + request = QNetworkRequest(url) + for key, value in self._headers.items(): + request.setRawHeader(key.encode("utf-8"), value.encode("utf-8")) + self.network_manager.get(request) + + def post_request( + self, endpoint: str, payload: dict | None = None, query_parameters: dict | None = None + ): + """ + POST request to the API endpoint with a JSON payload. + + Args: + endpoint (str): The API endpoint to send the POST request to. + payload (dict): The JSON payload to include in the POST request. + query_parameters (dict | None): Optional query parameters to include in the URL. + """ + if payload is None: + payload = {} + url = QUrl(self._base_url + endpoint) + if query_parameters: + query = QUrlQuery() + for key, value in query_parameters.items(): + query.addQueryItem(key, value) + url.setQuery(query) + request = QNetworkRequest(url) + + # Headers + for key, value in self._headers.items(): + request.setRawHeader(key.encode("utf-8"), value.encode("utf-8")) + request.setHeader(QNetworkRequest.KnownHeaders.ContentTypeHeader, "application/json") + + payload_dump = json.dumps(payload).encode("utf-8") + reply = self.network_manager.post(request, payload_dump) + + ################ + # API Methods + ################ + + @SafeSlot() + def login(self): + """Login to BEC Atlas with the provided username and password.""" + # TODO should we prompt here if already authenticated - and add option to skip login otherwise first destroy old token and re-authenticate? + self._show_login() + + def logout(self): + """Logout from BEC Atlas.""" + self.post_request("/user/logout", {}) + + def check_health(self): + """Check the health status of BEC Atlas.""" + self.get_request("/health") + + def get_experiments_for_realm(self, realm_id: str): + """Get the list of realms from BEC Atlas. Requires authentication.""" + endpoint = "/realms/experiments" + query_parameters = {"realm_id": realm_id} + self.get_request(endpoint, query_parameters=query_parameters) + + @SafeSlot(str, str) + def set_experiment(self, experiment_id: str, deployment_id: str) -> None: + """Set the current experiment information for the service.""" + self.post_request( + "/deployments/experiment", + query_parameters={"experiment_id": experiment_id, "deployment_id": deployment_id}, + ) diff --git a/bec_widgets/widgets/services/bec_atlas_admin_view/experiment_selection/__init__.py b/bec_widgets/widgets/services/bec_atlas_admin_view/experiment_selection/__init__.py new file mode 100644 index 000000000..e69de29bb diff --git a/bec_widgets/widgets/services/bec_atlas_admin_view/experiment_selection/experiment_mat_card.py b/bec_widgets/widgets/services/bec_atlas_admin_view/experiment_selection/experiment_mat_card.py new file mode 100644 index 000000000..ac885c9bc --- /dev/null +++ b/bec_widgets/widgets/services/bec_atlas_admin_view/experiment_selection/experiment_mat_card.py @@ -0,0 +1,259 @@ +"""Mat-card like widget to display experiment details. Optionally, a button on the bottom which the user can click to trigger the selection of the experiment.""" + +from bec_lib.messages import ExperimentInfoMessage +from qtpy.QtCore import Qt, Signal +from qtpy.QtWidgets import ( + QFrame, + QGraphicsDropShadowEffect, + QGridLayout, + QGroupBox, + QHBoxLayout, + QLabel, + QPushButton, + QSizePolicy, + QVBoxLayout, + QWidget, +) + +from bec_widgets.utils.bec_widget import BECWidget +from bec_widgets.utils.colors import get_theme_palette +from bec_widgets.utils.round_frame import RoundedFrame +from bec_widgets.widgets.services.bec_atlas_admin_view.experiment_selection.utils import ( + format_name, + format_schedule, +) +from bec_widgets.widgets.utility.visual.dark_mode_button.dark_mode_button import DarkModeButton + + +class BorderLessLabel(QLabel): + """A QLabel that does not show any border, even when stylesheets try to apply one.""" + + def __init__(self, *args, **kwargs): + super().__init__(*args, **kwargs) + self.setStyleSheet("border: none;") + self.setSizePolicy(QSizePolicy.Policy.Minimum, QSizePolicy.Policy.Preferred) + + +class ExperimentMatCard(BECWidget, QWidget): + + experiment_selected = Signal(dict) + + def __init__( + self, + parent=None, + show_activate_button: bool = True, + button_text: str = "Activate", + title: str = "Next Experiment", + **kwargs, + ): + super().__init__(parent=parent, theme_update=True, **kwargs) + + layout = QVBoxLayout(self) + layout.setContentsMargins(12, 8, 12, 8) + self.experiment_info = {} + self._abstract_text = "" + + # Add card frame with shadow and custom styling + self._card_frame = QFrame(parent=self) + layout = QVBoxLayout(self._card_frame) + self._card_frame.setSizePolicy(QSizePolicy.Preferred, QSizePolicy.Preferred) + palette = get_theme_palette() + self._card_frame.setStyleSheet( + f""" + border: 1px solid {palette.mid().color().name()}; + background: {palette.base().color().name()}; + """ + ) + shadow = QGraphicsDropShadowEffect(self._card_frame) + shadow.setBlurRadius(18) + shadow.setOffset(0, 4) + shadow.setColor(palette.shadow().color()) + self._card_frame.setGraphicsEffect(shadow) + + self._group_box = QGroupBox(self._card_frame) + self._group_box.setStyleSheet( + "QGroupBox { border: none; }; QLabel { border: none; padding: 0px; }" + ) + self._fill_group_box( + title=title, show_activate_button=show_activate_button, button_text=button_text + ) + self.apply_theme("light") + + def apply_theme(self, theme: str): + palette = get_theme_palette() + self._card_frame.setStyleSheet( + f""" + border: 1px solid {palette.mid().color().name()}; + background: {palette.base().color().name()}; + """ + ) + shadow = self._card_frame.graphicsEffect() + if isinstance(shadow, QGraphicsDropShadowEffect): + shadow.setColor(palette.shadow().color()) + + def _fill_group_box( + self, title: str, show_activate_button: bool, button_text: str = "Activate" + ): + group_layout = QVBoxLayout(self._group_box) + group_layout.setContentsMargins(16, 16, 16, 16) + group_layout.setSpacing(12) + + title_row = QHBoxLayout() + self._card_title = BorderLessLabel(title, self._group_box) + self._card_title.setStyleSheet( + """ + border: none; + font-size: 14px; + font-weight: 600; + """ + ) + + # Add title row and info button to QH layout, then add it to QV layout + title_row.addWidget(self._card_title) + title_row.addStretch(1) + group_layout.addLayout(title_row) + + self._card_grid = QGridLayout() + self._card_grid.setHorizontalSpacing(12) + self._card_grid.setVerticalSpacing(8) + self._card_grid.setColumnStretch(1, 1) + + self._card_pgroup = BorderLessLabel("-", self._group_box) + self._card_title_value = BorderLessLabel("-", self._group_box) + self._card_title_value.setWordWrap(True) + self._card_title_value.setSizePolicy(QSizePolicy.Expanding, QSizePolicy.Preferred) + self._card_name = BorderLessLabel("-", self._group_box) + self._card_start = BorderLessLabel("-", self._group_box) + self._card_end = BorderLessLabel("-", self._group_box) + + self._card_row_labels = [] + + def _row_label(text): + label = BorderLessLabel(text, self._group_box) + self._card_row_labels.append(label) + return label + + self._card_grid.addWidget(_row_label("Name"), 0, 0) + self._card_grid.addWidget(self._card_name, 0, 1) + self._card_grid.addWidget(_row_label("Title"), 1, 0) + self._card_grid.addWidget(self._card_title_value, 1, 1) + self._card_grid.addWidget(_row_label("P-group"), 2, 0) + self._card_grid.addWidget(self._card_pgroup, 2, 1) + self._card_grid.addWidget(_row_label("Schedule (start)"), 3, 0) + self._card_grid.addWidget(self._card_start, 3, 1) + self._card_grid.addWidget(_row_label("Schedule (end)"), 4, 0) + self._card_grid.addWidget(self._card_end, 4, 1) + + # Add to groupbox + group_layout.addLayout(self._card_grid) + + # Add abstract field at the bottom of the card. + self._abstract_label = BorderLessLabel("", self._group_box) + self._abstract_label.setWordWrap(True) + self._abstract_label.setSizePolicy(QSizePolicy.Expanding, QSizePolicy.Preferred) + group_layout.addWidget(self._abstract_label) + + # Add activate button at the bottom + self._activate_button = QPushButton(button_text, self._group_box) + self._activate_button.clicked.connect(self._emit_next_experiment) + self._activate_button.setSizePolicy( + QSizePolicy.Policy.Expanding, QSizePolicy.Policy.Preferred + ) + group_layout.addWidget(self._activate_button, alignment=Qt.AlignmentFlag.AlignHCenter) + self._activate_button.setVisible(show_activate_button) + self._activate_button.setEnabled(False) + + self._card_frame.layout().setContentsMargins(12, 12, 12, 12) + self._card_frame.layout().addWidget(self._group_box) + + card_row = QHBoxLayout() + card_row.addStretch(0) + card_row.addWidget(self._card_frame) + card_row.addStretch(0) + + layout = self.layout() + layout.addStretch(0) + layout.addLayout(card_row) + layout.addStretch(0) + + def _emit_next_experiment(self): + self.experiment_selected.emit(self.experiment_info) + + def set_experiment_info(self, info: ExperimentInfoMessage | dict): + """ + Set the experiment information to display on the card. + + Args: + info (ExperimentInfoMessage | dict): The experiment information to display. Can be either a + dictionary or an ExperimentInfoMessage instance. + """ + if isinstance(info, dict): + info = ExperimentInfoMessage(**info) + + start, end = format_schedule(info.schedule) + self._card_pgroup.setText(info.pgroup or "-") + self._card_title_value.setText(info.title or "-") + self._card_name.setText(format_name(info)) + self._card_start.setText(start or "-") + self._card_end.setText(end or "-") + self._abstract_text = (info.abstract or "").strip() + self._abstract_label.setText(self._abstract_text if self._abstract_text else "") + self.experiment_info = info.model_dump() + self._activate_button.setEnabled(True) + + def set_title(self, title: str): + """ + Set the title displayed at the top of the card. + + Args: + title (str): The title text to display. + """ + self._card_title.setText(title) + + +if __name__ == "__main__": + import sys + + from bec_qthemes import apply_theme + from qtpy.QtWidgets import QApplication + + exp_info = { + "_id": "p22622", + "owner_groups": ["admin"], + "access_groups": ["unx-sls_xda_bs", "p22622"], + "realm_id": "TestBeamline", + "proposal": "12345967", + "title": "Test Experiment for Mat Card Widget", + "firstname": "John", + "lastname": "Doe", + "email": "john.doe@psi.ch", + "account": "doe_j", + "pi_firstname": "Jane", + "pi_lastname": "Smith", + "pi_email": "jane.smith@psi.ch", + "pi_account": "smith_j", + "eaccount": "e22622", + "pgroup": "p22622", + "abstract": "This is a test abstract for the experiment mat card widget. It should be long enough to test text wrapping and display in the card. The abstract provides a brief overview of the experiment, its goals, and its significance. This text is meant to simulate a real abstract that might be associated with an experiment in the BEC Atlas system. The card should be able to handle abstracts of varying lengths without any issues, ensuring that the user can read the full abstract even if it is quite long.", + "schedule": [{"start": "01/01/2025 08:00:00", "end": "03/01/2025 18:00:00"}], + "proposal_submitted": "15/12/2024", + "proposal_expire": "31/12/2025", + "proposal_status": "Scheduled", + "delta_last_schedule": 30, + "mainproposal": "", + } + + app = QApplication(sys.argv) + + apply_theme("dark") + w = QWidget() + l = QVBoxLayout(w) + button = DarkModeButton() + widget = ExperimentMatCard() + widget.set_experiment_info(exp_info) + widget.set_title("Scheduled Experiment") + l.addWidget(button) + l.addWidget(widget) + w.resize(w.sizeHint()) + w.show() + sys.exit(app.exec()) diff --git a/bec_widgets/widgets/services/bec_atlas_admin_view/experiment_selection/experiment_selection.py b/bec_widgets/widgets/services/bec_atlas_admin_view/experiment_selection/experiment_selection.py new file mode 100644 index 000000000..e51d6cdb7 --- /dev/null +++ b/bec_widgets/widgets/services/bec_atlas_admin_view/experiment_selection/experiment_selection.py @@ -0,0 +1,430 @@ +"""Experiment Selection View for BEC Atlas Admin Widget""" + +from datetime import datetime +from typing import Any + +from bec_lib.logger import bec_logger +from qtpy.QtCore import Qt, Signal +from qtpy.QtWidgets import ( + QCheckBox, + QHBoxLayout, + QHeaderView, + QLabel, + QLineEdit, + QPushButton, + QSizePolicy, + QTableWidget, + QTableWidgetItem, + QTabWidget, + QVBoxLayout, + QWidget, +) +from thefuzz import fuzz + +from bec_widgets.utils.error_popups import SafeSlot +from bec_widgets.widgets.services.bec_atlas_admin_view.experiment_selection.experiment_mat_card import ( + ExperimentMatCard, +) + +# from bec_widgets.widgets.services.bec_atlas_admin_view.experiment_selection.material_push_button import ( +# MaterialPushButton, +# ) +from bec_widgets.widgets.services.bec_atlas_admin_view.experiment_selection.utils import ( + format_name, + format_schedule, +) + +logger = bec_logger.logger + +FUZZY_SEARCH_THRESHOLD = 80 + + +def is_match(text: str, data: dict[str, Any], relevant_keys: list[str], enable_fuzzy: bool) -> bool: + """ + Check if the text matches any of the relevant keys in the row data. + + Args: + text (str): The text to search for. + data (dict[str, Any]): The data to search in. + relevant_keys (list[str]): The keys to consider for searching. + enable_fuzzy (bool): Whether to use fuzzy matching. + Returns: + bool: True if a match is found, False otherwise. + """ + for key in relevant_keys: + data_value = str(data.get(key, "") or "") + if enable_fuzzy: + match_ratio = fuzz.partial_ratio(text.lower(), data_value.lower()) + if match_ratio >= FUZZY_SEARCH_THRESHOLD: + return True + else: + if text.lower() in data_value.lower(): + return True + return False + + +class ExperimentSelection(QWidget): + experiment_selected = Signal(dict) + + def __init__(self, experiment_infos=None, parent=None): + super().__init__(parent=parent) + self._experiment_infos = experiment_infos or [] + self._next_experiment = self._select_next_experiment(self._experiment_infos) + self._enable_fuzzy_search: bool = True + self._hidden_rows: set[int] = set() + self._headers: dict[str, str] = { + "pgroup": "P-group", + "title": "Title", + "name": "Name", + "schedule_start": "Schedule (start)", + "schedule_end": "Schedule (end)", + } + self._table_infos: list[dict[str, Any]] = [] + + main_layout = QVBoxLayout(self) + main_layout.setContentsMargins(16, 16, 16, 16) + main_layout.setSpacing(12) + main_layout.setAlignment(Qt.AlignmentFlag.AlignCenter) + self.setSizePolicy(QSizePolicy.Policy.Expanding, QSizePolicy.Policy.Expanding) + self.setAutoFillBackground(True) + + self._tabs = QTabWidget(self) + main_layout.addWidget(self._tabs, stretch=1) + + self._card_tab = ExperimentMatCard( + parent=self, show_activate_button=True, button_text="Activate Next Experiment" + ) + self._card_tab.experiment_selected.connect(self._emit_selected_experiment) + if self._next_experiment: + self._card_tab.set_experiment_info(self._next_experiment) + self._table_tab = QWidget(self) + self._tabs.addTab(self._card_tab, "Next Experiment") + self._tabs.addTab(self._table_tab, "Manual Selection") + + self._build_table_tab() + self._tabs.currentChanged.connect(self._on_tab_changed) + # main_layout.addStretch() + + button_layout = QHBoxLayout() + main_layout.addLayout(button_layout) + self._apply_table_filters() + self.restore_default_view() + + def restore_default_view(self): + """Reset the view to the default state, showing the next experiment card.""" + self._tabs.setCurrentWidget(self._card_tab) + + def set_experiment_infos(self, experiment_infos: list[dict]): + self._experiment_infos = experiment_infos + self._next_experiment = self._select_next_experiment(self._experiment_infos) + if self._next_experiment: + self._card_tab.set_experiment_info(self._next_experiment) + self._apply_table_filters() + + def _setup_search(self, layout: QVBoxLayout): + """ + Create components related to the search functionality + + Args: + layout (QVBoxLayout): The layout to which the search components will be added. + """ + + # Create search bar + search_layout = QHBoxLayout() + self.search_label = QLabel("Search:") + self.search_input = QLineEdit() + self.search_input.setPlaceholderText("Filter experiments...") + self.search_input.setClearButtonEnabled(True) + self.search_input.textChanged.connect(self._apply_row_filter) + search_layout.addWidget(self.search_label) + search_layout.addWidget(self.search_input) + + # Add exact match toggle + fuzzy_layout = QHBoxLayout() + self.fuzzy_label = QLabel("Exact Match:") + self.fuzzy_is_disabled = QCheckBox() + + self.fuzzy_is_disabled.stateChanged.connect(self._state_change_fuzzy_search) + self.fuzzy_is_disabled.setToolTip( + "Enable approximate matching (OFF) and exact matching (ON)" + ) + self.fuzzy_label.setToolTip("Enable approximate matching (OFF) and exact matching (ON)") + fuzzy_layout.addWidget(self.fuzzy_label) + fuzzy_layout.addWidget(self.fuzzy_is_disabled) + fuzzy_layout.addStretch() + + # Add both search components to the layout + self.search_controls = QHBoxLayout() + self.search_controls.addLayout(search_layout) + self.search_controls.addSpacing(20) # Add some space between the search box and toggle + self.search_controls.addLayout(fuzzy_layout) + + # Add filter section for proposals + + filter_layout = QHBoxLayout() + filter_layout.setContentsMargins(12, 0, 12, 0) + filter_layout.setSpacing(12) + self._with_proposals = QCheckBox("Show experiments with proposals", self) + self._without_proposals = QCheckBox("Show experiments without proposals", self) + self._with_proposals.setChecked(True) + self._without_proposals.setChecked(True) + self._with_proposals.toggled.connect(self._apply_table_filters) + self._without_proposals.toggled.connect(self._apply_table_filters) + filter_layout.addWidget(self._with_proposals) + filter_layout.addWidget(self._without_proposals) + filter_layout.addStretch(1) + self.search_controls.addLayout(filter_layout) + + # Insert the search controls layout at the top of the table + layout.addLayout(self.search_controls) + + def _build_table_tab(self): + layout = QVBoxLayout(self._table_tab) + layout.setContentsMargins(16, 16, 16, 16) + layout.setSpacing(8) + + self._setup_search(layout) + + # # Add filter section + # filter_layout = QHBoxLayout() + # self._with_proposals = QCheckBox("Show experiments with proposals", self) + # self._without_proposals = QCheckBox("Show experiments without proposals", self) + # self._with_proposals.setChecked(True) + # self._without_proposals.setChecked(True) + # self._with_proposals.toggled.connect(self._apply_table_filters) + # self._without_proposals.toggled.connect(self._apply_table_filters) + # filter_layout.addWidget(self._with_proposals) + # filter_layout.addWidget(self._without_proposals) + # filter_layout.addStretch(1) + # layout.addLayout(filter_layout) + + # Add table + hor_layout = QHBoxLayout() + self._table = QTableWidget(self._table_tab) + self._table.setColumnCount(5) + self._table.setHorizontalHeaderLabels(list(self._headers.values())) + vh = self._table.verticalHeader() + vh.setVisible(False) + vh.setDefaultSectionSize(vh.minimumSectionSize()) + self._table.setSelectionBehavior(QTableWidget.SelectionBehavior.SelectRows) + self._table.setSelectionMode(QTableWidget.SelectionMode.SingleSelection) + self._table.setWordWrap(True) + self._table.setStyleSheet("QTableWidget::item { padding: 4px; }") + + header = self._table.horizontalHeader() + header.setSectionResizeMode(0, QHeaderView.ResizeToContents) + header.setSectionResizeMode(1, QHeaderView.Stretch) + header.setSectionResizeMode(2, QHeaderView.ResizeToContents) + header.setSectionResizeMode(3, QHeaderView.ResizeToContents) + header.setSectionResizeMode(4, QHeaderView.ResizeToContents) + + self._table.itemSelectionChanged.connect(self._update_selection_state) + hor_layout.addWidget(self._table, stretch=5) + hor_layout.addSpacing(12) # Add space between table and side card + + # Add side card for experiment details + self._side_card = ExperimentMatCard( + parent=self, show_activate_button=True, button_text="Activate Next Experiment" + ) + self._side_card.experiment_selected.connect(self._emit_selected_experiment) + hor_layout.addWidget(self._side_card, stretch=2) # Ratio 5:2 between table and card + layout.addLayout(hor_layout) + + @SafeSlot() + @SafeSlot(int) + @SafeSlot(bool) # Overload for buttons + def _apply_table_filters(self, *args, **kwargs): + if self._tabs.currentWidget() is not self._table_tab: + return + + show_with = self._with_proposals.isChecked() + show_without = self._without_proposals.isChecked() + + self._table_infos = [] + for info in self._experiment_infos: + has_proposal = bool(info.get("proposal")) + if has_proposal and not show_with: + continue + if not has_proposal and not show_without: + continue + self._table_infos.append(info) + + self._populate_table() + self._update_selection_state() + + def _populate_table(self): + # Clear table before populating, this keeps headers intact + self._table.setRowCount(0) + # Refill table + self._table.setRowCount(len(self._table_infos)) + for row, info in enumerate(self._table_infos): + pgroup = info.get("pgroup", "") + title = info.get("title", "") + name = format_name(info) + start, end = format_schedule(info.get("schedule")) + + self._table.setItem(row, 0, QTableWidgetItem(pgroup)) + self._table.setItem(row, 1, QTableWidgetItem(title)) + self._table.setItem(row, 2, QTableWidgetItem(name)) + self._table.setItem(row, 3, QTableWidgetItem(start)) + self._table.setItem(row, 4, QTableWidgetItem(end)) + + width = self._table.viewport().width() + self._table.resizeRowsToContents() + self._table.resize(width, self._table.height()) + # self._table.resizeRowsToContents() + + @SafeSlot() + def _update_selection_state(self): + has_selection = False + if self._tabs.currentWidget() is not self._table_tab: + return + index = self._table.selectionModel().selectedRows() + if len(index) > 0: + index = index[0] + self._side_card.set_experiment_info(self._table_infos[index.row()]) + + def _emit_selected_experiment(self): + if self._tabs.currentWidget() is self._card_tab: + self.experiment_selected.emit(self._next_experiment) + logger.info(f"Emitting next experiment signal with info: {self._next_experiment}") + return + selected = self._table.selectionModel().selectedRows() + if not selected: + return + row = selected[0].row() + if 0 <= row < len(self._table_infos): + self.experiment_selected.emit(self._table_infos[row]) + logger.info(f"Emitting next experiment signal with info: {self._table_infos[row]}") + + def _select_next_experiment(self, experiment_infos: list[dict]) -> dict | None: + candidates = [] + for info in experiment_infos: + start, _ = format_schedule(info.get("schedule"), as_datetime=True) + if start is None: + continue + candidates.append((start, info)) + + if not candidates: + return experiment_infos[0] if experiment_infos else None + + now = datetime.now() + future = [entry for entry in candidates if entry[0] >= now] + pool = future or candidates + return min(pool, key=lambda entry: abs(entry[0] - now))[1] + + def _on_tab_changed(self, index): + if self._tabs.widget(index) is self._table_tab: + self._table.resizeRowsToContents() + self._side_card.set_experiment_info(self._next_experiment) + self._apply_table_filters() + + def _get_column_data(self, row) -> dict[str, str]: + output = {} + for ii, header in enumerate(self._headers.values()): + item = self._table.item(row, ii) + if item is None: + output[header] = "" + continue + output[header] = item.text() + return output + + @SafeSlot(str) + def _apply_row_filter(self, text_input: str): + """Apply a filter to the table rows based on the filter text.""" + if not text_input: + for row in self._hidden_rows: + self._table.setRowHidden(row, False) + self._hidden_rows.clear() + return + for row in range(self._table.rowCount()): + experiment_data = self._get_column_data(row) + if is_match( + text_input, experiment_data, list(self._headers.values()), self._enable_fuzzy_search + ): + self._table.setRowHidden(row, False) + self._hidden_rows.discard(row) + else: + self._table.setRowHidden(row, True) + self._hidden_rows.add(row) + + @SafeSlot(int) + def _state_change_fuzzy_search(self, enabled: int): + """Handle state changes for the fuzzy search toggle.""" + self._enable_fuzzy_search = not bool(enabled) + # Re-apply filter with updated fuzzy search setting + current_text = self.search_input.text() + self._apply_row_filter(current_text) + + +if __name__ == "__main__": + from qtpy.QtWidgets import QApplication + + experiment_infos = [ + { + "_id": "p22622", + "owner_groups": ["admin"], + "access_groups": ["unx-sls_xda_bs", "p22622"], + "realm_id": "TestBeamline", + "proposal": "12345967", + "title": "Test Experiment for Mat Card Widget", + "firstname": "John", + "lastname": "Doe", + "email": "john.doe@psi.ch", + "account": "doe_j", + "pi_firstname": "Jane", + "pi_lastname": "Smith", + "pi_email": "jane.smith@psi.ch", + "pi_account": "smith_j", + "eaccount": "e22622", + "pgroup": "p22622", + "abstract": "This is a test abstract for the experiment mat card widget. It should be long enough to test text wrapping and display in the card. The abstract provides a brief overview of the experiment, its goals, and its significance. This text is meant to simulate a real abstract that might be associated with an experiment in the BEC Atlas system. The card should be able to handle abstracts of varying lengths without any issues, ensuring that the user can read the full abstract even if it is quite long.", + "schedule": [{"start": "01/01/2025 08:00:00", "end": "03/01/2025 18:00:00"}], + "proposal_submitted": "15/12/2024", + "proposal_expire": "31/12/2025", + "proposal_status": "Scheduled", + "delta_last_schedule": 30, + "mainproposal": "", + }, + { + "_id": "p22623", + "owner_groups": ["admin"], + "access_groups": ["unx-sls_xda_bs", "p22623"], + "realm_id": "TestBeamline", + "proposal": "", + "title": "Experiment without Proposal", + "firstname": "Alice", + "lastname": "Johnson", + "email": "alice.johnson@psi.ch", + "account": "johnson_a", + "pi_firstname": "Bob", + "pi_lastname": "Brown", + "pi_email": "bob.brown@psi.ch", + "pi_account": "brown_b", + "eaccount": "e22623", + "pgroup": "p22623", + "abstract": "", + "schedule": [], + "proposal_submitted": "", + "proposal_expire": "", + "proposal_status": "", + "delta_last_schedule": None, + "mainproposal": "", + }, + ] + + app = QApplication([]) + from bec_qthemes import apply_theme + + from bec_widgets.widgets.utility.visual.dark_mode_button.dark_mode_button import DarkModeButton + + apply_theme("light") + w = QWidget() + l = QVBoxLayout(w) + dark_button = DarkModeButton() + l.addWidget(dark_button) + widget = ExperimentSelection(experiment_infos) + l.addWidget(widget) + w.resize(1280, 920) + w.show() + app.exec() diff --git a/bec_widgets/widgets/services/bec_atlas_admin_view/experiment_selection/utils.py b/bec_widgets/widgets/services/bec_atlas_admin_view/experiment_selection/utils.py new file mode 100644 index 000000000..37eab4ba3 --- /dev/null +++ b/bec_widgets/widgets/services/bec_atlas_admin_view/experiment_selection/utils.py @@ -0,0 +1,67 @@ +"""Utility functions for experiment selection.""" + +from datetime import datetime +from typing import Literal + +from bec_lib.messages import ExperimentInfoMessage + + +def format_name(info: dict | ExperimentInfoMessage) -> str: + """Format the name from the experiment info.""" + info = ExperimentInfoMessage.model_validate(info) if isinstance(info, dict) else info + firstname = info.firstname + lastname = info.lastname + return " ".join(part for part in [firstname, lastname] if part) + + +def format_schedule( + schedule: list[dict[Literal["start", "end"], str]] | None, as_datetime: bool = False +) -> tuple[str, str] | tuple[datetime | None, datetime | None]: + """Format the schedule information to display start and end times.""" + if not schedule: + return (None, None) if as_datetime else ("", "") + start, end = _pick_schedule_entry(schedule) + if as_datetime: + return start, end + return format_datetime(start), format_datetime(end) + + +def _pick_schedule_entry( + schedule: list[dict[Literal["start", "end"], str]], +) -> tuple[datetime | None, datetime | None]: + """Pick the most relevant schedule entry based on the current time.""" + now = datetime.now() + candidates = [] + for item in schedule: + if not item: + continue + start_raw = item.get("start") + parsed = _parse_schedule_start(start_raw) + if parsed is None: + continue + candidates.append((parsed, item)) + + if not candidates: + return None, None + + future = [entry for entry in candidates if entry[0] >= now] + pool = future or candidates + chosen_start, chosen_item = min(pool, key=lambda entry: abs(entry[0] - now)) + end_raw = chosen_item.get("end") + return chosen_start, _parse_schedule_start(end_raw) + + +def _parse_schedule_start(value) -> datetime | None: + """Parse a schedule start string into a datetime object.""" + if not value: + return None + try: + return datetime.strptime(value, "%d/%m/%Y %H:%M:%S") + except ValueError: + return None + + +def format_datetime(value) -> str: + if not value: + return "" + return value.strftime("%Y-%m-%d %H:%M") diff --git a/bec_widgets/widgets/services/bec_atlas_admin_view/login_dialog.py b/bec_widgets/widgets/services/bec_atlas_admin_view/login_dialog.py new file mode 100644 index 000000000..99e860490 --- /dev/null +++ b/bec_widgets/widgets/services/bec_atlas_admin_view/login_dialog.py @@ -0,0 +1,70 @@ +"""Login dialog for user authentication.""" + +from bec_qthemes import apply_theme +from qtpy.QtCore import Qt, Signal +from qtpy.QtWidgets import QDialog, QLabel, QLineEdit, QPushButton, QVBoxLayout + + +class LoginDialog(QDialog): + credentials_entered = Signal(str, str) + + def __init__(self, parent=None): + super().__init__(parent=parent) + self.setWindowTitle("Login") + self.setModal(True) + self.setFixedWidth(320) + + # Slightly increased padding + self.setStyleSheet( + """ + QLineEdit { + padding: 8px; + } + """ + ) + + title = QLabel("Sign in", parent=self) + title.setAlignment(Qt.AlignmentFlag.AlignCenter) + title.setStyleSheet("font-size: 18px; font-weight: 600;") + + self.username = QLineEdit(parent=self) + self.username.setPlaceholderText("Username") + + self.password = QLineEdit(parent=self) + self.password.setPlaceholderText("Password") + self.password.setEchoMode(QLineEdit.Password) + + self.ok_btn = QPushButton("Sign in", parent=self) + self.ok_btn.setDefault(True) + self.ok_btn.clicked.connect(self._emit_credentials) + + layout = QVBoxLayout(self) + layout.setContentsMargins(32, 32, 32, 32) + layout.setSpacing(16) + + layout.addWidget(title) + layout.addSpacing(8) + layout.addWidget(self.username) + layout.addWidget(self.password) + layout.addSpacing(12) + layout.addWidget(self.ok_btn) + + self.username.setFocus() + + def _emit_credentials(self): + self.credentials_entered.emit(self.username.text().strip(), self.password.text()) + self.accept() + + +if __name__ == "__main__": + import sys + + from qtpy.QtWidgets import QApplication + + app = QApplication(sys.argv) + apply_theme("light") + + dialog = LoginDialog() + + dialog.credentials_entered.connect(lambda u, p: print(f"Username: {u}, Password: {p}")) + dialog.exec_()