mirror of
https://github.com/bec-project/bec_widgets.git
synced 2026-05-09 08:12:15 +02:00
Compare commits
1 Commits
| 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 (
|
from bec_widgets.widgets.services.bec_atlas_admin_view.experiment_selection.experiment_selection import (
|
||||||
ExperimentSelection,
|
ExperimentSelection,
|
||||||
)
|
)
|
||||||
|
from bec_widgets.widgets.services.bec_messaging_config.bec_messaging_config_widget import (
|
||||||
|
BECMessagingConfigWidget,
|
||||||
|
)
|
||||||
|
|
||||||
if TYPE_CHECKING: # pragma: no cover
|
if TYPE_CHECKING: # pragma: no cover
|
||||||
from qtpy.QtWidgets import QToolBar
|
from qtpy.QtWidgets import QToolBar
|
||||||
@@ -104,7 +107,8 @@ class OverviewWidget(QGroupBox):
|
|||||||
content_layout = QVBoxLayout(content)
|
content_layout = QVBoxLayout(content)
|
||||||
content.setFrameShape(QFrame.Shape.StyledPanel)
|
content.setFrameShape(QFrame.Shape.StyledPanel)
|
||||||
content.setFrameShadow(QFrame.Shadow.Raised)
|
content.setFrameShadow(QFrame.Shadow.Raised)
|
||||||
content.setStyleSheet("""
|
content.setStyleSheet(
|
||||||
|
"""
|
||||||
QFrame
|
QFrame
|
||||||
{
|
{
|
||||||
border: 1px solid #cccccc;
|
border: 1px solid #cccccc;
|
||||||
@@ -113,7 +117,8 @@ class OverviewWidget(QGroupBox):
|
|||||||
{
|
{
|
||||||
border: none;
|
border: none;
|
||||||
}
|
}
|
||||||
""")
|
"""
|
||||||
|
)
|
||||||
content_layout.setAlignment(Qt.AlignmentFlag.AlignCenter)
|
content_layout.setAlignment(Qt.AlignmentFlag.AlignCenter)
|
||||||
content.setFixedSize(400, 280)
|
content.setFixedSize(400, 280)
|
||||||
|
|
||||||
@@ -299,6 +304,11 @@ class BECAtlasAdminView(BECWidget, QWidget):
|
|||||||
self.experiment_selection.setVisible(False)
|
self.experiment_selection.setVisible(False)
|
||||||
self.stacked_layout.addWidget(self.experiment_selection)
|
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
|
# Connect signals
|
||||||
self.overview_widget.login_requested.connect(self._on_login_requested)
|
self.overview_widget.login_requested.connect(self._on_login_requested)
|
||||||
self.overview_widget.change_experiment_requested.connect(
|
self.overview_widget.change_experiment_requested.connect(
|
||||||
@@ -392,6 +402,7 @@ class BECAtlasAdminView(BECWidget, QWidget):
|
|||||||
"""Show the overview panel."""
|
"""Show the overview panel."""
|
||||||
self.overview_widget.setVisible(True)
|
self.overview_widget.setVisible(True)
|
||||||
self.experiment_selection.setVisible(False)
|
self.experiment_selection.setVisible(False)
|
||||||
|
self.messaging_config_widget.setVisible(False)
|
||||||
self.stacked_layout.setCurrentWidget(self.overview_widget)
|
self.stacked_layout.setCurrentWidget(self.overview_widget)
|
||||||
|
|
||||||
def _on_experiment_selection_selected(self):
|
def _on_experiment_selection_selected(self):
|
||||||
@@ -401,12 +412,20 @@ class BECAtlasAdminView(BECWidget, QWidget):
|
|||||||
return
|
return
|
||||||
self.overview_widget.setVisible(False)
|
self.overview_widget.setVisible(False)
|
||||||
self.experiment_selection.setVisible(True)
|
self.experiment_selection.setVisible(True)
|
||||||
|
self.messaging_config_widget.setVisible(False)
|
||||||
self.stacked_layout.setCurrentWidget(self.experiment_selection)
|
self.stacked_layout.setCurrentWidget(self.experiment_selection)
|
||||||
|
|
||||||
def _on_messaging_services_selected(self):
|
def _on_messaging_services_selected(self):
|
||||||
"""Show the messaging services panel."""
|
"""Show the messaging services panel."""
|
||||||
logger.info("Messaging services panel is not implemented yet.")
|
if not self._authenticated:
|
||||||
return
|
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
|
## Internal slots
|
||||||
@@ -446,6 +465,7 @@ class BECAtlasAdminView(BECWidget, QWidget):
|
|||||||
atlas_url=self._atlas_url,
|
atlas_url=self._atlas_url,
|
||||||
)
|
)
|
||||||
self.atlas_http_service._set_current_deployment_info(deployment)
|
self.atlas_http_service._set_current_deployment_info(deployment)
|
||||||
|
self.messaging_config_widget.populate_from_deployment(deployment)
|
||||||
|
|
||||||
def _fetch_available_experiments(self):
|
def _fetch_available_experiments(self):
|
||||||
"""Fetch the list of available experiments for the authenticated user."""
|
"""Fetch the list of available experiments for the authenticated user."""
|
||||||
@@ -501,9 +521,7 @@ class BECAtlasAdminView(BECWidget, QWidget):
|
|||||||
|
|
||||||
if authenticated:
|
if authenticated:
|
||||||
self.toolbar.components.get_action("experiment_selection").action.setEnabled(True)
|
self.toolbar.components.get_action("experiment_selection").action.setEnabled(True)
|
||||||
self.toolbar.components.get_action("messaging_services").action.setEnabled(
|
self.toolbar.components.get_action("messaging_services").action.setEnabled(True)
|
||||||
False
|
|
||||||
) # TODO activate once messaging is added
|
|
||||||
self.toolbar.components.get_action("logout").action.setEnabled(True)
|
self.toolbar.components.get_action("logout").action.setEnabled(True)
|
||||||
self._fetch_available_experiments() # Fetch experiments upon successful authentication
|
self._fetch_available_experiments() # Fetch experiments upon successful authentication
|
||||||
self._atlas_info_widget.set_logged_in(info.email)
|
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