Add mechanism to record when a multigrid controller has exited

Author: d-j-hatton

src/murfey/client/multigrid_control.py

@@ -34,6 +34,8 @@ class MultigridController:
     rsync_url: str = ""
     rsync_module: str = "data"
     demo: bool = False
+    dormant: bool = False
+    multigrid_watcher_active: bool = True
     processing_enabled: bool = True
     do_transfer: bool = True
     dummy_dc: bool = False
@@ -95,6 +97,9 @@ def __post_init__(self):
             register_client=False,
         )
 
+    def _multigrid_watcher_finalised(self):
+        self.multigrid_watcher_active = False
+
     def _start_rsyncer_multigrid(
         self,
         source: Path,

src/murfey/client/watchdir_multigrid.py

@@ -114,3 +114,5 @@ def _process(self):
             if first_loop:
                 first_loop = False
             time.sleep(15)
+
+        self.notify(final=True)

src/murfey/instrument_server/api.py

@@ -176,6 +176,7 @@ def start_multigrid_watcher(
             destination_overrides=watcher_spec.destination_overrides,
         )
     )
+    watchers[session_id].subscribe(controllers[session_id]._multigrid_watcher_finalised)
     watchers[session_id].start()
     return {"success": True}
 

src/murfey/util/__init__.py

@@ -228,19 +228,31 @@ class Observer:
     def __init__(self):
         self._listeners: list[Callable[..., Awaitable[None] | None]] = []
         self._secondary_listeners: list[Callable[..., Awaitable[None] | None]] = []
+        self._final_listeners: list[Callable[..., Awaitable[None] | None]] = []
         super().__init__()
 
     def subscribe(
-        self, fn: Callable[..., Awaitable[None] | None], secondary: bool = False
+        self,
+        fn: Callable[..., Awaitable[None] | None],
+        secondary: bool = False,
+        final: bool = False,
     ):
-        if secondary:
+        if final:
+            self._final_listeners.append(fn)
+        elif secondary:
             self._secondary_listeners.append(fn)
         else:
             self._listeners.append(fn)
 
-    async def anotify(self, *args, secondary: bool = False, **kwargs) -> None:
+    async def anotify(
+        self, *args, secondary: bool = False, final: bool = False, **kwargs
+    ) -> None:
         awaitables: list[Awaitable] = []
-        listeners = self._secondary_listeners if secondary else self._listeners
+        listeners = (
+            self._secondary_listeners
+            if secondary
+            else self._final_listeners if final else self._listeners
+        )
         for notify_function in listeners:
             result = notify_function(*args, **kwargs)
             if result is not None and inspect.isawaitable(result):
@@ -253,9 +265,15 @@ async def _await_all(awaitables: list[Awaitable]):
         for awaitable in asyncio.as_completed(awaitables):
             await awaitable
 
-    def notify(self, *args, secondary: bool = False, **kwargs) -> None:
+    def notify(
+        self, *args, secondary: bool = False, final: bool = False, **kwargs
+    ) -> None:
         awaitables: list[Awaitable] = []
-        listeners = self._secondary_listeners if secondary else self._listeners
+        listeners = (
+            self._secondary_listeners
+            if secondary
+            else self._final_listeners if final else self._listeners
+        )
         for notify_function in listeners:
             result = notify_function(*args, **kwargs)
             if result is not None and inspect.isawaitable(result):