src/aare/daq/daq.py

@@ -6,6 +6,7 @@ from collections.abc import Callable
 from datetime import UTC, datetime
 from math import ceil
 from pathlib import Path
+from typing import Any
 
 from aarecommon.config.beamline import cfg_get
 from aarecommon.config.logger import setup_logger
@@ -303,10 +304,13 @@ class AareDAQ:
             pgroup_provider=_DAQPGroupProvider(self),
         )
 
-    def _cached_detector_metadata(self) -> dict:
+    def shutdown(self):
+        self._devs.bec_worker.shutdown()
+
+    def _cached_detector_metadata(self) -> dict[str, Any]:
         return self._cfg.cached_detector_metadata
 
-    def refresh_detector_metadata_cache(self) -> dict[str, object]:
+    def refresh_detector_metadata_cache(self) -> dict[str, Any]:
         det_cfg = self._jfjoch.detector()
         dtz_low = self._cfg._coerce_optional_float(self._devs.dtz_low)
         dtz_high = self._cfg._coerce_optional_float(self._devs.dtz_high)

src/aare/daq/devices.py

@@ -34,9 +34,7 @@ class BeamlineDevices:
         BEAMLINE = beamline.value.upper()
         self.tell = make_tell_client(beamline)
         self.aerotech = aerotech.AerotechController(beamline)
-        logger.debug("initialising BEC worker")
         self.bec_worker = BECClientWorker(beamline)
-        logger.debug("initialising BEC worker done")
         self._smargon = smargon.Smargon(beamline)
         self.exp_shutter = ExperimentalHutchShutter(beamline=self._beamline)
         # Personnel Safety System: gates whether the robot is allowed to move.

src/aare/daq/server.py

@@ -134,6 +134,7 @@ async def lifespan(application: FastAPI):
 
     # Shutdown: add cleanup here if needed
     logger.info(f"Worker {os.getpid()} shutting down.")
+    daq.shutdown()
 
 
 app = FastAPI(lifespan=lifespan)
@@ -2530,6 +2531,8 @@ def main():
         proxy_headers=False,
         log_config=get_uvicorn_logging_config(),
         timeout_worker_healthcheck=30,
+        timeout_graceful_shutdown=1,
+        timeout_keep_alive=30,
     )
 
 

src/aare/devices/bec_worker.py

@@ -11,7 +11,6 @@ from aarecommon.models.models import BeamlineStateEnum
 from bec_ipython_client import BECIPythonClient
 from bec_ipython_client.signals import OperationMode
 from bec_lib.device import RPCError, ScanRequestError
-from bec_lib.procedures.helper import FrontendProcedureHelper
 from bec_lib.service_config import ServiceConfig
 
 from aare.beamline_dispatch.beamline_dispatch import get_beamline_dispatch
@@ -70,6 +69,7 @@ def _bec_state_to_aare_state(bec_state: BeamlineState) -> BeamlineStateEnum:
 
 class BECClientWorker:
     def __init__(self, beamline: MXBeamline, name: str = "default"):
+        logger.debug(f"initialising BEC worker for {beamline}")
         BEAMLINE = beamline.value.lower()
         self.beamline = beamline
         if self.beamline is MXBeamline.X06DA:
@@ -87,31 +87,29 @@ class BECClientWorker:
 
         if self.beamline is MXBeamline.SIMULATED:
             self.simulated = True
-
+            host = "localhost"
         else:
             self.simulated = False
-            logger.debug(f"Initializing BECClientWorker for {BEAMLINE} beamline")
             host = cfg_get("daq.hardware.bec_url", f"{BEAMLINE}-bec-001.psi.ch")
-            service_config = ServiceConfig(redis={"host": host, "port": 6379})
-            service_config.config["log_writer"]["base_path"] = "/tmp/logs"
-            # service_config.config["user_macros"]["base_path"]=f'/sls/{BEAMLINE}/config/bec/production/pxiii_bec/pxiii_bec'
-            # print(service_config.config)
+        service_config = ServiceConfig(redis={"host": host, "port": 6379})
+        service_config.config["log_writer"]["base_path"] = "/tmp/logs"
+        try:
             self.client = BECIPythonClient(config=service_config, mode=OperationMode.Procedure)
             self.client.start()
-            # self.client.config.update_session_with_file("/sls/x10sa/config/bec/production/bec/bec_lib/bec_lib/config_helper.py")
             self.dev = self.client.device_manager.devices
-            print(self.dev.keys())
             self.scans = self.client.scans
             self.macros = self.dispatch.bec_macros
-            self.helper = FrontendProcedureHelper(self.client.connector)
             self._set_scilog_tags()
-            try:
-                self._init_beamline_environment()
-            except Exception:
-                logger.exception("Error initialising BEC devices")
-                sys.exit(1)
+            self._init_beamline_environment()
+        except Exception:
+            logger.exception("Error initialising BEC devices")
+            self.client.shutdown()
+            sys.exit(1)
         logger.debug(f"simulated is {self.simulated}")
 
+    def shutdown(self):
+        self.client.shutdown()
+
     def _init_beamline_environment(self):
         try:
             self.position_devices, self.planner = self.macros.init_beamline_environment()
@@ -246,28 +244,6 @@ class BECClientWorker:
         except Exception:
             logger.exception("Error sending scilog message")
 
-    def run_macro(self, macro_name: str, *args, queue: str = "default", **kwargs):
-        if self.simulated:
-            logger.debug(f"Simulating macro {macro_name}")
-            return None
-        try:
-            return self.client.proc.run_macro(macro_name, *args, queue=queue)
-        except Exception as e:
-            raise self._bec_error(e, operation=f"run_macro:{macro_name}") from e
-
-    def run_macro_blocked(self, macro_name: str, *args, queue: str = "default", **kwargs):
-        if self.simulated:
-            logger.debug(f"Simulating macro {macro_name}")
-            return None
-        try:
-            status = self.run_macro(macro_name, *args, queue=queue)
-            print(status)
-            status.wait()
-            print(status)
-            return status
-        except Exception as e:
-            raise self._bec_error(e, operation=f"run_macro_blocked:{macro_name}") from e
-
     @log_timing(logger, "BEC move_to")
     def move_to(self, state: BeamlineState):
         start = time.perf_counter()