From 941f3a9535eb2b8b370b63ae2da1b585dd711089 Mon Sep 17 00:00:00 2001 From: appel_c Date: Tue, 17 Feb 2026 17:42:27 +0100 Subject: [PATCH] refactor: cleanup widgets --- bec_widgets/applications/main_app.py | 1 + .../views/admin_view/admin_widget.py | 15 +- .../bec_atlas_admin_view.py | 432 +++++++++++++----- .../bec_atlas_http_service.py | 330 +++++++++---- .../experiment_mat_card.py | 2 + .../experiment_selection.py | 20 +- 6 files changed, 588 insertions(+), 212 deletions(-) diff --git a/bec_widgets/applications/main_app.py b/bec_widgets/applications/main_app.py index d41f7830..b4069603 100644 --- a/bec_widgets/applications/main_app.py +++ b/bec_widgets/applications/main_app.py @@ -85,6 +85,7 @@ class BECMainApp(BECMainWindow): widget=self.admin_view, id="admin_view", mini_text="Admin", + from_top=False, ) if self._show_examples: diff --git a/bec_widgets/applications/views/admin_view/admin_widget.py b/bec_widgets/applications/views/admin_view/admin_widget.py index 75e6f1fa..35b1cec4 100644 --- a/bec_widgets/applications/views/admin_view/admin_widget.py +++ b/bec_widgets/applications/views/admin_view/admin_widget.py @@ -2,19 +2,10 @@ from __future__ import annotations -from typing import TYPE_CHECKING - -from bec_lib.endpoints import MessageEndpoints -from bec_lib.messages import DeploymentInfoMessage -from qtpy.QtCore import Qt -from qtpy.QtWidgets import QSizePolicy, QStackedLayout, QVBoxLayout, QWidget +from qtpy.QtWidgets import QVBoxLayout, QWidget from bec_widgets.utils.bec_widget import BECWidget -from bec_widgets.utils.error_popups import SafeSlot from bec_widgets.widgets.services.bec_atlas_admin_view.bec_atlas_admin_view import BECAtlasAdminView -from bec_widgets.widgets.services.bec_atlas_admin_view.experiment_selection.experiment_selection import ( - ExperimentSelection, -) class AdminWidget(BECWidget, QWidget): @@ -31,14 +22,14 @@ class AdminWidget(BECWidget, QWidget): 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__": +# pylint: disable=ungrouped-imports +if __name__ == "__main__": # pragma: no cover from bec_qthemes import apply_theme from qtpy.QtWidgets import QApplication 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 index 4dddf61d..bf631d50 100644 --- 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 @@ -2,28 +2,32 @@ from __future__ import annotations -import json -from typing import TYPE_CHECKING +import time 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.QtCore import Qt, QTimer, Signal from qtpy.QtWidgets import ( + QFrame, + QGroupBox, QHBoxLayout, - QPushButton, + QLabel, QSizePolicy, QStackedLayout, QVBoxLayout, QWidget, ) +from bec_widgets.utils.bec_login import BECLogin 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.actions import MaterialIconAction, WidgetAction 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 ( + ATLAS_ENPOINTS, + AuthenticatedUserInfo, BECAtlasHTTPService, HTTPResponse, ) @@ -34,54 +38,205 @@ from bec_widgets.widgets.services.bec_atlas_admin_view.experiment_selection.expe ExperimentSelection, ) -if TYPE_CHECKING: # pragma: no cover - from bec_lib.messages import ExperimentInfoMessage - logger = bec_logger.logger -class OverviewWidget(QWidget): +class OverviewWidget(QGroupBox): """Overview Widget for the BEC Atlas Admin view""" - login_requested = Signal() + login_requested = Signal(str, str) + 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.login_requested.emit() + self.setContentsMargins(12, 0, 12, 6) + self._authenticated = False + # Root layout + self.root_layout = QVBoxLayout(self) + self.root_layout.setContentsMargins(0, 0, 0, 0) + self.root_layout.setSpacing(0) + + # Stacked Layout to switch between login form and overview content + 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) + + self._init_login_view() + self._init_experiment_overview() + self.stacked_layout.setCurrentWidget(self._login_widget) + self._experiment_overview_widget.setVisible(False) + + def set_experiment_info(self, experiment_info: ExperimentInfoMessage): + """Set the experiment information for the overview widget.""" + self._experiment_overview_widget.set_experiment_info(experiment_info) + + @SafeSlot(bool) + def set_authenticated(self, authenticated: bool): + """Set the authentication state of the overview widget.""" + self._authenticated = authenticated + if authenticated: + self.stacked_layout.setCurrentWidget(self._experiment_overview_widget) + self._experiment_overview_widget.setVisible(True) + else: + self.stacked_layout.setCurrentWidget(self._login_widget) + self._experiment_overview_widget.setVisible(False) + + def _init_login_view(self): + """Initialize the login view.""" + self._login_widget = QWidget() + layout = QHBoxLayout(self._login_widget) + self._login_widget.setAutoFillBackground(True) + self._login_widget.setSizePolicy(QSizePolicy.Policy.Expanding, QSizePolicy.Policy.Expanding) + layout.setSpacing(16) + + content = QFrame() + content_layout = QVBoxLayout(content) + content.setFrameShape(QFrame.Shape.StyledPanel) + content.setFrameShadow(QFrame.Shadow.Raised) + content.setStyleSheet( + """ + QFrame + { + border: 1px solid #cccccc; + } + QLabel + { + border: none; + } + """ + ) + content_layout.setAlignment(Qt.AlignmentFlag.AlignCenter) + content.setFixedSize(400, 280) + + self._login = BECLogin(parent=self) + self._login.credentials_entered.connect(self.login_requested.emit) + content_layout.addWidget(self._login) + layout.addWidget(content) + self.stacked_layout.addWidget(self._login_widget) + + def _init_experiment_overview(self): + """Initialize the experiment overview content.""" + self._experiment_overview_widget = ExperimentMatCard( + show_activate_button=True, + parent=self, + title="Current Experiment", + button_text="Change Experiment", + ) + self._experiment_overview_widget.experiment_selected.connect(self._on_experiment_selected) + layout = QVBoxLayout(self._experiment_overview_widget) + self._experiment_overview_widget.setAutoFillBackground(True) + self._experiment_overview_widget.setSizePolicy( + QSizePolicy.Policy.Expanding, QSizePolicy.Policy.Expanding + ) + layout.setSpacing(16) + layout.setAlignment(Qt.AlignmentFlag.AlignCenter) + self.stacked_layout.addWidget(self._experiment_overview_widget) @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) + def _on_experiment_selected(self, _): + """Handle the change experiment button click.""" + self.change_experiment_requested.emit() + + +class CustomLogoutAction(MaterialIconAction): + """Custom logout action that can be enabled/disabled based on authentication state.""" + + def __init__(self, parent=None): + super().__init__( + icon_name="logout", + tooltip="Logout", + label_text="Logout", + text_position="under", + parent=parent, + filled=True, + ) + self.action.setEnabled(False) # Initially disabled until authenticated + self._tick_timer = QTimer(parent) + self._tick_timer.setInterval(1000) + self._tick_timer.timeout.connect(self._on_tick) + self._login_remaining_s = 0 + + def set_authenticated(self, auth_info: AuthenticatedUserInfo | None): + """Enable or disable the logout action based on authentication state.""" + if not auth_info: + self._tick_timer.stop() + self._login_remaining_s = 0 + self.action.setEnabled(False) + self.update_label() # Reset Label text + return # No need to set the timer if we're not authenticated + self._login_remaining_s = max(0, int(auth_info.exp - time.time())) if auth_info else 0 + self.action.setEnabled(True) + if self._login_remaining_s > 0: + self._tick_timer.start() + + def _on_tick(self) -> None: + """Handle the timer countdown tick to update the remaining logout time.""" + self._login_remaining_s -= 1 + if self._login_remaining_s <= 0: + self.set_authenticated(None) # This will disable the action and stop the timer + return + + self.update_label() # Optionally update the label to show remaining time + + def update_label(self): + """Update the label text of the logout action.""" + if self._login_remaining_s > 0: + label_text = f"{self.label_text} ({self._login_remaining_s}s)" + else: + label_text = self.label_text + self.action.setText(label_text) + + def cleanup(self): + """Cleanup the timer when the action is destroyed.""" + if self._tick_timer.isActive(): + self._tick_timer.stop() + + +class AtlasConnectionInfo(QWidget): + + def __init__(self, parent=None): + super().__init__(parent=parent) + layout = QVBoxLayout(self) + layout.setAlignment(Qt.AlignmentFlag.AlignRight | Qt.AlignmentFlag.AlignVCenter) + self.setSizePolicy(QSizePolicy.Policy.Expanding, QSizePolicy.Policy.Expanding) + layout.setContentsMargins(6, 6, 6, 6) + layout.setSpacing(8) + self._bl_info_label = QLabel(self) + self._atlas_url_label = QLabel(self) + layout.addWidget(self._bl_info_label) + layout.addWidget(self._atlas_url_label) + self._atlas_url_text = "" + + def set_info(self, realm_id: str, bl_name: str, atlas_url: str): + """Set the connection information for the BEC Atlas API.""" + bl_info = f"{realm_id} @ {bl_name}" + self._bl_info_label.setText(bl_info) + self._atlas_url_label.setText(atlas_url) + self._atlas_url_text = atlas_url + + def set_logged_in(self, email: str): + """Show login status in the atlas info widget.""" + self._atlas_url_label.setText(f"{self._atlas_url_text} | {email}") + + def clear_login(self): + """Clear login status from the atlas info widget.""" + self._atlas_url_label.setText(self._atlas_url_text) 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 + self, + parent=None, + atlas_url: str = "http://localhost/api/v1", + client=None, # "https://bec-atlas-dev.psi.ch/api/v1", client=None ): + super().__init__(parent=parent, client=client) # State variables @@ -90,6 +245,7 @@ class BECAtlasAdminView(BECWidget, QWidget): self._current_session_info = None self._current_experiment_info = None self._authenticated = False + self._atlas_url = atlas_url # Root layout self.root_layout = QVBoxLayout(self) @@ -100,7 +256,7 @@ class BECAtlasAdminView(BECWidget, QWidget): self.toolbar = ModularToolBar(self) self.init_toolbar() self.root_layout.insertWidget(0, self.toolbar) - self.toolbar.show_bundles(["view", "auth"]) + self.toolbar.show_bundles(["view", "atlas_info", "auth"]) # Stacked layout to switch between overview, experiment selection and messaging services # It is added below the toolbar @@ -110,23 +266,28 @@ class BECAtlasAdminView(BECWidget, QWidget): 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.login_requested.connect(self.login) - - # 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"} ) + # Overview widget + self.overview_widget = OverviewWidget(parent=self) + self.stacked_layout.addWidget(self.overview_widget) + + # Experiment Selection widget + self.experiment_selection = ExperimentSelection(parent=self) + self.experiment_selection.setVisible(False) + self.stacked_layout.addWidget(self.experiment_selection) + # Connect signals - self.atlas_http_service.http_response_received.connect(self._on_http_response_received) + self.overview_widget.login_requested.connect(self._on_login_requested) + self.overview_widget.change_experiment_requested.connect( + self._on_experiment_selection_selected + ) + self.authenticated.connect(self.overview_widget.set_authenticated) + self.experiment_selection.experiment_selected.connect(self._on_experiment_selected) + self.atlas_http_service.http_response.connect(self._on_http_response_received) self.atlas_http_service.authenticated.connect(self._on_authenticated) self.bec_dispatcher.connect_slot( @@ -135,27 +296,6 @@ class BECAtlasAdminView(BECWidget, QWidget): from_start=True, ) - @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 @@ -196,27 +336,21 @@ class BECAtlasAdminView(BECWidget, QWidget): 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) + # Atlas Info + self._atlas_info_widget = AtlasConnectionInfo(parent=self) + atlas_info = WidgetAction(widget=self._atlas_info_widget, parent=self) + self.toolbar.components.add_safe("atlas_info", atlas_info) # Logout - logout_action = MaterialIconAction( - icon_name="logout", - tooltip="Logout", - label_text="Logout", - text_position="under", - parent=self, - filled=True, - ) + # logout_action = MaterialIconAction( + # icon_name="logout", + # tooltip="Logout", + # label_text="Logout", + # text_position="under", + # parent=self, + # filled=True, + # ) + logout_action = CustomLogoutAction(parent=self) logout_action.action.triggered.connect(self.logout) logout_action.action.setEnabled(False) # Initially disabled until authenticated self.toolbar.components.add_safe("logout", logout_action) @@ -228,12 +362,20 @@ class BECAtlasAdminView(BECWidget, QWidget): view_bundle.add_action("messaging_services") self.toolbar.add_bundle(view_bundle) + # Add atlas_info to toolbar + atlas_info_bundle = ToolbarBundle("atlas_info", self.toolbar.components) + atlas_info_bundle.add_action("atlas_info") + self.toolbar.add_bundle(atlas_info_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) + ######################## + ## Toolbar icon slots + ######################## + def _on_overview_selected(self): """Show the overview panel.""" self.overview_widget.setVisible(True) @@ -260,6 +402,45 @@ class BECAtlasAdminView(BECWidget, QWidget): # self.overview_widget.setVisible(False) # self.experiment_selection.setVisible(False) + ######################## + ## Internal slots + ######################## + + @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(str, str, popup_error=True) + def _on_login_requested(self, username: str, password: str): + """Handle login requested signal from the overview widget.""" + # Logout first to clear any existing session and cookies before attempting a new login + if self._authenticated: + logger.info("Existing session detected, logging out before attempting new login.") + self.logout() + # Now login with new credentials + self.login(username, password) + + @SafeSlot(dict, dict) + def _update_deployment_info(self, msg: dict, _: 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) + + self._atlas_info_widget.set_info( + realm_id=self._current_experiment_info.realm_id or "", + bl_name=self._current_deployment_info.name or "", + atlas_url=self._atlas_url, + ) + self.atlas_http_service._set_current_deployment_info(deployment) + 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? @@ -276,51 +457,90 @@ class BECAtlasAdminView(BECWidget, QWidget): return self.atlas_http_service.get_experiments_for_realm(current_realm_id) + ######################## + ## HTTP Service response handling + ######################## + 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: + logger.debug( + f"HTTP Response received: {response.request_url} with status {response.status}" + ) + if ATLAS_ENPOINTS.REALMS_EXPERIMENTS in response.request_url: experiments = response.data if isinstance(response.data, list) else [] + # Filter experiments to only include those that the user has write access to self.experiment_selection.set_experiment_infos(experiments) - self._on_experiment_selection_selected() # Switch to experiment selection once experiments are loaded + # self._on_experiment_selection_selected() # Stick to overview + elif ATLAS_ENPOINTS.SET_EXPERIMENT in response.request_url: + self._on_overview_selected() # Reconsider this as the overview is now the login. + # Reconsider once queue is ready + # elif ATLAS_ENPOINTS.REALMS_EXPERIMENTS in response.request_url: - def _on_authenticated(self, authenticated: bool) -> None: + @SafeSlot(dict) + def _on_authenticated(self, auth_info: dict) -> None: """Handle authentication state change.""" + authenticated = False + # Only if the user has owner access to the deployment, we consider them to be fully authenticated + # This means that although they may authenticate against atlas, they won't be able to see any + # extra information here + if auth_info: + info = AuthenticatedUserInfo.model_validate(auth_info) + if ( + self._current_deployment_info + and info.deployment_id == self._current_deployment_info.deployment_id + ): + authenticated = True + else: + logger.warning( + f"Authenticated user {info.email} does not have access to the current deployment {self._current_deployment_info.name if self._current_deployment_info else ''}." + ) 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 + self._atlas_info_widget.set_logged_in(info.email) + self.toolbar.components.get_action("logout").set_authenticated(info) 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 + self._atlas_info_widget.clear_login() # Clear login status in atlas info widget on logout + self.toolbar.components.get_action("logout").set_authenticated(None) - @SafeSlot(dict) + ################ + ## API Methods + ################ + + @SafeSlot(dict, popup_error=True) 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() + @SafeSlot(str, str, popup_error=True) + def login(self, username: str, password: str) -> None: + """Login to the BEC Atlas API with the provided username and password.""" + self.atlas_http_service.login(username=username, password=password) + @SafeSlot(popup_error=True) def logout(self) -> None: """Logout from the BEC Atlas API.""" self.atlas_http_service.logout() + def get_user_info(self): + """Get the current user information from the BEC Atlas API.""" + self.atlas_http_service.get_user_info() + + ############### + ## Cleanup + ############### + def cleanup(self): self.atlas_http_service.cleanup() return super().cleanup() @@ -362,10 +582,14 @@ if __name__ == "__main__": "delta_last_schedule": 30, "mainproposal": "", } - from bec_lib.messages import ExperimentInfoMessage + from bec_lib.messages import DeploymentInfoMessage, ExperimentInfoMessage, SessionInfoMessage - proposal_info = ExperimentInfoMessage(**exp_info_dict) - window.set_experiment_info(proposal_info) + # proposal_info = ExperimentInfoMessage(**exp_info_dict) + # session_info = SessionInfoMessage(name="Test Session", experiment=proposal_info) + # deployment_info = DeploymentInfoMessage( + # deployment_id="test_deployment_001", active_session=session_info + # ) + # 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 index e4153032..96e10c2b 100644 --- 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 @@ -1,27 +1,90 @@ -import json +""" +This module is a QWidget-based HTTP service, responsible for interacting with the +BEC Atlas API using a QNetworkAccessManager. It prov""" +import json +import time +from enum import StrEnum +from typing import Literal + +import jwt +from bec_lib.logger import bec_logger +from bec_lib.messages import DeploymentInfoMessage from pydantic import BaseModel -from qtpy.QtCore import QUrl, QUrlQuery, Signal +from qtpy.QtCore import QObject, QTimer, 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 + +logger = bec_logger.logger + + +class ATLAS_ENPOINTS(StrEnum): + """Constants for BEC Atlas API endpoints.""" + + LOGIN = "/user/login" + LOGOUT = "/user/logout" + REALMS_EXPERIMENTS = "/realms/experiments" + SET_EXPERIMENT = "/deployments/experiment" + USER_INFO = "/user/me" + DEPLOYMENT_INFO = "/deployments/id" + + +class BECAtlasHTTPError(Exception): + """Custom exception for BEC Atlas HTTP errors.""" class HTTPResponse(BaseModel): + """Model representing an HTTP response.""" + request_url: str headers: dict status: int - data: dict | list | str + data: dict | list | str # Check with Klaus if str is deprecated + + +class AuthenticatedUserInfo(BaseModel): + """Model representing authenticated user information.""" + + email: str + exp: float + groups: set[str] + deployment_id: str + + +class AuthenticatedTimer(QObject): + """Timer to track authentication expiration and emit a signal when the token expires.""" + + expired = Signal() + + def __init__(self, parent=None): + super().__init__(parent) + self._timer = QTimer(self) + self._timer.setSingleShot(True) + self._timer.timeout.connect(self._on_expired) + + def start(self, duration_seconds: float): + """Start the timer with the given duration in seconds.""" + self._timer.start(int(duration_seconds * 1000)) + + def stop(self): + """Stop the timer.""" + self._timer.stop() + + @SafeSlot() + def _on_expired(self): + """Handle the timer expiration by emitting the expired signal.""" + logger.info("Authentication token has expired.") + self.expired.emit() 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) + http_response = Signal(dict) # HTTPResponse.model_dump() dict + authenticated = Signal(dict) # AuthenticatedUserInfo.model_dump() dict or {} + authentication_expires = Signal(float) def __init__(self, parent=None, base_url: str = "", headers: dict | None = None): super().__init__(parent) @@ -31,7 +94,50 @@ class BECAtlasHTTPService(QWidget): self._base_url = base_url self.network_manager = QNetworkAccessManager(self) self.network_manager.finished.connect(self._handle_response) - self._authenticated = False + self._auth_user_info: AuthenticatedUserInfo | None = None + self._auth_timer = self._create_auth_timer() + self._current_deployment_info = None + + def _create_auth_timer(self) -> AuthenticatedTimer: + """Create and connect the authenticated timer to handle token expiration.""" + timer = AuthenticatedTimer(self) + timer.expired.connect(self.__clear_login_info) + return timer + + @property + def auth_user_info(self) -> AuthenticatedUserInfo | None: + """Get the authenticated user information, including email and token expiration time.""" + return self._auth_user_info + + def __set_auth_info(self, login_info: dict[Literal["email", "exp"], str | float]): + """Set the authenticated user information after a successful login.""" + login_info.update({"groups": []}) # Initialize groups as empty until we fetch user info + login_info.update( + { + "deployment_id": ( + self._current_deployment_info.deployment_id + if self._current_deployment_info + else "" + ) + } + ) + self._auth_user_info = AuthenticatedUserInfo(**login_info) + # Start timer to clear auth info once token expires + exp_time = login_info.get("exp", 0) + current_time = time.time() # TODO should we use server time to avoid clock skew issues? + duration = max(0, exp_time - current_time) + self._auth_timer.start(duration) + + def __set_auth_groups(self, groups: list[str]): + """Set the authenticated user's groups after fetching user info.""" + if self._auth_user_info is not None: + self._auth_user_info.groups = set(groups) + + def __clear_login_info(self, skip_logout: bool = False): + """Clear the authenticated user information after logout.""" + self._auth_user_info = None + if not skip_logout: + self.logout() # Ensure we also logout on the server side and invalidate the session def closeEvent(self, event): self.cleanup() @@ -39,7 +145,7 @@ class BECAtlasHTTPService(QWidget): def cleanup(self): """Cleanup connection, destroy authenticate cookies.""" - + logger.info("Cleaning up BECAtlasHTTPService: disconnecting signals and clearing cookies.") # Disconnect signals to avoid handling responses after cleanup self.network_manager.finished.disconnect(self._handle_response) @@ -50,6 +156,7 @@ class BECAtlasHTTPService(QWidget): for cookie in self.network_manager.cookieJar().cookiesForUrl(QUrl(self._base_url)): self.network_manager.cookieJar().deleteCookie(cookie) + @SafeSlot(QNetworkReply, popup_error=True) def _handle_response(self, reply: QNetworkReply): """ Handle the HTTP response from the server. @@ -58,69 +165,77 @@ class BECAtlasHTTPService(QWidget): reply (QNetworkReply): The network reply object containing the response. """ status = reply.attribute(QNetworkRequest.Attribute.HttpStatusCodeAttribute) - raw_bytes = bytes(reply.readAll()) + 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) + # Any unsuccessful status code should raise here + if status != 200: + raise BECAtlasHTTPError( + f"HTTP request for {request_url} failed with status code {status} and response: {raw_bytes.decode('utf-8')}" + ) - # 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: + if len(raw_bytes) > 0: + data = json.loads(raw_bytes.decode()) + else: data = {} + if ATLAS_ENPOINTS.LOGIN.value in request_url: + # If it's a login response, don't forward the token + # but extract the expiration time and emit it + token = data.get("access_token") + data = jwt.decode(token, options={"verify_signature": False}) + self.authentication_expires.emit(data.get("exp", 0)) + # Now we set the auth info, and then fetch the user info to get the groups + self.__set_auth_info(data) + # Fetch information about the deployment info + self.get_user_info() # Fetch groups, then emit authenticated once groups are set on auth_user + elif ATLAS_ENPOINTS.LOGOUT.value in request_url: + self._auth_timer.stop() # Stop the timer if it was running + self.__clear_login_info(skip_logout=True) # Skip calling logout again + self.authenticated.emit({}) + elif ATLAS_ENPOINTS.USER_INFO.value in request_url: + groups = data.get("groups", []) + email = data.get("email", "") + # Second step of authentication: We also have all groups now + if self.auth_user_info is not None and self.auth_user_info.email == email: + self.__set_auth_groups(groups) + if self._current_deployment_info is not None: + # Now we need to fetch the deployment info to get the owner groups and check access rights, + # Then we can emit the authenticated signal with the full user info including groups if access is + # granted. Otherwise, we emit nothing and show a warning that the user does not have the access + # rights for the current deployment. + self.get_deployment_info( + deployment_id=self._current_deployment_info.deployment_id + ) + elif ATLAS_ENPOINTS.DEPLOYMENT_INFO.value in request_url: + owner_groups = data.get("owner_groups", []) + if self.auth_user_info is not None: + if not self.auth_user_info.groups.isdisjoint(owner_groups): + self.authenticated.emit(self.auth_user_info.model_dump()) + else: + self._show_warning( + text=f"User {self.auth_user_info.email} does not have access to the active deployment {data.get('name', '')}." + ) + self.logout() # Logout to clear auth info and stop timer since user does not have access + response = HTTPResponse(request_url=request_url, headers=headers, status=status, data=data) - self.http_response_received.emit(response.model_dump()) + self.http_response.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_warning(self, text: str): + """Show a warning message to the user.""" + msg = QMessageBox(self) + msg.setIcon(QMessageBox.Icon.Warning) + msg.setText(text) + msg.setWindowTitle("Authentication Warning") + msg.exec_() - 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 + ####################### + # GET/POST Request Methods + ####################### - 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): + def _get_request(self, endpoint: str, query_parameters: dict | None = None): """ GET request to the API endpoint. @@ -128,6 +243,7 @@ class BECAtlasHTTPService(QWidget): endpoint (str): The API endpoint to send the GET request to. query_parameters (dict | None): Optional query parameters to include in the URL. """ + logger.info(f"Sending GET request to {endpoint}.") url = QUrl(self._base_url + endpoint) if query_parameters: query = QUrlQuery() @@ -136,10 +252,10 @@ class BECAtlasHTTPService(QWidget): url.setQuery(query) request = QNetworkRequest(url) for key, value in self._headers.items(): - request.setRawHeader(key.encode("utf-8"), value.encode("utf-8")) + request.setRawHeader(key.encode(), value.encode()) self.network_manager.get(request) - def post_request( + def _post_request( self, endpoint: str, payload: dict | None = None, query_parameters: dict | None = None ): """ @@ -150,6 +266,7 @@ class BECAtlasHTTPService(QWidget): payload (dict): The JSON payload to include in the POST request. query_parameters (dict | None): Optional query parameters to include in the URL. """ + logger.info(f"Sending GET request to {endpoint}.") if payload is None: payload = {} url = QUrl(self._base_url + endpoint) @@ -162,40 +279,87 @@ class BECAtlasHTTPService(QWidget): # Headers for key, value in self._headers.items(): - request.setRawHeader(key.encode("utf-8"), value.encode("utf-8")) + request.setRawHeader(key.encode(), value.encode()) request.setHeader(QNetworkRequest.KnownHeaders.ContentTypeHeader, "application/json") - payload_dump = json.dumps(payload).encode("utf-8") - reply = self.network_manager.post(request, payload_dump) + payload_dump = json.dumps(payload).encode() + self.network_manager.post(request, payload_dump) + + def _set_current_deployment_info(self, deployment_info: dict | DeploymentInfoMessage): + """ + Set the current deployment information for the service. + + Args: + deployment_info (dict | DeploymentInfoMessage): The deployment information to set. + """ + if isinstance(deployment_info, dict): + deployment_info = DeploymentInfoMessage.model_validate(deployment_info) + self._current_deployment_info = deployment_info ################ # 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() + @SafeSlot(str, str, popup_error=True) + def login(self, username: str, password: str): + """ + Login to BEC Atlas with the provided username and password. + Args: + username (str): The username for authentication. + password (str): The password for authentication. + """ + self._post_request( + endpoint=ATLAS_ENPOINTS.LOGIN.value, + payload={"username": username, "password": password}, + ) + + @SafeSlot(popup_error=True) 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") + self._post_request(endpoint=ATLAS_ENPOINTS.LOGOUT.value) + @SafeSlot(str, popup_error=True) 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) + """ + Get the list of realms from BEC Atlas. Requires authentication. + + Args: + realm_id (str): The ID of the realm to retrieve experiments for. + """ + self._get_request( + endpoint=ATLAS_ENPOINTS.REALMS_EXPERIMENTS.value, + query_parameters={"realm_id": realm_id}, + ) @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", + """ + Set the current experiment information for the service. + + Args: + experiment_id (str): The ID of the experiment to set. + deployment_id (str): The ID of the deployment associated with the experiment. + """ + self._post_request( + endpoint=ATLAS_ENPOINTS.SET_EXPERIMENT.value, query_parameters={"experiment_id": experiment_id, "deployment_id": deployment_id}, ) + + @SafeSlot(popup_error=True) + def get_user_info(self): + """Get the current user information from BEC Atlas. Requires authentication.""" + self._get_request(endpoint=ATLAS_ENPOINTS.USER_INFO.value) + + @SafeSlot(str, popup_error=True) + def get_deployment_info(self, deployment_id: str): + """ + Get the deployment information for a given deployment ID. Requires authentication. + + Args: + deployment_id (str): The ID of the deployment to retrieve information for. + """ + self._get_request( + endpoint=ATLAS_ENPOINTS.DEPLOYMENT_INFO.value, + query_parameters={"deployment_id": deployment_id}, + ) 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 index 746e990b..ac885c9b 100644 --- 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 @@ -161,6 +161,7 @@ class ExperimentMatCard(BECWidget, QWidget): ) 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) @@ -198,6 +199,7 @@ class ExperimentMatCard(BECWidget, QWidget): 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): """ 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 index 8687559d..bbb99093 100644 --- 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 @@ -82,8 +82,8 @@ class ExperimentSelection(QWidget): self._table_infos: list[dict[str, Any]] = [] main_layout = QVBoxLayout(self) - main_layout.setContentsMargins(16, 16, 16, 16) - main_layout.setSpacing(12) + main_layout.setContentsMargins(0, 0, 0, 0) + # main_layout.setSpacing(12) main_layout.setAlignment(Qt.AlignmentFlag.AlignCenter) self.setSizePolicy(QSizePolicy.Policy.Expanding, QSizePolicy.Policy.Expanding) self.setAutoFillBackground(True) @@ -94,6 +94,7 @@ class ExperimentSelection(QWidget): 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) @@ -105,10 +106,6 @@ class ExperimentSelection(QWidget): # main_layout.addStretch() button_layout = QHBoxLayout() - self._select_button = QPushButton("Activate", self) - self._select_button.setEnabled(False) - self._select_button.clicked.connect(self._emit_selected_experiment) - button_layout.addWidget(self._select_button) main_layout.addLayout(button_layout) self._apply_table_filters() self.restore_default_view() @@ -226,7 +223,10 @@ class ExperimentSelection(QWidget): 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=False) + 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) @@ -235,7 +235,6 @@ class ExperimentSelection(QWidget): @SafeSlot(bool) # Overload for buttons def _apply_table_filters(self, *args, **kwargs): if self._tabs.currentWidget() is not self._table_tab: - self._select_button.setEnabled(True) return show_with = self._with_proposals.isChecked() @@ -279,16 +278,11 @@ class ExperimentSelection(QWidget): def _update_selection_state(self): has_selection = False if self._tabs.currentWidget() is not self._table_tab: - self._select_button.setEnabled(True) return index = self._table.selectionModel().selectedRows() - if not index: - has_selection = False if len(index) > 0: index = index[0] self._side_card.set_experiment_info(self._table_infos[index.row()]) - has_selection = True - self._select_button.setEnabled(has_selection) def _emit_selected_experiment(self): if self._tabs.currentWidget() is self._card_tab: