mirror of
https://github.com/bec-project/bec_widgets.git
synced 2026-04-10 02:30:54 +02:00
Compare commits
1 Commits
main
...
feature/me
| Author | SHA1 | Date | |
|---|---|---|---|
| de5a58a63b |
@@ -42,6 +42,9 @@ from bec_widgets.widgets.services.bec_atlas_admin_view.experiment_selection.expe
|
||||
from bec_widgets.widgets.services.bec_atlas_admin_view.experiment_selection.experiment_selection import (
|
||||
ExperimentSelection,
|
||||
)
|
||||
from bec_widgets.widgets.services.bec_messaging_config.bec_messaging_config_widget import (
|
||||
BECMessagingConfigWidget,
|
||||
)
|
||||
|
||||
if TYPE_CHECKING: # pragma: no cover
|
||||
from qtpy.QtWidgets import QToolBar
|
||||
@@ -104,7 +107,8 @@ class OverviewWidget(QGroupBox):
|
||||
content_layout = QVBoxLayout(content)
|
||||
content.setFrameShape(QFrame.Shape.StyledPanel)
|
||||
content.setFrameShadow(QFrame.Shadow.Raised)
|
||||
content.setStyleSheet("""
|
||||
content.setStyleSheet(
|
||||
"""
|
||||
QFrame
|
||||
{
|
||||
border: 1px solid #cccccc;
|
||||
@@ -113,7 +117,8 @@ class OverviewWidget(QGroupBox):
|
||||
{
|
||||
border: none;
|
||||
}
|
||||
""")
|
||||
"""
|
||||
)
|
||||
content_layout.setAlignment(Qt.AlignmentFlag.AlignCenter)
|
||||
content.setFixedSize(400, 280)
|
||||
|
||||
@@ -299,6 +304,11 @@ class BECAtlasAdminView(BECWidget, QWidget):
|
||||
self.experiment_selection.setVisible(False)
|
||||
self.stacked_layout.addWidget(self.experiment_selection)
|
||||
|
||||
# Messaging Services widget
|
||||
self.messaging_config_widget = BECMessagingConfigWidget(parent=self)
|
||||
self.messaging_config_widget.setVisible(False)
|
||||
self.stacked_layout.addWidget(self.messaging_config_widget)
|
||||
|
||||
# Connect signals
|
||||
self.overview_widget.login_requested.connect(self._on_login_requested)
|
||||
self.overview_widget.change_experiment_requested.connect(
|
||||
@@ -392,6 +402,7 @@ class BECAtlasAdminView(BECWidget, QWidget):
|
||||
"""Show the overview panel."""
|
||||
self.overview_widget.setVisible(True)
|
||||
self.experiment_selection.setVisible(False)
|
||||
self.messaging_config_widget.setVisible(False)
|
||||
self.stacked_layout.setCurrentWidget(self.overview_widget)
|
||||
|
||||
def _on_experiment_selection_selected(self):
|
||||
@@ -401,12 +412,20 @@ class BECAtlasAdminView(BECWidget, QWidget):
|
||||
return
|
||||
self.overview_widget.setVisible(False)
|
||||
self.experiment_selection.setVisible(True)
|
||||
self.messaging_config_widget.setVisible(False)
|
||||
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.")
|
||||
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)
|
||||
self.messaging_config_widget.setVisible(True)
|
||||
if self._current_deployment_info is not None:
|
||||
self.messaging_config_widget.populate_from_deployment(self._current_deployment_info)
|
||||
self.stacked_layout.setCurrentWidget(self.messaging_config_widget)
|
||||
|
||||
########################
|
||||
## Internal slots
|
||||
@@ -446,6 +465,7 @@ class BECAtlasAdminView(BECWidget, QWidget):
|
||||
atlas_url=self._atlas_url,
|
||||
)
|
||||
self.atlas_http_service._set_current_deployment_info(deployment)
|
||||
self.messaging_config_widget.populate_from_deployment(deployment)
|
||||
|
||||
def _fetch_available_experiments(self):
|
||||
"""Fetch the list of available experiments for the authenticated user."""
|
||||
@@ -501,9 +521,7 @@ class BECAtlasAdminView(BECWidget, QWidget):
|
||||
|
||||
if authenticated:
|
||||
self.toolbar.components.get_action("experiment_selection").action.setEnabled(True)
|
||||
self.toolbar.components.get_action("messaging_services").action.setEnabled(
|
||||
False
|
||||
) # TODO activate once messaging is added
|
||||
self.toolbar.components.get_action("messaging_services").action.setEnabled(True)
|
||||
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)
|
||||
|
||||
@@ -0,0 +1,315 @@
|
||||
"""Module for the BEC messaging configuration widget."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
|
||||
from qtpy.QtCore import Qt, QTimer, Signal # type: ignore[attr-defined]
|
||||
from qtpy.QtWidgets import (
|
||||
QGroupBox,
|
||||
QHBoxLayout,
|
||||
QLabel,
|
||||
QPushButton,
|
||||
QSizePolicy,
|
||||
QSplitter,
|
||||
QTabWidget,
|
||||
QVBoxLayout,
|
||||
QWidget,
|
||||
)
|
||||
|
||||
from bec_widgets.utils.colors import apply_theme
|
||||
from bec_widgets.widgets.services.bec_messaging_config.service_cards import (
|
||||
CardType,
|
||||
ScopeListWidget,
|
||||
card_from_service,
|
||||
make_card,
|
||||
)
|
||||
from bec_widgets.widgets.services.bec_messaging_config.service_scope_event_table import (
|
||||
ServiceScopeEventTableWidget,
|
||||
)
|
||||
|
||||
|
||||
class ServiceConfigPanel(QWidget):
|
||||
"""Panel that manages global and local service scopes for one service type.
|
||||
|
||||
Args:
|
||||
card_type (CardType): The service type used when adding new scope cards.
|
||||
parent (QWidget | None): The parent widget.
|
||||
"""
|
||||
|
||||
config_changed = Signal()
|
||||
|
||||
def __init__(self, card_type: CardType, parent: QWidget | None = None) -> None:
|
||||
super().__init__(parent)
|
||||
self._card_type: CardType = card_type
|
||||
|
||||
root = QVBoxLayout(self)
|
||||
root.setContentsMargins(12, 12, 12, 12)
|
||||
root.setSpacing(0)
|
||||
|
||||
splitter = QSplitter(Qt.Orientation.Vertical)
|
||||
|
||||
# ── Local settings box ────────────────────────────────────────────
|
||||
self._local_box = QGroupBox("Current Experiment")
|
||||
self._local_list = ScopeListWidget()
|
||||
self._local_list.cards_changed.connect(self.config_changed)
|
||||
self._local_add_btn = QPushButton("+ Add")
|
||||
self._local_add_btn.setFixedWidth(120)
|
||||
self._local_add_btn.clicked.connect(
|
||||
lambda: self._local_list.add_card(make_card(self._card_type))
|
||||
)
|
||||
local_layout = QVBoxLayout(self._local_box)
|
||||
local_layout.setContentsMargins(16, 16, 16, 16)
|
||||
local_layout.setSpacing(12)
|
||||
local_layout.addWidget(self._local_add_btn, 0, Qt.AlignmentFlag.AlignRight)
|
||||
local_layout.addWidget(self._local_list, 1)
|
||||
splitter.addWidget(self._local_box)
|
||||
|
||||
# ── Global settings box ───────────────────────────────────────────
|
||||
self._global_box = QGroupBox("All Experiments")
|
||||
self._global_list = ScopeListWidget()
|
||||
self._global_list.cards_changed.connect(self.config_changed)
|
||||
self._global_add_btn = QPushButton("+ Add")
|
||||
self._global_add_btn.setFixedWidth(120)
|
||||
self._global_add_btn.clicked.connect(
|
||||
lambda: self._global_list.add_card(make_card(self._card_type))
|
||||
)
|
||||
global_layout = QVBoxLayout(self._global_box)
|
||||
global_layout.setContentsMargins(16, 16, 16, 16)
|
||||
global_layout.setSpacing(12)
|
||||
global_layout.addWidget(self._global_add_btn, 0, Qt.AlignmentFlag.AlignRight)
|
||||
global_layout.addWidget(self._global_list, 1)
|
||||
splitter.addWidget(self._global_box)
|
||||
|
||||
splitter.setSizes([300, 300])
|
||||
root.addWidget(splitter, 1)
|
||||
|
||||
# ------------------------------------------------------------------
|
||||
def load_services(self, deployment_services: list, session_services: list) -> None:
|
||||
"""Populate both lists with services matching the panel service type."""
|
||||
self._clear_list(self._global_list)
|
||||
self._clear_list(self._local_list)
|
||||
for info in deployment_services:
|
||||
if getattr(info, "service_type", None) == self._card_type:
|
||||
self._global_list.add_card(card_from_service(info))
|
||||
for info in session_services:
|
||||
if getattr(info, "service_type", None) == self._card_type:
|
||||
self._local_list.add_card(card_from_service(info))
|
||||
|
||||
@staticmethod
|
||||
def _clear_list(list_widget: ScopeListWidget) -> None:
|
||||
"""Remove all cards from *list_widget*."""
|
||||
list_widget.clear_cards()
|
||||
|
||||
# ------------------------------------------------------------------
|
||||
def get_data(self) -> dict:
|
||||
"""Collect all card data from both the deployment and session lists."""
|
||||
return {
|
||||
"deployment": self._collect(self._global_list),
|
||||
"session": self._collect(self._local_list),
|
||||
}
|
||||
|
||||
@staticmethod
|
||||
def _collect(list_widget: ScopeListWidget) -> list[dict]:
|
||||
return [card.get_data() for card in list_widget.cards()]
|
||||
|
||||
|
||||
class BECMessagingConfigWidget(QWidget):
|
||||
"""Widget to configure SciLog, Signal, and MS Teams messaging services."""
|
||||
|
||||
def __init__(self, parent: QWidget | None = None) -> None:
|
||||
super().__init__(parent)
|
||||
self.setWindowTitle("BEC Messaging Configuration")
|
||||
self.setMinimumSize(540, 500)
|
||||
|
||||
root = QVBoxLayout(self)
|
||||
root.setContentsMargins(16, 16, 16, 16)
|
||||
root.setSpacing(12)
|
||||
|
||||
content_splitter = QSplitter(Qt.Orientation.Horizontal)
|
||||
|
||||
# ── Tab widget ────────────────────────────────────────────────────
|
||||
self._tabs = QTabWidget()
|
||||
|
||||
self._scilog_panel = ServiceConfigPanel("scilog")
|
||||
self._signal_panel = ServiceConfigPanel("signal")
|
||||
self._teams_panel = ServiceConfigPanel("teams")
|
||||
|
||||
for panel in (self._scilog_panel, self._signal_panel, self._teams_panel):
|
||||
panel.config_changed.connect(self._refresh_scope_event_table)
|
||||
|
||||
self._tabs.addTab(self._scilog_panel, "SciLog")
|
||||
self._tabs.addTab(self._signal_panel, "Signal")
|
||||
self._tabs.addTab(self._teams_panel, "MS Teams")
|
||||
|
||||
content_splitter.addWidget(self._tabs)
|
||||
|
||||
self._scope_event_table = ServiceScopeEventTableWidget(self)
|
||||
content_splitter.addWidget(self._scope_event_table)
|
||||
content_splitter.setStretchFactor(0, 3)
|
||||
content_splitter.setStretchFactor(1, 2)
|
||||
|
||||
root.addWidget(content_splitter, 1)
|
||||
|
||||
# ── Bottom action bar ─────────────────────────────────────────────
|
||||
bottom_row = QHBoxLayout()
|
||||
bottom_row.setSpacing(12)
|
||||
|
||||
self._status_label = QLabel("")
|
||||
self._status_label.setSizePolicy(QSizePolicy.Policy.Expanding, QSizePolicy.Policy.Preferred)
|
||||
bottom_row.addWidget(self._status_label, 1)
|
||||
|
||||
save_btn = QPushButton("Save && Apply")
|
||||
save_btn.setDefault(True)
|
||||
save_btn.clicked.connect(self._mock_save_to_atlas_api)
|
||||
bottom_row.addWidget(save_btn)
|
||||
|
||||
root.addLayout(bottom_row)
|
||||
|
||||
# ------------------------------------------------------------------
|
||||
# Initialisation from backend message
|
||||
# ------------------------------------------------------------------
|
||||
|
||||
def populate_from_deployment(self, msg: DeploymentInfoMessage) -> None:
|
||||
"""Populate all panels from a deployment info message.
|
||||
|
||||
Args:
|
||||
msg (DeploymentInfoMessage): Deployment information containing deployment and session services.
|
||||
"""
|
||||
deployment_services = list(msg.messaging_services)
|
||||
session_services = (
|
||||
list(msg.active_session.messaging_services) if msg.active_session is not None else []
|
||||
)
|
||||
self._scilog_panel.load_services(deployment_services, session_services)
|
||||
self._signal_panel.load_services(deployment_services, session_services)
|
||||
self._teams_panel.load_services(deployment_services, session_services)
|
||||
self._refresh_scope_event_table()
|
||||
|
||||
# ------------------------------------------------------------------
|
||||
# Dummy REST methods (replace with real requests calls later)
|
||||
# ------------------------------------------------------------------
|
||||
|
||||
def _build_payload(self) -> dict:
|
||||
"""Collect the current UI state as a serializable dictionary."""
|
||||
return {
|
||||
"scilog": self._scilog_panel.get_data(),
|
||||
"signal": self._signal_panel.get_data(),
|
||||
"teams": self._teams_panel.get_data(),
|
||||
"event_subscriptions": self._scope_event_table.get_data(),
|
||||
}
|
||||
|
||||
def _refresh_scope_event_table(self) -> None:
|
||||
"""Refresh the event subscription table from the current service cards."""
|
||||
self._scope_event_table.set_services(self._collect_services_for_event_table())
|
||||
|
||||
def _collect_services_for_event_table(self) -> list[dict]:
|
||||
"""Collect all configured services for the event subscription table."""
|
||||
service_rows: list[dict] = []
|
||||
for panel in (self._scilog_panel, self._signal_panel, self._teams_panel):
|
||||
panel_data = panel.get_data()
|
||||
for source_name in ("deployment", "session"):
|
||||
for service in panel_data[source_name]:
|
||||
service_rows.append({**service, "source": source_name})
|
||||
return service_rows
|
||||
|
||||
def _mock_save_to_atlas_api(self) -> None:
|
||||
"""Simulate saving the current configuration to Atlas."""
|
||||
payload = self._build_payload()
|
||||
print("─" * 60)
|
||||
print("[BECMessagingConfigWidget] _mock_save_to_atlas_api payload:")
|
||||
print(json.dumps(payload, indent=2))
|
||||
print("─" * 60)
|
||||
self._set_status("✅ Saved!", timeout_ms=4000)
|
||||
|
||||
# ------------------------------------------------------------------
|
||||
# Status bar helper
|
||||
# ------------------------------------------------------------------
|
||||
|
||||
def _set_status(self, message: str, *, timeout_ms: int = 0) -> None:
|
||||
"""Show a status message and optionally clear it after a timeout.
|
||||
|
||||
Args:
|
||||
message (str): The message to display in the status label.
|
||||
timeout_ms (int): Time in milliseconds before clearing the message.
|
||||
"""
|
||||
self._status_label.setText(message)
|
||||
if timeout_ms > 0:
|
||||
QTimer.singleShot(timeout_ms, lambda: self._status_label.setText(""))
|
||||
|
||||
|
||||
if __name__ == "__main__": # pragma: no cover
|
||||
import sys
|
||||
|
||||
from bec_lib.messages import (
|
||||
DeploymentInfoMessage,
|
||||
MessagingConfig,
|
||||
MessagingServiceScopeConfig,
|
||||
SciLogServiceInfo,
|
||||
SessionInfoMessage,
|
||||
SignalServiceInfo,
|
||||
TeamsServiceInfo,
|
||||
)
|
||||
from qtpy.QtWidgets import QApplication
|
||||
|
||||
app = QApplication(sys.argv)
|
||||
apply_theme("dark")
|
||||
|
||||
# ── Build a realistic mock DeploymentInfoMessage ──────────────────
|
||||
mock_deployment = DeploymentInfoMessage(
|
||||
deployment_id="dep-0001",
|
||||
name="mockup-beamline",
|
||||
messaging_config=MessagingConfig(
|
||||
signal=MessagingServiceScopeConfig(enabled=True),
|
||||
teams=MessagingServiceScopeConfig(enabled=True),
|
||||
scilog=MessagingServiceScopeConfig(enabled=True),
|
||||
),
|
||||
messaging_services=[
|
||||
SciLogServiceInfo(
|
||||
id="sl-global-1",
|
||||
scope="beamline",
|
||||
enabled=True,
|
||||
name="Beamline Log",
|
||||
logbook_id="lb-99001",
|
||||
),
|
||||
TeamsServiceInfo(
|
||||
id="teams-global-1",
|
||||
scope="beamline",
|
||||
enabled=True,
|
||||
name="BEC Channel",
|
||||
workflow_webhook_url="https://outlook.office.com/webhook/…",
|
||||
),
|
||||
SignalServiceInfo(
|
||||
id="signal-global-1",
|
||||
scope="beamline",
|
||||
enabled=False,
|
||||
name=None,
|
||||
group_id=None,
|
||||
group_link=None,
|
||||
),
|
||||
],
|
||||
active_session=SessionInfoMessage(
|
||||
name="session-2026-03-07",
|
||||
messaging_services=[
|
||||
SciLogServiceInfo(
|
||||
id="sl-local-1",
|
||||
scope="experiment",
|
||||
enabled=True,
|
||||
name="My Notebook",
|
||||
logbook_id="lb-12345",
|
||||
),
|
||||
SignalServiceInfo(
|
||||
id="signal-local-1",
|
||||
scope="experiment",
|
||||
enabled=True,
|
||||
name="Lab Signal Group",
|
||||
group_id="grp-8a3f291c",
|
||||
group_link="https://signal.group/#grp-8a3f291c",
|
||||
),
|
||||
],
|
||||
),
|
||||
)
|
||||
|
||||
widget = BECMessagingConfigWidget()
|
||||
widget.populate_from_deployment(mock_deployment)
|
||||
widget.show()
|
||||
sys.exit(app.exec())
|
||||
@@ -0,0 +1,429 @@
|
||||
"""Module for service scope cards used by the messaging configuration widget."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import uuid
|
||||
from enum import IntEnum
|
||||
from typing import TYPE_CHECKING, Literal, Type
|
||||
|
||||
from bec_qthemes import material_icon
|
||||
from qtpy.QtCore import QRegularExpression, Qt, QTimer, Signal # type: ignore[attr-defined]
|
||||
from qtpy.QtGui import QRegularExpressionValidator
|
||||
from qtpy.QtWidgets import (
|
||||
QCheckBox,
|
||||
QFrame,
|
||||
QHBoxLayout,
|
||||
QLabel,
|
||||
QLineEdit,
|
||||
QPushButton,
|
||||
QScrollArea,
|
||||
QSizePolicy,
|
||||
QSpacerItem,
|
||||
QStackedLayout,
|
||||
QToolButton,
|
||||
QVBoxLayout,
|
||||
QWidget,
|
||||
)
|
||||
|
||||
if TYPE_CHECKING: # pragma: no cover
|
||||
from bec_lib import messages
|
||||
|
||||
CardType = Literal["scilog", "signal", "teams"]
|
||||
|
||||
|
||||
class ScopeListWidget(QScrollArea):
|
||||
"""A scrollable list that stacks scope cards neatly at the top."""
|
||||
|
||||
cards_changed = Signal()
|
||||
|
||||
def __init__(self, parent: QWidget | None = None) -> None:
|
||||
super().__init__(parent)
|
||||
self.setWidgetResizable(True)
|
||||
self.setFrameShape(QFrame.Shape.NoFrame)
|
||||
|
||||
self._container = QWidget()
|
||||
self._layout = QVBoxLayout(self._container)
|
||||
self._layout.setContentsMargins(4, 8, 4, 8)
|
||||
self._layout.setSpacing(16)
|
||||
|
||||
self._spacer = QSpacerItem(0, 0, QSizePolicy.Policy.Minimum, QSizePolicy.Policy.Expanding)
|
||||
self._layout.addSpacerItem(self._spacer)
|
||||
|
||||
self.setWidget(self._container)
|
||||
|
||||
def add_card(self, card: BaseScopeCard) -> None:
|
||||
"""Insert a card above the trailing spacer.
|
||||
|
||||
Args:
|
||||
card (BaseScopeCard): The card widget to add to the list.
|
||||
"""
|
||||
idx = self._layout.count() - 1
|
||||
self._layout.insertWidget(idx, card)
|
||||
card.delete_requested.connect(lambda: self._remove_card(card))
|
||||
card.delete_requested.connect(self.cards_changed)
|
||||
card.data_changed.connect(self.cards_changed)
|
||||
self.cards_changed.emit()
|
||||
|
||||
def clear_cards(self) -> None:
|
||||
"""Remove all cards without touching the trailing spacer."""
|
||||
for index in range(self._layout.count() - 2, -1, -1):
|
||||
item = self._layout.itemAt(index)
|
||||
if item is None:
|
||||
continue
|
||||
card = item.widget()
|
||||
if isinstance(card, BaseScopeCard):
|
||||
self._layout.removeWidget(card)
|
||||
card.deleteLater()
|
||||
self.cards_changed.emit()
|
||||
|
||||
def cards(self) -> list[BaseScopeCard]:
|
||||
"""Return the cards currently stored in the list."""
|
||||
results: list[BaseScopeCard] = []
|
||||
for index in range(self._layout.count()):
|
||||
item = self._layout.itemAt(index)
|
||||
if item is None:
|
||||
continue
|
||||
card = item.widget()
|
||||
if isinstance(card, BaseScopeCard):
|
||||
results.append(card)
|
||||
return results
|
||||
|
||||
def _remove_card(self, card: BaseScopeCard) -> None:
|
||||
self._layout.removeWidget(card)
|
||||
card.deleteLater()
|
||||
self.cards_changed.emit()
|
||||
|
||||
|
||||
class BaseScopeCard(QFrame):
|
||||
"""Base card with shared identity, scope, and enabled fields."""
|
||||
|
||||
delete_requested = Signal()
|
||||
data_changed = Signal()
|
||||
|
||||
def __init__(self, parent: QWidget | None = None) -> None:
|
||||
super().__init__(parent)
|
||||
self._id: str = str(uuid.uuid4())
|
||||
|
||||
self.setFrameShape(QFrame.Shape.StyledPanel)
|
||||
self.setFrameShadow(QFrame.Shadow.Raised)
|
||||
self.setStyleSheet(
|
||||
"BaseScopeCard {"
|
||||
" border: 1px solid palette(mid);"
|
||||
" border-radius: 6px;"
|
||||
" background: palette(base);"
|
||||
"}"
|
||||
)
|
||||
|
||||
root = QVBoxLayout(self)
|
||||
root.setContentsMargins(20, 16, 20, 20)
|
||||
root.setSpacing(14)
|
||||
|
||||
header_row = QHBoxLayout()
|
||||
header_row.setSpacing(10)
|
||||
|
||||
self.enabled_checkbox = QCheckBox("Enabled")
|
||||
self.enabled_checkbox.setChecked(True)
|
||||
self.enabled_checkbox.toggled.connect(self.data_changed)
|
||||
header_row.addWidget(self.enabled_checkbox)
|
||||
header_row.addStretch(1)
|
||||
|
||||
self._delete_btn = QToolButton()
|
||||
delete_icon = material_icon(
|
||||
"delete", size=(25, 25), convert_to_pixmap=False, filled=False, color="#CC181E"
|
||||
)
|
||||
self._delete_btn.setToolTip("Delete this scope configuration")
|
||||
self._delete_btn.setIcon(delete_icon)
|
||||
self._delete_btn.clicked.connect(self.delete_requested)
|
||||
header_row.addWidget(self._delete_btn)
|
||||
|
||||
root.addLayout(header_row)
|
||||
|
||||
identity_row = QHBoxLayout()
|
||||
identity_row.setSpacing(16)
|
||||
|
||||
scope_col = QVBoxLayout()
|
||||
scope_col.setSpacing(4)
|
||||
scope_col.addWidget(QLabel("Scope"))
|
||||
self.scope_edit = QLineEdit()
|
||||
self.scope_edit.setPlaceholderText("e.g. user, admin")
|
||||
self.scope_edit.textChanged.connect(self.data_changed)
|
||||
scope_col.addWidget(self.scope_edit)
|
||||
identity_row.addLayout(scope_col, 1)
|
||||
|
||||
name_col = QVBoxLayout()
|
||||
name_col.setSpacing(4)
|
||||
name_col.addWidget(QLabel("Name (optional)"))
|
||||
self.name_edit = QLineEdit()
|
||||
self.name_edit.setPlaceholderText("display name")
|
||||
self.name_edit.textChanged.connect(self.data_changed)
|
||||
name_col.addWidget(self.name_edit)
|
||||
identity_row.addLayout(name_col, 1)
|
||||
|
||||
root.addLayout(identity_row)
|
||||
|
||||
self.content_layout = QVBoxLayout()
|
||||
self.content_layout.setContentsMargins(0, 0, 0, 0)
|
||||
self.content_layout.setSpacing(12)
|
||||
root.addLayout(self.content_layout)
|
||||
|
||||
def get_data(self) -> dict:
|
||||
"""Return the common payload for a messaging service card."""
|
||||
return {
|
||||
"id": self._id,
|
||||
"scope": self.scope_edit.text(),
|
||||
"enabled": self.enabled_checkbox.isChecked(),
|
||||
"name": self.name_edit.text() or None,
|
||||
}
|
||||
|
||||
def set_data(self, info: messages.MessagingService) -> None: # type: ignore[name-defined]
|
||||
"""Populate the shared card fields from a messaging service.
|
||||
|
||||
Args:
|
||||
info (messages.MessagingService): The service object used to populate the card.
|
||||
"""
|
||||
self._id = info.id
|
||||
self.scope_edit.setText(info.scope)
|
||||
self.enabled_checkbox.setChecked(info.enabled)
|
||||
self.name_edit.setText(info.name or "")
|
||||
|
||||
|
||||
class SciLogScopeCard(BaseScopeCard):
|
||||
"""Card used to configure SciLog service settings."""
|
||||
|
||||
def __init__(self, parent: QWidget | None = None) -> None:
|
||||
super().__init__(parent)
|
||||
|
||||
col = QVBoxLayout()
|
||||
col.setSpacing(4)
|
||||
col.addWidget(QLabel("Logbook ID"))
|
||||
self.logbook_id_edit = QLineEdit()
|
||||
self.logbook_id_edit.setPlaceholderText("e.g. lb-12345")
|
||||
self.logbook_id_edit.textChanged.connect(self.data_changed)
|
||||
col.addWidget(self.logbook_id_edit)
|
||||
self.content_layout.addLayout(col)
|
||||
|
||||
def get_data(self) -> dict:
|
||||
"""Return the SciLog-specific payload for this card."""
|
||||
data = super().get_data()
|
||||
data["service_type"] = "scilog"
|
||||
data["logbook_id"] = self.logbook_id_edit.text()
|
||||
return data
|
||||
|
||||
def set_data(self, info: messages.SciLogServiceInfo) -> None: # type: ignore[override]
|
||||
"""Populate the card from SciLog service information.
|
||||
|
||||
Args:
|
||||
info (messages.SciLogServiceInfo): The SciLog service object used to populate the card.
|
||||
"""
|
||||
super().set_data(info)
|
||||
self.logbook_id_edit.setText(info.logbook_id)
|
||||
|
||||
|
||||
class TeamsScopeCard(BaseScopeCard):
|
||||
"""Card used to configure MS Teams service settings."""
|
||||
|
||||
def __init__(self, parent: QWidget | None = None) -> None:
|
||||
super().__init__(parent)
|
||||
|
||||
fields_row = QHBoxLayout()
|
||||
fields_row.setSpacing(16)
|
||||
|
||||
col = QVBoxLayout()
|
||||
col.setSpacing(4)
|
||||
col.addWidget(QLabel("Workflow Webhook URL"))
|
||||
self.workflow_webhook_url_edit = edit = QLineEdit(parent=self)
|
||||
edit.setPlaceholderText("e.g. https://outlook.office.com/webhook/…")
|
||||
edit.textChanged.connect(self.data_changed)
|
||||
col.addWidget(edit)
|
||||
fields_row.addLayout(col, 1)
|
||||
|
||||
self.content_layout.addLayout(fields_row)
|
||||
|
||||
def get_data(self) -> dict:
|
||||
"""Return the MS Teams-specific payload for this card."""
|
||||
data = super().get_data()
|
||||
data["service_type"] = "teams"
|
||||
data["workflow_webhook_url"] = self.workflow_webhook_url_edit.text()
|
||||
return data
|
||||
|
||||
def set_data(self, info: messages.TeamsServiceInfo) -> None: # type: ignore[override]
|
||||
"""Populate the card from MS Teams service information.
|
||||
|
||||
Args:
|
||||
info (messages.TeamsServiceInfo): The MS Teams service object used to populate the card.
|
||||
"""
|
||||
super().set_data(info)
|
||||
self.workflow_webhook_url_edit.setText(info.workflow_webhook_url)
|
||||
|
||||
|
||||
class _SignalState(IntEnum):
|
||||
UNCONFIGURED = 0
|
||||
PENDING = 1
|
||||
CONFIGURED = 2
|
||||
|
||||
|
||||
class SignalScopeCard(BaseScopeCard):
|
||||
"""Card used to configure Signal service settings and linking state."""
|
||||
|
||||
_MOCK_GROUP_ID = "grp-8a3f291c"
|
||||
|
||||
def __init__(self, parent: QWidget | None = None) -> None:
|
||||
super().__init__(parent)
|
||||
|
||||
self._state = _SignalState.UNCONFIGURED
|
||||
self._mock_group_id: str = ""
|
||||
self._mock_group_link: str = ""
|
||||
|
||||
stacked_container = QWidget()
|
||||
self._stacked = QStackedLayout(stacked_container)
|
||||
self._stacked.setContentsMargins(0, 0, 0, 0)
|
||||
self.content_layout.addWidget(stacked_container)
|
||||
|
||||
self._build_unconfigured_page()
|
||||
self._build_pending_page()
|
||||
self._build_configured_page()
|
||||
|
||||
self._stacked.setCurrentIndex(_SignalState.UNCONFIGURED)
|
||||
|
||||
def _build_unconfigured_page(self) -> None:
|
||||
page = QWidget()
|
||||
row = QHBoxLayout(page)
|
||||
row.setContentsMargins(0, 0, 0, 0)
|
||||
row.setSpacing(6)
|
||||
|
||||
phone_col = QVBoxLayout()
|
||||
phone_col.setSpacing(4)
|
||||
phone_col.addWidget(QLabel("Phone Number"))
|
||||
self._phone_edit = QLineEdit()
|
||||
self._phone_edit.setValidator(
|
||||
QRegularExpressionValidator(QRegularExpression(r"^\+\S*$"), self._phone_edit)
|
||||
)
|
||||
self._phone_edit.setPlaceholderText("+41791234567")
|
||||
self._phone_edit.textChanged.connect(self.data_changed)
|
||||
phone_col.addWidget(self._phone_edit)
|
||||
row.addLayout(phone_col, 1)
|
||||
|
||||
start_linking_btn = QPushButton("Start Linking")
|
||||
start_linking_btn.setFixedWidth(100)
|
||||
start_linking_btn.clicked.connect(self._on_ping_clicked)
|
||||
row.addWidget(start_linking_btn, 0, Qt.AlignmentFlag.AlignBottom)
|
||||
|
||||
self._stacked.addWidget(page)
|
||||
|
||||
def _build_pending_page(self) -> None:
|
||||
page = QWidget()
|
||||
row = QHBoxLayout(page)
|
||||
row.setContentsMargins(0, 0, 0, 0)
|
||||
row.setSpacing(6)
|
||||
|
||||
waiting_lbl = QLabel("⏳ Waiting for you to reply on Signal…")
|
||||
waiting_lbl.setSizePolicy(QSizePolicy.Policy.Expanding, QSizePolicy.Policy.Preferred)
|
||||
row.addWidget(waiting_lbl)
|
||||
|
||||
cancel_btn = QPushButton("Cancel")
|
||||
cancel_btn.clicked.connect(self._on_cancel_clicked)
|
||||
row.addWidget(cancel_btn)
|
||||
|
||||
self._stacked.addWidget(page)
|
||||
|
||||
def _build_configured_page(self) -> None:
|
||||
page = QWidget()
|
||||
row = QHBoxLayout(page)
|
||||
row.setContentsMargins(0, 0, 0, 0)
|
||||
row.setSpacing(6)
|
||||
|
||||
self._linked_lbl = QLabel("🟢 Linked (Group ID: —)")
|
||||
self._linked_lbl.setSizePolicy(QSizePolicy.Policy.Expanding, QSizePolicy.Policy.Preferred)
|
||||
row.addWidget(self._linked_lbl)
|
||||
|
||||
unlink_btn = QPushButton("Unlink")
|
||||
unlink_btn.clicked.connect(self._on_unlink_clicked)
|
||||
row.addWidget(unlink_btn)
|
||||
|
||||
self._stacked.addWidget(page)
|
||||
|
||||
def _on_ping_clicked(self) -> None:
|
||||
self._state = _SignalState.PENDING
|
||||
self._stacked.setCurrentIndex(_SignalState.PENDING)
|
||||
QTimer.singleShot(3000, self._mock_backend_confirmation)
|
||||
|
||||
def _mock_backend_confirmation(self) -> None:
|
||||
if self._state != _SignalState.PENDING:
|
||||
return
|
||||
self._mock_group_id = self._MOCK_GROUP_ID
|
||||
self._mock_group_link = f"https://signal.group/#{self._mock_group_id}"
|
||||
self._linked_lbl.setText(f"🟢 Linked (Group ID: {self._mock_group_id})")
|
||||
self._state = _SignalState.CONFIGURED
|
||||
self._stacked.setCurrentIndex(_SignalState.CONFIGURED)
|
||||
self.data_changed.emit()
|
||||
|
||||
def _on_cancel_clicked(self) -> None:
|
||||
self._state = _SignalState.UNCONFIGURED
|
||||
self._stacked.setCurrentIndex(_SignalState.UNCONFIGURED)
|
||||
self.data_changed.emit()
|
||||
|
||||
def _on_unlink_clicked(self) -> None:
|
||||
self._mock_group_id = ""
|
||||
self._mock_group_link = ""
|
||||
self._state = _SignalState.UNCONFIGURED
|
||||
self._stacked.setCurrentIndex(_SignalState.UNCONFIGURED)
|
||||
self.data_changed.emit()
|
||||
|
||||
def get_data(self) -> dict:
|
||||
"""Return the Signal-specific payload for this card."""
|
||||
data = super().get_data()
|
||||
data["service_type"] = "signal"
|
||||
configured = self._state == _SignalState.CONFIGURED
|
||||
data["group_id"] = self._mock_group_id if configured else None
|
||||
data["group_link"] = self._mock_group_link if configured else None
|
||||
return data
|
||||
|
||||
def set_data(self, info: messages.SignalServiceInfo) -> None: # type: ignore[override]
|
||||
"""Populate the card from Signal service information.
|
||||
|
||||
Args:
|
||||
info (messages.SignalServiceInfo): The Signal service object used to populate the card.
|
||||
"""
|
||||
super().set_data(info)
|
||||
if info.group_id:
|
||||
self._mock_group_id = info.group_id
|
||||
self._mock_group_link = info.group_link or ""
|
||||
self._linked_lbl.setText(f"🟢 Linked (Group ID: {self._mock_group_id})")
|
||||
self._state = _SignalState.CONFIGURED
|
||||
self._stacked.setCurrentIndex(_SignalState.CONFIGURED)
|
||||
return
|
||||
self._mock_group_id = ""
|
||||
self._mock_group_link = ""
|
||||
self._state = _SignalState.UNCONFIGURED
|
||||
self._stacked.setCurrentIndex(_SignalState.UNCONFIGURED)
|
||||
|
||||
|
||||
_CARD_CLASSES: dict[CardType, Type[BaseScopeCard]] = {
|
||||
"scilog": SciLogScopeCard,
|
||||
"signal": SignalScopeCard,
|
||||
"teams": TeamsScopeCard,
|
||||
}
|
||||
|
||||
|
||||
def make_card(card_type: CardType) -> BaseScopeCard:
|
||||
"""Create a new service card for the requested card type.
|
||||
|
||||
Args:
|
||||
card_type (CardType): The service type for the card to create.
|
||||
"""
|
||||
return _CARD_CLASSES[card_type]()
|
||||
|
||||
|
||||
def card_from_service(info: object) -> BaseScopeCard:
|
||||
"""Create and populate a card from a messaging service object.
|
||||
|
||||
Args:
|
||||
info (object): A messaging service object with a ``service_type`` attribute.
|
||||
"""
|
||||
service_type: str = getattr(info, "service_type", "")
|
||||
card_class = _CARD_CLASSES.get(service_type) # type: ignore[arg-type]
|
||||
if card_class is None:
|
||||
raise ValueError(f"Unknown service_type: {service_type!r}")
|
||||
card = card_class()
|
||||
card.set_data(info) # type: ignore[arg-type]
|
||||
return card
|
||||
@@ -0,0 +1,125 @@
|
||||
"""Module for the service scope event subscription table widget."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from qtpy.QtCore import Qt
|
||||
from qtpy.QtWidgets import (
|
||||
QAbstractItemView,
|
||||
QCheckBox,
|
||||
QHBoxLayout,
|
||||
QHeaderView,
|
||||
QSizePolicy,
|
||||
QTableWidget,
|
||||
QVBoxLayout,
|
||||
QWidget,
|
||||
)
|
||||
|
||||
|
||||
class ServiceScopeEventTableWidget(QWidget):
|
||||
"""Widget that manages per-scope event subscriptions for messaging services."""
|
||||
|
||||
EVENT_NAMES = ("new_scan", "scan_finished", "alarm")
|
||||
|
||||
def __init__(self, parent: QWidget | None = None) -> None:
|
||||
super().__init__(parent)
|
||||
self._services: list[dict] = []
|
||||
self._subscriptions: dict[str, dict[str, bool]] = {}
|
||||
|
||||
root = QVBoxLayout(self)
|
||||
root.setContentsMargins(0, 0, 0, 0)
|
||||
root.setSpacing(0)
|
||||
|
||||
self._table = QTableWidget(len(self.EVENT_NAMES), 0, self)
|
||||
self._table.setEditTriggers(QAbstractItemView.EditTrigger.NoEditTriggers)
|
||||
self._table.setSelectionMode(QAbstractItemView.SelectionMode.NoSelection)
|
||||
self._table.setAlternatingRowColors(True)
|
||||
self._table.setSizePolicy(QSizePolicy.Policy.Expanding, QSizePolicy.Policy.Expanding)
|
||||
self._table.setVerticalHeaderLabels(list(self.EVENT_NAMES))
|
||||
self._table.horizontalHeader().setStretchLastSection(True)
|
||||
|
||||
header = self._table.horizontalHeader()
|
||||
header.setSectionResizeMode(QHeaderView.ResizeMode.Stretch)
|
||||
self._table.verticalHeader().setSectionResizeMode(QHeaderView.ResizeMode.ResizeToContents)
|
||||
|
||||
root.addWidget(self._table, 1)
|
||||
|
||||
def set_services(self, services: list[dict]) -> None:
|
||||
"""Update the table rows to match the current services.
|
||||
|
||||
Args:
|
||||
services (list[dict]): Service dictionaries collected from the service configuration panels.
|
||||
"""
|
||||
self._services = [dict(service) for service in services]
|
||||
known_ids = {str(service.get("id", "")) for service in self._services if service.get("id")}
|
||||
self._subscriptions = {
|
||||
service_id: subscriptions
|
||||
for service_id, subscriptions in self._subscriptions.items()
|
||||
if service_id in known_ids
|
||||
}
|
||||
|
||||
self._table.clearContents()
|
||||
self._table.setRowCount(len(self.EVENT_NAMES))
|
||||
self._table.setColumnCount(len(self._services))
|
||||
self._table.setHorizontalHeaderLabels(
|
||||
[self._format_service_label(service) for service in self._services]
|
||||
)
|
||||
|
||||
for column, service in enumerate(self._services):
|
||||
service_id = str(service.get("id", ""))
|
||||
|
||||
event_states = self._subscriptions.setdefault(
|
||||
service_id, {event_name: False for event_name in self.EVENT_NAMES}
|
||||
)
|
||||
for row, event_name in enumerate(self.EVENT_NAMES):
|
||||
self._table.setCellWidget(
|
||||
row,
|
||||
column,
|
||||
self._make_checkbox_cell(
|
||||
service_id, event_name, event_states.get(event_name, False)
|
||||
),
|
||||
)
|
||||
|
||||
def get_data(self) -> list[dict]:
|
||||
"""Return the event subscriptions for the current services."""
|
||||
results: list[dict] = []
|
||||
for service in self._services:
|
||||
service_id = str(service.get("id", ""))
|
||||
results.append(
|
||||
{
|
||||
"id": service_id,
|
||||
"source": service.get("source"),
|
||||
"service_type": service.get("service_type"),
|
||||
"scope": service.get("scope"),
|
||||
"events": dict(
|
||||
self._subscriptions.get(
|
||||
service_id, {event_name: False for event_name in self.EVENT_NAMES}
|
||||
)
|
||||
),
|
||||
}
|
||||
)
|
||||
return results
|
||||
|
||||
def _format_service_label(self, service: dict) -> str:
|
||||
service_name = str(service.get("service_type", ""))
|
||||
scope_name = str(service.get("scope", ""))
|
||||
source_name = str(service.get("source", ""))
|
||||
return f"{service_name}\n{scope_name}\n({source_name})"
|
||||
|
||||
def _make_checkbox_cell(self, service_id: str, event_name: str, checked: bool) -> QWidget:
|
||||
checkbox = QCheckBox()
|
||||
checkbox.setChecked(checked)
|
||||
checkbox.toggled.connect(
|
||||
lambda state, current_service_id=service_id, current_event_name=event_name: self._set_event_state(
|
||||
current_service_id, current_event_name, state
|
||||
)
|
||||
)
|
||||
|
||||
container = QWidget()
|
||||
layout = QHBoxLayout(container)
|
||||
layout.setContentsMargins(0, 0, 0, 0)
|
||||
layout.setAlignment(Qt.AlignmentFlag.AlignCenter)
|
||||
layout.addWidget(checkbox)
|
||||
return container
|
||||
|
||||
def _set_event_state(self, service_id: str, event_name: str, checked: bool) -> None:
|
||||
self._subscriptions.setdefault(service_id, {})[event_name] = checked
|
||||
Reference in New Issue
Block a user