f

Author: wakonig

bec_lib/bec_lib/messages.py

@@ -1744,7 +1744,15 @@ class NotificationConfigMessage(BECMessage):
 
     msg_type: ClassVar[str] = "notification_config_message"
     routes: dict[
-        Literal["new_scan", "scan_completed", "alarm", "scan_interlock"] | str,
+        Literal[
+            "new_scan",
+            "scan_completed",
+            "alarm_warning",
+            "alarm_minor",
+            "alarm_major",
+            "scan_interlock",
+        ]
+        | str,
         list[NotificationServiceTarget],
     ] = Field(default_factory=dict)
 
@@ -1754,7 +1762,7 @@ class NotificationMessage(BECMessage):
     Message for notifications to be sent to messaging services.
 
     Args:
-        event (str): Event name, e.g. "new_scan", "scan_completed", "alarm", "scan_interlock"
+        event (str): Event name, e.g. "new_scan", "scan_completed", "alarm_warning", "alarm_minor", "alarm_major", "scan_interlock"
         content (dict): Content of the notification message, can contain any relevant information about the event
         metadata (dict, optional): Additional metadata. Defaults to None.
     """

bec_lib/bec_lib/messaging_hooks.py

@@ -29,7 +29,9 @@ class MessagingEvent(str, enum.Enum):
 
     SCAN = "new_scan"
     SCAN_COMPLETED = "scan_completed"
-    ALARM = "alarm"
+    ALARM_WARNING = "alarm_warning"
+    ALARM_MINOR = "alarm_minor"
+    ALARM_MAJOR = "alarm_major"
     SCAN_INTERLOCK = "scan_interlock"
 
 

bec_lib/bec_lib/messaging_services.py

@@ -199,15 +199,25 @@ def set_default_scope(self, scope: str | list[str] | None) -> None:
 
     def set_auto_notifications(
         self,
-        event_type: Literal["new_scan", "scan_completed", "alarm", "scan_interlock"] | str,
+        event_type: (
+            Literal[
+                "new_scan",
+                "scan_completed",
+                "alarm_warning",
+                "alarm_minor",
+                "alarm_major",
+                "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.
+            event_type (Literal["new_scan", "scan_completed", "alarm_warning", "alarm_minor", "alarm_major", "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.
         """

bec_lib/bec_lib/redis_connector.py

@@ -659,7 +659,12 @@ 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)
+        event_by_severity = {
+            0: MessagingEvent.ALARM_WARNING,
+            1: MessagingEvent.ALARM_MINOR,
+            2: MessagingEvent.ALARM_MAJOR,
+        }
+        self.notify(event_by_severity[int(severity)], compact_message)
 
     def notify(
         self,

bec_lib/tests/test_messaging_hooks.py

@@ -122,15 +122,17 @@ def test_messaging_manager_routes_generic_notification_to_signal_as_plain_text(
     messaging_manager.on_notification_config(
         messages.NotificationConfigMessage(
             routes={
-                "alarm": [messages.NotificationServiceTarget(service_name="signal", scope="ops")]
+                "alarm_major": [
+                    messages.NotificationServiceTarget(service_name="signal", scope="ops")
+                ]
             }
         )
     )
 
     messaging_manager.on_notification(
-        MessagingEvent.ALARM,
+        MessagingEvent.ALARM_MAJOR,
         messages.NotificationMessage(
-            event="alarm",
+            event="alarm_major",
             message=messages.NotificationMessageObject()
             .add_text("Beamline checks failed", bold=True, color="red")
             .add_tags(["alarm"])
@@ -165,7 +167,9 @@ def test_handle_notification_routes_message_from_topic_suffix(messaging_manager)
 
 def test_handle_notification_config_forwards_message_value(messaging_manager):
     config_message = messages.NotificationConfigMessage(
-        routes={"alarm": [messages.NotificationServiceTarget(service_name="signal", scope="ops")]}
+        routes={
+            "alarm_minor": [messages.NotificationServiceTarget(service_name="signal", scope="ops")]
+        }
     )
     msg_obj = MessageObject(MessageEndpoints.notification_config().endpoint, config_message)
 
@@ -196,7 +200,7 @@ def test_to_service_message_for_teams_strips_html_and_ignores_non_text(messaging
         {"data": available_services}
     )
     message = messages.NotificationMessage(
-        event="alarm",
+        event="alarm_major",
         message=messages.NotificationMessageObject()
         .add_text("Beamline checks failed", bold=True, color="red")
         .add_tags(["alarm"])
@@ -221,7 +225,7 @@ def test_to_service_message_for_signal_preserves_files_and_plain_text(messaging_
         filename="alarm.png", mime_type="image/png", data=b"png-bytes", width=128, height=64
     )
     message = messages.NotificationMessage(
-        event="alarm",
+        event="alarm_warning",
         message=messages.NotificationMessageObject()
         .add_text("Beamline checks failed", italic=True)
         .add_tags(["alarm"])
@@ -241,7 +245,7 @@ def test_to_service_message_for_signal_preserves_files_and_plain_text(messaging_
 
 def test_to_service_message_raises_for_unsupported_service(messaging_manager):
     message = messages.NotificationMessage(
-        event="alarm",
+        event="alarm_minor",
         message=messages.NotificationMessageObject().add_text("Beamline checks failed")._content,
     )
 

bec_lib/tests/test_messaging_service.py

@@ -354,7 +354,7 @@ def test_notification_message_object_to_scilog_message(scilog_service):
     try:
         scilog_message = manager.to_service_message(
             scilog_service,
-            messages.NotificationMessage(event="alarm", message=notification._content),
+            messages.NotificationMessage(event="alarm_major", message=notification._content),
         )
     finally:
         manager.shutdown()
@@ -386,7 +386,7 @@ def test_notification_message_object_to_signal_message(signal_service):
     try:
         signal_message = manager.to_service_message(
             signal_service,
-            messages.NotificationMessage(event="alarm", message=notification._content),
+            messages.NotificationMessage(event="alarm_major", message=notification._content),
         )
     finally:
         manager.shutdown()
@@ -717,7 +717,7 @@ def test_messaging_service_tracks_external_notification_config_updates(
                     messages.NotificationServiceTarget(service_name="signal", scope="beamline-ops"),
                     messages.NotificationServiceTarget(service_name="scilog", scope="default"),
                 ],
-                "alarm": [
+                "alarm_major": [
                     messages.NotificationServiceTarget(
                         service_name="scilog", scope=["default", "secondary"]
                     )
@@ -730,13 +730,13 @@ def test_messaging_service_tracks_external_notification_config_updates(
     while (
         time.time() < deadline
         and scilog_service._auto_notifications  # pylint: disable=protected-access
-        != {"new_scan": ["default"], "alarm": ["default", "secondary"]}
+        != {"new_scan": ["default"], "alarm_major": ["default", "secondary"]}
     ):
         time.sleep(0.01)
 
     assert scilog_service._auto_notifications == {  # pylint: disable=protected-access
         "new_scan": ["default"],
-        "alarm": ["default", "secondary"],
+        "alarm_major": ["default", "secondary"],
     }
 
 

bec_lib/tests/test_redis_connector.py

@@ -64,14 +64,37 @@ def test_redis_connector_send_client_info(connector):
 
 
 @pytest.mark.parametrize(
-    "severity, alarm_type, msg, compact_msg, metadata",
+    "severity, expected_event, alarm_type, msg, compact_msg, metadata",
     [
-        [Alarms.MAJOR, "alarm", "content1", "compact_msg", {"metadata": "metadata1"}],
-        [Alarms.MINOR, "alarm", "content1", "compact_msg", {"metadata": "metadata1"}],
-        [Alarms.WARNING, "alarm", "content1", "compact_msg", {"metadata": "metadata1"}],
+        [
+            Alarms.MAJOR,
+            MessagingEvent.ALARM_MAJOR,
+            "alarm",
+            "content1",
+            "compact_msg",
+            {"metadata": "metadata1"},
+        ],
+        [
+            Alarms.MINOR,
+            MessagingEvent.ALARM_MINOR,
+            "alarm",
+            "content1",
+            "compact_msg",
+            {"metadata": "metadata1"},
+        ],
+        [
+            Alarms.WARNING,
+            MessagingEvent.ALARM_WARNING,
+            "alarm",
+            "content1",
+            "compact_msg",
+            {"metadata": "metadata1"},
+        ],
     ],
 )
-def test_redis_connector_raise_alarm(connector, severity, alarm_type, msg, compact_msg, metadata):
+def test_redis_connector_raise_alarm(
+    connector, severity, expected_event, alarm_type, msg, compact_msg, metadata
+):
     with (
         mock.patch.object(connector, "set_and_publish", return_value=None),
         mock.patch.object(connector, "notify", return_value=None),
@@ -84,7 +107,7 @@ def test_redis_connector_raise_alarm(connector, severity, alarm_type, msg, compa
         connector.set_and_publish.assert_called_once_with(
             MessageEndpoints.alarm(), AlarmMessage(severity=severity, info=info, metadata=metadata)
         )
-        connector.notify.assert_called_once_with(MessagingEvent.ALARM, compact_msg)
+        connector.notify.assert_called_once_with(expected_event, compact_msg)
 
 
 @pytest.mark.parametrize(

bec_server/bec_server/actors/scan_interlock.py

@@ -6,6 +6,8 @@
     ScanInterlockModifyStateTableMessage,
     ScanInterlockStateTableContent,
 )
+from bec_lib.messaging_hooks import MessagingEvent
+from bec_lib.messaging_services import NotificationMessageObject
 from bec_server.actors.actor import BlStateActor
 
 logger = bec_logger.logger
@@ -82,6 +84,13 @@ def mismatched_states(self):
             ]
 
     def some_mismatch_action(self, client: BECClient):
+        notification = NotificationMessageObject()
+        notification.add_text(
+            f"Scan interlock triggered for beamline states: {self.mismatched_states}", color="red"
+        )
+        notification.add_tags("scan_interlock")
+        self.client.connector.notify(MessagingEvent.SCAN_INTERLOCK, notification)
+
         if self.client.queue is None:
             return
         self.client.queue.add_queue_lock(
@@ -100,6 +109,10 @@ def _unlock(self):
         if (q := self.client.queue) is not None:
             if (curr_q := q.queue_storage.current_scan_queue) is not None:
                 if (primary := curr_q.get("primary")) is not None and primary.locks != []:
+                    notification = NotificationMessageObject()
+                    notification.add_text("Scan interlock cleared", color="green")
+                    notification.add_tags("scan_interlock")
+                    self.client.connector.notify(MessagingEvent.SCAN_INTERLOCK, notification)
                     self.client.queue.remove_queue_lock(queue="primary", lock_id=self._LOCK_ID)
 
     def run(self):

bec_server/tests/tests_scan_server/test_scan_interlock.py

@@ -3,6 +3,8 @@
 
 import pytest
 
+from bec_lib.messaging_hooks import MessagingEvent
+from bec_lib.messaging_services import NotificationMessageObject
 from bec_server.actors.scan_interlock import ScanInterlockActor
 
 
@@ -94,6 +96,11 @@ def test_on_state_modification_remove_missing(self, actor, mock_client):
 
     def test_some_mismatch_action_adds_lock(self, actor, mock_client):
         actor.some_mismatch_action(mock_client)
+        mock_client.connector.notify.assert_called_once()
+        event, notification = mock_client.connector.notify.call_args.args
+        assert event == MessagingEvent.SCAN_INTERLOCK
+        assert isinstance(notification, NotificationMessageObject)
+        assert notification._content[1].tags == ["scan_interlock"]
         mock_client.queue.add_queue_lock.assert_called_once_with(
             queue="primary",
             reason="Interlock for beamline states: []",
@@ -104,6 +111,7 @@ def test_some_mismatch_action_skips_if_no_queue(self, actor, mock_client):
         add_queue_lock = mock_client.queue.add_queue_lock
         actor.client.queue = None
         actor.some_mismatch_action(mock_client)
+        mock_client.connector.notify.assert_called_once()
         add_queue_lock.assert_not_called()
 
     def test_all_match_action_unlocks(self, actor):
@@ -114,6 +122,23 @@ def test_all_match_action_unlocks(self, actor):
 
     def test_unlock_removes_lock(self, actor, mock_client):
         actor._unlock()
+        mock_client.connector.notify.assert_called_once()
+        event, notification = mock_client.connector.notify.call_args.args
+        assert event == MessagingEvent.SCAN_INTERLOCK
+        assert isinstance(notification, NotificationMessageObject)
+        assert (
+            notification._content[0].content
+            == '<p><mark class="pen-green">Scan interlock cleared</mark></p>'
+        )
+        assert notification._content[1].tags == ["scan_interlock"]
         mock_client.queue.remove_queue_lock.assert_called_with(
             queue="primary", lock_id="ScanInterlockActor"
         )
+
+    def test_unlock_skips_notify_without_lock(self, actor, mock_client):
+        mock_client.queue.queue_storage.current_scan_queue = {"primary": MagicMock(locks=[])}
+
+        actor._unlock()
+
+        mock_client.connector.notify.assert_not_called()
+        mock_client.queue.remove_queue_lock.assert_not_called()