f

Author: wakonig

bec_lib/bec_lib/utils/scan_utils.py

@@ -7,7 +7,8 @@
 import csv
 import datetime
 from collections import defaultdict
-from typing import TYPE_CHECKING
+from collections.abc import Mapping
+from typing import TYPE_CHECKING, Any
 
 from typeguard import typechecked
 
@@ -62,6 +63,46 @@ def scan_to_csv(
     _write_csv(output_name=output_name, delimiter=delimiter, dialect=dialect, output=data_output)
 
 
+def compose_cli_input_from_scan_info(scan_info: Mapping[str, Any] | Any) -> str:
+    """Compose a user-facing scan call string from a scan info object."""
+    scan_name = _get_scan_info_value(scan_info, "scan_name")
+    if not scan_name:
+        return "scans.unknown()"
+
+    request_inputs = _get_scan_info_value(scan_info, "request_inputs") or {}
+    arg_bundle = list(request_inputs.get("arg_bundle") or [])
+    inputs = request_inputs.get("inputs") or {}
+    kwargs = request_inputs.get("kwargs") or {}
+
+    call_parts: list[str] = [_render_scan_cli_value(value) for value in arg_bundle]
+    call_parts.extend(f"{name}={_render_scan_cli_value(value)}" for name, value in inputs.items())
+    call_parts.extend(f"{name}={_render_scan_cli_value(value)}" for name, value in kwargs.items())
+    return f"scans.{scan_name}({', '.join(call_parts)})"
+
+
+def _get_scan_info_value(scan_info: Mapping[str, Any] | Any, key: str) -> Any:
+    if isinstance(scan_info, Mapping):
+        return scan_info.get(key)
+    return getattr(scan_info, key, None)
+
+
+def _render_scan_cli_value(value: Any) -> str:
+    if isinstance(value, Mapping):
+        rendered_items = ", ".join(
+            f"{_render_scan_cli_value(key)}: {_render_scan_cli_value(val)}"
+            for key, val in value.items()
+        )
+        return "{" + rendered_items + "}"
+    if isinstance(value, tuple):
+        inner = ", ".join(_render_scan_cli_value(item) for item in value)
+        if len(value) == 1:
+            inner += ","
+        return f"({inner})"
+    if isinstance(value, list):
+        return "[" + ", ".join(_render_scan_cli_value(item) for item in value) + "]"
+    return repr(value)
+
+
 def _write_csv(output_name: str, delimiter: str, output: list, dialect: str = None) -> None:
     """Write csv file.
 

bec_server/bec_server/scan_server/generator_scan_worker.py

@@ -12,6 +12,7 @@
 from bec_lib.file_utils import compile_file_components
 from bec_lib.logger import bec_logger
 from bec_lib.messaging_hooks import MessagingEvent
+from bec_lib.utils.scan_utils import compose_cli_input_from_scan_info
 
 from .errors import DeviceInstructionError, ScanAbortion, UserScanInterruption
 from .scan_queue import InstructionQueueItem, InstructionQueueStatus, RequestBlock
@@ -342,13 +343,20 @@ def _send_scan_status(
         self.worker.device_manager.connector.set_and_publish(
             MessageEndpoints.scan_status(), msg, pipe=pipe
         )
+        cli_input = compose_cli_input_from_scan_info(self.current_scan_info)
+        scan_number = self.current_scan_info.get("scan_number")
+        scan_id = self.current_scan_id
         if status == "open":
             self.worker.device_manager.connector.notify(
-                MessagingEvent.SCAN, f"Scan started: {self.current_scan_id}", pipe=pipe
+                MessagingEvent.SCAN,
+                f"Scan started: scan_number={scan_number} ({cli_input}, scan_id={scan_id})",
+                pipe=pipe,
             )
         elif status in {"closed", "user_completed"}:
             self.worker.device_manager.connector.notify(
-                MessagingEvent.SCAN_COMPLETED, f"Scan completed: {self.current_scan_id}", pipe=pipe
+                MessagingEvent.SCAN_COMPLETED,
+                f"Scan completed: scan_number={scan_number} ({cli_input}, scan_id={scan_id})",
+                pipe=pipe,
             )
         pipe.execute()
 

bec_server/bec_server/scan_server/scans/scan_actions.py

@@ -15,6 +15,7 @@
 from bec_lib.file_utils import compile_file_components
 from bec_lib.logger import bec_logger
 from bec_lib.messaging_hooks import MessagingEvent
+from bec_lib.utils.scan_utils import compose_cli_input_from_scan_info
 from bec_server.scan_server.scan_stubs import ScanStubStatus
 
 if TYPE_CHECKING:
@@ -1063,14 +1064,19 @@ def _send_scan_status(
             MessageEndpoints.public_scan_info(scan.scan_info.scan_id), msg, pipe=pipe, expire=expire
         )
         self._connector.set_and_publish(MessageEndpoints.scan_status(), msg, pipe=pipe)
+        cli_input = compose_cli_input_from_scan_info(scan.scan_info)
+        scan_number = scan.scan_info.scan_number
+        scan_id = scan.scan_info.scan_id
         if status == "open":
             self._connector.notify(
-                MessagingEvent.SCAN, f"Scan started: {scan.scan_info.scan_id}", pipe=pipe
+                MessagingEvent.SCAN,
+                f"Scan started: scan_number={scan_number} ({cli_input}, scan_id={scan_id})",
+                pipe=pipe,
             )
         elif status in {"closed", "user_completed"}:
             self._connector.notify(
                 MessagingEvent.SCAN_COMPLETED,
-                f"Scan completed: {scan.scan_info.scan_id}",
+                f"Scan completed: scan_number={scan_number} ({cli_input}, scan_id={scan_id})",
                 pipe=pipe,
             )
         pipe.execute()

bec_server/tests/tests_scan_server/scans_v4/test_scan_actions.py

@@ -11,6 +11,7 @@
 from bec_lib.messaging_hooks import MessagingEvent
 from bec_lib.tests.fixtures import dm_with_devices  # noqa: F401
 from bec_lib.tests.utils import ConnectorMock
+from bec_lib.utils.scan_utils import compose_cli_input_from_scan_info
 from bec_server.scan_server.instruction_handler import InstructionHandler
 from bec_server.scan_server.scan_stubs import ScanStubStatus
 from bec_server.scan_server.scans.scan_base import ScanBase, ScanInfo
@@ -107,6 +108,40 @@ def test_scan_info_stores_scan_report_device_objects_as_names(dm_with_devices):
     assert scan_info.scan_report_devices == ["samz"]
 
 
+def test_compose_cli_input_from_scan_info_uses_named_inputs():
+    scan_info = ScanInfo(
+        scan_name="_v4_test_scan",
+        scan_id="scan-id-test",
+        scan_type=None,
+        request_inputs={
+            "arg_bundle": [],
+            "inputs": {"device": "samx", "target": 1},
+            "kwargs": {"relative": True},
+        },
+    )
+
+    assert (
+        compose_cli_input_from_scan_info(scan_info)
+        == "scans._v4_test_scan(device='samx', target=1, relative=True)"
+    )
+
+
+def test_compose_cli_input_from_scan_info_uses_arg_bundle():
+    scan_info = {
+        "scan_name": "line_scan",
+        "request_inputs": {
+            "arg_bundle": ["samx", -5, 5, 3],
+            "inputs": {},
+            "kwargs": {"exp_time": 0.1},
+        },
+    }
+
+    assert (
+        compose_cli_input_from_scan_info(scan_info)
+        == "scans.line_scan('samx', -5, 5, 3, exp_time=0.1)"
+    )
+
+
 def _last_device_instruction(ctx, action):
     return _sent_device_instructions(ctx, action)[-1]
 
@@ -531,6 +566,7 @@ def test_send_scan_status_publishes_message(action_context):
     ctx.connector.notify = mock.MagicMock()
     status_msg = messages.ScanStatusMessage(scan_id="scan-id-test", status="closed", info={})
     ctx.actions._build_scan_status_message = mock.MagicMock(return_value=status_msg)
+    ctx.scan.scan_info.request_inputs = {"arg_bundle": [], "inputs": {}, "kwargs": {}}
 
     ctx.actions._send_scan_status("closed", reason="alarm")
 
@@ -542,7 +578,9 @@ def test_send_scan_status_publishes_message(action_context):
         MessageEndpoints.scan_status(), status_msg, pipe=pipe
     )
     ctx.connector.notify.assert_called_once_with(
-        MessagingEvent.SCAN_COMPLETED, "Scan completed: scan-id-test", pipe=pipe
+        MessagingEvent.SCAN_COMPLETED,
+        "Scan completed: scan_number=1 (scans._v4_test_scan(), scan_id=scan-id-test)",
+        pipe=pipe,
     )
     pipe.execute.assert_called_once_with()
 
@@ -556,11 +594,18 @@ def test_send_scan_status_publishes_new_scan_notification(action_context):
     ctx.connector.notify = mock.MagicMock()
     status_msg = messages.ScanStatusMessage(scan_id="scan-id-test", status="open", info={})
     ctx.actions._build_scan_status_message = mock.MagicMock(return_value=status_msg)
+    ctx.scan.scan_info.request_inputs = {
+        "arg_bundle": [],
+        "inputs": {"device": "samx"},
+        "kwargs": {},
+    }
 
     ctx.actions._send_scan_status("open")
 
     ctx.connector.notify.assert_called_once_with(
-        MessagingEvent.SCAN, "Scan started: scan-id-test", pipe=pipe
+        MessagingEvent.SCAN,
+        "Scan started: scan_number=1 (scans._v4_test_scan(device='samx'), scan_id=scan-id-test)",
+        pipe=pipe,
     )
 
 

bec_server/tests/tests_scan_server/test_generator_scan_worker.py

@@ -403,7 +403,15 @@ def test_send_scan_status(generator_worker_mock, status, expire):
     worker = generator_worker_mock
     worker.worker.device_manager.connector = ConnectorMock()
     worker.current_scan_id = str(uuid.uuid4())
-    worker.current_scan_info = {"scan_number": 5}
+    worker.current_scan_info = {
+        "scan_number": 5,
+        "scan_name": "line_scan",
+        "request_inputs": {
+            "arg_bundle": ["samx", -5, 5, 3],
+            "inputs": {},
+            "kwargs": {"exp_time": 0.1},
+        },
+    }
     worker._send_scan_status(status)
     scan_info_msgs = [
         msg
@@ -423,10 +431,16 @@ def test_send_scan_status(generator_worker_mock, status, expire):
     ]
     if status == "open":
         assert len(notification_msgs) == 1
-        assert notification_msgs[0]["msg"].message == f"Scan started: {worker.current_scan_id}"
+        assert (
+            notification_msgs[0]["msg"].message == "Scan started: scan_number=5 "
+            f"(scans.line_scan('samx', -5, 5, 3, exp_time=0.1), scan_id={worker.current_scan_id})"
+        )
     elif status == "closed":
         assert len(notification_msgs) == 1
-        assert notification_msgs[0]["msg"].message == f"Scan completed: {worker.current_scan_id}"
+        assert (
+            notification_msgs[0]["msg"].message == "Scan completed: scan_number=5 "
+            f"(scans.line_scan('samx', -5, 5, 3, exp_time=0.1), scan_id={worker.current_scan_id})"
+        )
     else:
         assert notification_msgs == []