cleanup

csaxs_bec/bec_ipython_client/plugins/tool_box/debug_tools.py

@@ -20,17 +20,16 @@ from rich.text import Text
 from slugify import slugify
 
 if TYPE_CHECKING:
-    from bec_ipython_client.main import BECClient
+    from bec_ipython_client.main import BECIPythonClient
+    from bec_lib.devicemanager import DeviceManagerBase
     from bec_lib.scans import Scans
     from bec_widgets.cli.client_utils import BECGuiClient
 
     scans: Scans  # type: ignore[no-redef]
 
-    bec: BECClient  # type: ignore[no-redef]
+    bec: BECIPythonClient  # type: ignore[no-redef]
 
-    gui: BECGuiClient  # type: ignore[no-redef]
-
-    dev: bec.device_manager
+    dev: DeviceManagerBase  # type: ignore[no-redef]
 
 
 class Detector(BaseModel):
@@ -49,7 +48,7 @@ def to_identifier(text: str) -> str:
     name = re.sub(r"[^a-zA-Z0-9_]", "", name)
 
     if not name:
-        return "_"
+        raise ValueError(f"Cannot convert '{text}' to a valid identifier.")
 
     if name[0].isdigit():
         name = f"_{name}"
@@ -60,12 +59,10 @@ def to_identifier(text: str) -> str:
 class DebugTools:
     """A collection of debugging tools for the BEC IPython client at cSAXS."""
 
-    _PURPOSE = " ".join(
-        [
-            "Debugging helpers for the cSAXS BEC IPython client. These tools are intended for ",
-            "advanced users and developers to diagnose and troubleshoot issues within the BEC environment.",
-            "Below are the available methods together with a brief description of their functionality.",
-        ]
+    _PURPOSE = (
+        "Debugging helpers for the cSAXS BEC IPython client. These tools are intended for advanced users "
+        "and developers to diagnose and troubleshoot issues within the BEC environment. "
+        "Below are the available methods together with a brief description of their functionality."
     )
 
     ######################
@@ -108,17 +105,12 @@ class DebugTools:
     ### MCS Card Check ###
     #####################
 
-    def _check_if_mcs_card_is_loaded(self):
-        """Check if the MCS card device is loaded in the current BEC session."""
-        if "mcs" not in dev:
-            raise RuntimeError("MCS device is not loaded in the current active BEC session.")
-
-    def _check_if_ddg_is_loaded(self):
-        """Check if the DDG1 device is loaded in the current BEC session."""
-        if "ddg1" not in dev:
-            raise RuntimeError("DDG1 device is not loaded in the current active BEC session.")
-        if "ddg2" not in dev:
-            raise RuntimeError("DDG2 device is not loaded in the current active BEC session.")
+    def _check_if_device_is_loaded(self, device_name: str):
+        """Check if a device is loaded in the current BEC session."""
+        if device_name not in dev:
+            raise RuntimeError(
+                f"Device {device_name} was not loaded in the current active BEC session."
+            )
 
     def mcs_test_acquire(
         self, mode: Literal["high_frame", "medium_frame", "low_frame"] = "high_frame"
@@ -133,14 +125,15 @@ class DebugTools:
                 - 'medium_frame': Tests medium frame rates with moderate exposure times.
                 - 'low_frame': Tests low frame rates with longer exposure times.
         """
-        self._check_if_mcs_card_is_loaded()
-        self._check_if_ddg_is_loaded()
+        self._check_if_device_is_loaded("mcs")
+        self._check_if_device_is_loaded("ddg1")
+        self._check_if_device_is_loaded("ddg2")
 
         if mode == "high_frame":
-            burst_frames = np.random.randint(10000, 100000)  # between 10000 and 100000
+            burst_frames = np.random.randint(10_000, 100_000)  # between 10000 and 100000
             cycles = np.random.randint(5, 20)  # between 5 and 20
             exp_time = (
-                np.random.rand() * (0.001 - 0.000201) + 0.000201
+                np.random.rand() * (0.001 - 0.201e-3) + 0.201e-3
             )  # between 0.000201 ms and 0.001 s
         elif mode == "medium_frame":
             burst_frames = np.random.randint(50, 500)  # between 50 and 500
@@ -153,7 +146,7 @@ class DebugTools:
         else:
             raise ValueError(f"Invalid mode '{mode}' specified for acquire scan test.")
         print(
-            f"Starting acquire measurement with exp_time={exp_time}, burst_frames={burst_frames}, cycles={cycles}"
+            f"Starting acquire measurement with exp_time={exp_time:.6f}, burst_frames={burst_frames}, cycles={cycles}"
         )
         s = scans.acquire(
             exp_time=exp_time, frames_per_trigger=burst_frames, burst_at_each_point=cycles
@@ -246,8 +239,8 @@ class DebugTools:
             return False
 
     def open_it_service_page(self):
-        """Open the overview of IT services hosted by Science IT (Leo Sala) for cSAXS."""
-        gui = bec.gui
+        """Open the overview of IT services hosted by Science IT Infrastructure and Services for cSAXS."""
+        gui: BECGuiClient = bec.gui
         dock_area = gui.new()
         print("Opening IT service page in new dock...")
         url = "https://metrics.psi.ch/d/saf8mxv/x12sa?orgId=1&from=now-24h&to=now&timezone=browser&var-receiver_hosts=sls-jfjoch-001.psi.ch&var-writer_hosts=xbl-daq-34.psi.ch&var-beamline=X12SA&var-slurm_partitions=csaxs&var-receiver_services=broker&var-writer_services=writer&refresh=15m"

csaxs_bec/devices/epics/mcs_card/mcs_card_csaxs.py

@@ -233,7 +233,7 @@ class MCSCardCSAXS(PSIDeviceBase, MCSCard):
         # Start monitoring thread
         self._scan_done_thread.start()
 
-    def _on_counter_update(self, value, **kwargs) -> None:
+    def _on_counter_update(self, value: float | np.ndarray, **kwargs) -> None:
         """
         Callback for counter updates of the mca channels (1-32). This callback is attached
         to each mca channel PV on the MCS card. It collects data from all channels