segmenter: Introduce live segmentation

controller/imager/main.py

@@ -382,6 +382,7 @@ def __init__(self) -> None:
         self._stop_receiving_mqtt = threading.Event()  # close() was called
         self._done = threading.Event()  # run_discrete() finished or stop() was called
         self._discrete_run = threading.Lock()  # mutex on starting the pump
+        self._waiting_for_pump = False  # FIX: Only accept Done after pump command sent
 
     def open(self) -> None:
         """Start the pump MQTT client.
@@ -412,7 +413,17 @@ def _receive_messages(self) -> None:
                 self._mqtt.read_message()
                 continue
 
+            # FIX: Only process Done if we are actually waiting for pump to finish
+            # This prevents retained/stale Done messages from triggering early return
+            if not self._waiting_for_pump:
+                loguru.logger.debug(
+                    f"Ignoring pump Done (not waiting for pump): {self._mqtt.msg['payload']}"
+                )
+                self._mqtt.read_message()
+                continue
+
             loguru.logger.debug(f"The pump has stopped: {self._mqtt.msg['payload']}")
+            self._waiting_for_pump = False  # FIX: Clear waiting flag
             self._mqtt.client.unsubscribe("status/pump")
             self._mqtt.read_message()
             self._done.set()
@@ -436,6 +447,7 @@ def run_discrete(self, settings: stopflow.DiscretePumpSettings) -> None:
         # We ignore the pylint error here because the lock can only be released from a different
         # thread (the thread which calls the `handle_status_update()` method):
         self._discrete_run.acquire()  # pylint: disable=consider-using-with
+        self._waiting_for_pump = False  # FIX: Not waiting yet (ignore retained messages)
         self._done.clear()
         self._mqtt.client.subscribe("status/pump")
         self._mqtt.client.publish(
@@ -446,9 +458,11 @@ def run_discrete(self, settings: stopflow.DiscretePumpSettings) -> None:
                     "direction": settings.direction.value,
                     "flowrate": settings.flowrate,
                     "volume": settings.volume,
+                    "from_acquisition": True,  # FIX: Tag as acquisition command
                 }
             ),
         )
+        self._waiting_for_pump = True  # FIX: NOW we're waiting for pump Done
         self._done.wait()
 
     def stop(self) -> None:
@@ -457,7 +471,7 @@ def stop(self) -> None:
             raise RuntimeError("MQTT client was not initialized yet!")
 
         self._mqtt.client.subscribe("status/pump")
-        self._mqtt.client.publish("actuator/pump", '{"action": "stop"}')
+        self._mqtt.client.publish("actuator/pump", '{"action": "stop", "from_acquisition": true}')
 
     def close(self) -> None:
         """Close the pump MQTT client, if it's currently open.

controller/imager/stopflow.py

@@ -133,7 +133,10 @@ def run_step(self) -> typing.Optional[tuple[int, str]]:
                 + f"{capture_path}...",
             )
             self._camera.capture_file(capture_path)
-            os.sync()
+            # FIX: Use fsync on specific file to ensure write completes before MQTT publish
+            # os.sync() is system-wide and async - doesn't guarantee this file is written
+            with open(capture_path, "rb") as f:
+                os.fsync(f.fileno())
             # Note(ethanjli): updating the integrity file is the responsibility of the code which
             # calls this `run_step()` method.
 

controller/pump/main.py

@@ -21,6 +21,8 @@
 pump_max_speed = 50
 
 pump_started = False
+# FIX: Track acquisition state to prevent UI commands from interrupting acquisition
+acquisition_in_progress = False
 
 pump_stepper = Motor(pin=23, spi_bus=0, spi_device=0)
 pump_stepper.acceleration = 2000
@@ -52,19 +54,40 @@ async def start() -> None:
     async with client, task_group:
         _ = await asyncio.gather(
             client.subscribe("actuator/pump"),
+            client.subscribe("status/imager"),  # FIX: Track acquisition state
             # publish_status(),
         )
         async for message in client.messages:
             task_group.create_task(handle_message(message))
 
 
 async def handle_message(message) -> None:
+    global acquisition_in_progress
+
+    # FIX: Handle imager status messages to track acquisition state
+    if message.topic.matches("status/imager"):
+        payload = json.loads(message.payload.decode("utf-8"))
+        status = payload.get("status", "")
+        if status == "Started":
+            print("Acquisition started - locking pump for acquisition commands only")
+            acquisition_in_progress = True
+        elif status in ("Done", "Interrupted"):
+            print("Acquisition ended - unlocking pump for manual control")
+            acquisition_in_progress = False
+        return
+
     if not message.topic.matches("actuator/pump"):
         return
 
     payload = json.loads(message.payload.decode("utf-8"))
     pprint(payload)
 
+    # FIX: During acquisition, only accept commands tagged with from_acquisition
+    is_from_acquisition = payload.get("from_acquisition", False)
+    if acquisition_in_progress and not is_from_acquisition:
+        print(f"Ignoring pump command during acquisition (use from_acquisition flag): {payload}")
+        return
+
     action = payload.get("action")
     if action is not None:
         await handle_action(action, payload)

node-red/projects/dashboard/flows.json

@@ -2183,8 +2183,8 @@
         "initialize": "",
         "finalize": "",
         "libs": [],
-        "x": 650,
-        "y": 180,
+        "x": 710,
+        "y": 140,
         "wires": [
             []
         ]
@@ -2557,7 +2557,7 @@
         "templateScope": "widget:ui",
         "className": "",
         "x": 800,
-        "y": 180,
+        "y": 200,
         "wires": [
             [
                 "1db1c2e3e19e85ff"