feat(notifications): add support for notifications and notification hooks

Author: wakonig

bec_lib/bec_lib/endpoints.py

@@ -1995,6 +1995,38 @@ def available_messaging_services():
             message_op=MessageOp.STREAM,
         )
 
+    @staticmethod
+    def notification(event_type: str):
+        """
+        Endpoint for transient notification events that SciHub can route to
+        configured messaging services.
+
+        Args:
+            event_type (str): Notification event name such as ``new_scan``.
+
+        Returns:
+            EndpointInfo: Endpoint for notification events.
+        """
+        endpoint = f"{EndpointType.INTERNAL.value}/messaging_services/notification/{event_type}"
+        return EndpointInfo(
+            endpoint=endpoint, message_type=messages.NotificationMessage, message_op=MessageOp.SEND
+        )
+
+    @staticmethod
+    def notification_config():
+        """
+        Endpoint for persisted notification routing configuration.
+
+        Returns:
+            EndpointInfo: Endpoint for notification routing config.
+        """
+        endpoint = f"{EndpointType.USER.value}/messaging_services/notification_config"
+        return EndpointInfo(
+            endpoint=endpoint,
+            message_type=messages.NotificationConfigMessage,
+            message_op=MessageOp.SET_PUBLISH,
+        )
+
     @staticmethod
     def message_service_ingest(deployment_name: str):
         """

bec_lib/bec_lib/messages.py

@@ -1728,6 +1728,39 @@ class MessagingConfig(BaseModel):
     scilog: MessagingServiceScopeConfig
 
 
+class NotificationServiceTarget(BaseModel):
+    service_name: Literal["signal", "teams", "scilog"]
+    scope: str | list[str] | None = None
+
+
+class NotificationMessage(BECMessage):
+    """
+    Message for notification events that should be routed through configured
+    messaging services.
+
+    Args:
+        message (str): Notification body text.
+    """
+
+    msg_type: ClassVar[str] = "notification_message"
+    message: str
+
+
+class NotificationConfigMessage(BECMessage):
+    """
+    Routing configuration for notification events.
+
+    Args:
+        routes: Mapping of event name to messaging service targets.
+    """
+
+    msg_type: ClassVar[str] = "notification_config_message"
+    routes: dict[
+        Literal["new_scan", "scan_completed", "alarm", "scan_interlock"] | str,
+        list[NotificationServiceTarget],
+    ] = Field(default_factory=dict)
+
+
 AvailableMessagingServices = Annotated[
     Union[SignalServiceInfo, SciLogServiceInfo, TeamsServiceInfo],
     Field(discriminator="service_type"),

bec_lib/bec_lib/messaging_hooks.py

@@ -0,0 +1,114 @@
+from __future__ import annotations
+
+import enum
+from typing import TYPE_CHECKING, cast
+
+from bec_lib import messages
+from bec_lib.connector import MessageObject
+from bec_lib.endpoints import MessageEndpoints
+from bec_lib.logger import bec_logger
+from bec_lib.messaging_services import (
+    SciLogMessagingService,
+    SignalMessagingService,
+    TeamsMessagingService,
+)
+
+if TYPE_CHECKING:
+    from bec_lib.redis_connector import RedisConnector
+
+logger = bec_logger.logger
+
+
+class MessagingEvent(str, enum.Enum):
+    """
+    Enumeration of messaging events that can trigger configured hooks.
+    """
+
+    SCAN = "new_scan"
+    SCAN_COMPLETED = "scan_completed"
+    ALARM = "alarm"
+    SCAN_INTERLOCK = "scan_interlock"
+
+
+class MessagingManager:
+    """
+    Manage notification routing from internal events to concrete messaging
+    services.
+    """
+
+    def __init__(self, connector: RedisConnector):
+        self.connector = connector
+        self.config: dict[str, list[messages.NotificationServiceTarget]] = {}
+        signal = SignalMessagingService(self.connector)
+        scilog = SciLogMessagingService(self.connector)
+        teams = TeamsMessagingService(self.connector)
+        self._service_by_name = {"signal": signal, "scilog": scilog, "teams": teams}
+
+        self.connector.register(
+            patterns=MessageEndpoints.notification("*"), cb=self._handle_notification
+        )
+        self.connector.register(
+            topics=MessageEndpoints.notification_config(), cb=self._handle_notification_config
+        )
+
+        config_msg = self.connector.get(MessageEndpoints.notification_config())
+        if config_msg is not None:
+            self.on_notification_config(config_msg)
+
+    def _handle_notification(self, msg_obj: MessageObject[messages.NotificationMessage], **_kwargs):
+        prefix = MessageEndpoints.notification("").endpoint
+        event_type_str = msg_obj.topic.removeprefix(prefix)
+        self.on_notification(event_type_str, cast(messages.NotificationMessage, msg_obj.value))
+
+    def _handle_notification_config(
+        self, msg_obj: MessageObject[messages.NotificationConfigMessage], **_kwargs
+    ):
+        self.on_notification_config(cast(messages.NotificationConfigMessage, msg_obj.value))
+
+    def on_notification(self, event_type: str, message: messages.NotificationMessage) -> None:
+        """
+        Handle a notification event by routing it to the configured messaging services.
+
+        Args:
+            event_type(str): The type of the event that triggered the notification.
+            message(messages.NotificationMessage): The notification message containing the content to be sent.
+        """
+        routes = self.config.get(event_type, [])
+        for route in routes:
+            service = self._service_by_name.get(route.service_name)
+            if service is None:
+                logger.warning(f"Unknown messaging service: {route.service_name}")
+                continue
+            try:
+                logger.info(
+                    f"Routing notification for {event_type} to {route.service_name}: {message.message}"
+                )
+                service.new().add_text(message.message).send(scope=route.scope)
+            except RuntimeError as exc:
+                logger.warning(
+                    f"Failed to send notification for {event_type} via {route.service_name}: {exc}"
+                )
+
+    def on_notification_config(self, message: messages.NotificationConfigMessage) -> None:
+        """
+        Update the notification routing configuration based on the received message.
+
+        Args:
+            message(NotificationConfigMessage): The message containing the new routing configuration.
+        """
+        config: dict[str, list[messages.NotificationServiceTarget]] = {}
+        for event_name, targets in message.routes.items():
+            config[event_name] = targets
+        self.config = config
+
+    def shutdown(self) -> None:
+        """
+        Shutdown the messaging manager by unregistering all notification handlers.
+        """
+
+        self.connector.unregister(
+            patterns=MessageEndpoints.notification("*"), cb=self._handle_notification
+        )
+        self.connector.unregister(
+            topics=MessageEndpoints.notification_config(), cb=self._handle_notification_config
+        )

bec_lib/bec_lib/messaging_services.py

@@ -1,5 +1,6 @@
 from __future__ import annotations
 
+import enum
 import mimetypes
 import os
 from abc import ABC
@@ -9,6 +10,7 @@
 from bec_lib.endpoints import MessageEndpoints
 
 if TYPE_CHECKING:
+    from bec_lib.connector import MessageObject
     from bec_lib.redis_connector import RedisConnector
 
 # Type variable for the message object class
@@ -126,6 +128,7 @@ class MessagingService(ABC, Generic[MessageObjectT]):
     def __init__(self, redis_connector: RedisConnector) -> None:
         self._redis_connector = redis_connector
         self._scopes: set[str] = set()
+        self._auto_notifications: dict[str, list[str]] = {}
         self._enabled = False
         self._default_scope: str | list[str] | None = None
         self._service_config: messages.AvailableMessagingServicesMessage | None = None
@@ -134,6 +137,12 @@ def __init__(self, redis_connector: RedisConnector) -> None:
             cb=self._on_new_scope_change_msg,
             from_start=True,
         )
+        self._redis_connector.register(
+            MessageEndpoints.notification_config(), cb=self._on_notification_config_change_msg
+        )
+        config_msg = self._redis_connector.get(MessageEndpoints.notification_config())
+        if config_msg is not None:
+            self._update_auto_notifications(config_msg)
 
     def set_default_scope(self, scope: str | list[str] | None) -> None:
         """
@@ -146,6 +155,133 @@ def set_default_scope(self, scope: str | list[str] | None) -> None:
             raise ValueError(f"Scope '{scope}' is not available for this messaging service.")
         self._default_scope = scope
 
+    def set_auto_notifications(
+        self,
+        event_type: Literal["new_scan", "scan_completed", "alarm", "scan_interlock"] | str,
+        enabled: bool,
+        scopes: list[str] | str | None = None,
+    ) -> None:
+        """
+        Set automatic notifications for a specific event type.
+
+        Args:
+            event_type (Literal["new_scan", "scan_completed", "alarm", "scan_interlock"] | str): The type of event to set notifications for.
+            enabled (bool): Whether to enable or disable notifications for the event.
+            scopes (list[str] | str | None): The scopes to apply the notifications to.
+        """
+        event_name = event_type.value if isinstance(event_type, enum.Enum) else event_type
+        scopes_list: list[str] = []
+        if scopes is not None:
+            if isinstance(scopes, str):
+                scopes_list = [scopes]
+            else:
+                scopes_list = scopes
+
+            for scope in scopes_list:
+                if scope not in self._scopes:
+                    raise ValueError(
+                        f"Scope '{scope}' is not available for this messaging service."
+                    )
+        else:
+            if self._default_scope is not None:
+                scopes_list = (
+                    [self._default_scope]
+                    if isinstance(self._default_scope, str)
+                    else self._default_scope
+                )
+            elif not self._SUPPORTS_EMPTY_SCOPES:
+                raise ValueError(
+                    "Scopes must be provided when there is no default scope and empty scopes are not supported."
+                )
+
+        if enabled:
+            # merge with existing scopes if already enabled for this event type
+            existing_scopes = set(self._auto_notifications.get(event_name, []))
+            existing_scopes.update(scopes_list)
+            self._auto_notifications[event_name] = list(existing_scopes)
+        else:
+            # if disabling, remove the scopes for this event type, or the entire event type if no scopes are provided
+            if scopes_list:
+                existing_scopes = set(self._auto_notifications.get(event_name, []))
+                existing_scopes.difference_update(scopes_list)
+                if existing_scopes:
+                    self._auto_notifications[event_name] = list(existing_scopes)
+                else:
+                    self._auto_notifications.pop(event_name, None)
+            else:
+                self._auto_notifications.pop(event_name, None)
+
+        self._sync_auto_notifications_config(event_name, enabled=enabled, scopes=scopes_list)
+
+    def _sync_auto_notifications_config(
+        self, event_name: str, enabled: bool, scopes: list[str]
+    ) -> None:
+        config_msg = self._redis_connector.get(MessageEndpoints.notification_config())
+        if config_msg is None:
+            config_msg = messages.NotificationConfigMessage()
+
+        routes = {name: list(targets) for name, targets in config_msg.routes.items()}
+        event_routes = list(routes.get(event_name, []))
+
+        if enabled:
+            scopes_to_add = scopes or [None]
+            for scope in scopes_to_add:
+                target = messages.NotificationServiceTarget(
+                    service_name=self._SERVICE_NAME, scope=scope
+                )
+                if not any(existing == target for existing in event_routes):
+                    event_routes.append(target)
+        else:
+            if scopes:
+                event_routes = [
+                    target
+                    for target in event_routes
+                    if not (target.service_name == self._SERVICE_NAME and target.scope in scopes)
+                ]
+            else:
+                event_routes = [
+                    target for target in event_routes if target.service_name != self._SERVICE_NAME
+                ]
+
+        if event_routes:
+            routes[event_name] = event_routes
+        else:
+            routes.pop(event_name, None)
+
+        self._redis_connector.set_and_publish(
+            MessageEndpoints.notification_config(),
+            messages.NotificationConfigMessage(routes=routes, metadata=config_msg.metadata),
+        )
+
+    def _on_notification_config_change_msg(
+        self,
+        message: (
+            MessageObject[messages.NotificationConfigMessage]
+            | dict[str, messages.NotificationConfigMessage]
+        ),
+    ) -> None:
+        config_msg = message.value if hasattr(message, "value") else message["data"]
+        self._update_auto_notifications(config_msg)
+
+    def _update_auto_notifications(self, config_msg: messages.NotificationConfigMessage) -> None:
+        auto_notifications: dict[str, list[str]] = {}
+        for event_name, targets in config_msg.routes.items():
+            scopes: list[str] = []
+            has_matching_service = False
+            for target in targets:
+                if target.service_name != self._SERVICE_NAME:
+                    continue
+                has_matching_service = True
+                if isinstance(target.scope, str):
+                    scopes.append(target.scope)
+                elif isinstance(target.scope, list):
+                    scopes.extend(target.scope)
+
+            if has_matching_service:
+                auto_notifications[event_name] = list(dict.fromkeys(scopes))
+
+        self._auto_notifications = auto_notifications
+
     def _on_new_scope_change_msg(
         self, message: dict[str, messages.AvailableMessagingServicesMessage]
     ) -> None:

bec_lib/bec_lib/redis_connector.py

@@ -57,7 +57,9 @@
     DynamicMetricDict,
     DynamicMetricMessage,
     ErrorInfo,
+    NotificationMessage,
 )
+from bec_lib.messaging_hooks import MessagingEvent
 from bec_lib.serialization import MsgpackSerialization
 
 logger = bec_logger.logger
@@ -655,6 +657,23 @@ def raise_alarm(self, severity: Alarms, info: ErrorInfo, metadata: dict | None =
         """
         alarm_msg = AlarmMessage(severity=severity, info=info, metadata=metadata or {})
         self.set_and_publish(MessageEndpoints.alarm(), alarm_msg)
+        compact_message = info.compact_error_message or info.error_message or info.exception_type
+        self.notify(MessagingEvent.ALARM, compact_message)
+
+    def notify(self, event: MessagingEvent, message: str, pipe: Pipeline | None = None) -> None:
+        """
+        Publish a notification event for downstream routing by SciHub.
+
+        Args:
+            event(MessagingEvent): The type of the event that triggered the notification.
+            message(str): The notification message containing the content to be sent.
+            pipe(Pipeline, optional): Optional pipeline to enqueue the publish operation into.
+        """
+        self.send(
+            MessageEndpoints.notification(event.value),
+            NotificationMessage(message=message),
+            pipe=pipe,
+        )
 
     def pipeline(self) -> redis.client.Pipeline:
         """Create a new pipeline"""