1
0
mirror of https://github.com/bec-project/bec_widgets.git synced 2026-04-10 02:30:54 +02:00

Compare commits

...

1 Commits

Author SHA1 Message Date
de5a58a63b feat: add messaging config view 2026-03-20 14:49:16 +01:00
5 changed files with 894 additions and 7 deletions

View File

@@ -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)

View File

@@ -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())

View File

@@ -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

View File

@@ -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