Send more data back from smart scan to enable if a zip was created

src/openflexure_microscope_server/things/smart_scan.py

@@ -17,6 +17,7 @@ from typing import (
     Any,
     Callable,
     Concatenate,
+    Literal,
     Mapping,
     Optional,
     ParamSpec,
@@ -52,6 +53,15 @@ AnyModel = Annotated[
 ]
 
 
+class LiveScanDetails(BaseModel):
+    """Details of ongoing scan used for the UI."""
+
+    name: str
+    stitch_timestamp: Optional[float] = None
+    image_count: int = 0
+    scan_phase: Literal["Setup", "Scanning", "Stitching", "Complete"] = "Setup"
+
+
 class ActiveScanData(scan_directories.BaseScanData):
     """A model for the scan data during an ongoing scan.
 
@@ -277,12 +287,21 @@ class SmartScanThing(OFMThing):
 
     _preview_stitcher: Optional[stitching.PreviewStitcher] = None
 
-    _latest_scan_name: Optional[str] = None
+    _latest_scan_live_details: Optional[LiveScanDetails] = None
 
     @lt.property
-    def latest_scan_name(self) -> Optional[str]:
-        """The name of the last scan to be started."""
-        return self._latest_scan_name
+    def latest_scan_live_details(self) -> Optional[LiveScanDetails]:
+        """The details of any ongoing scan, or the previous scan if no san ongoing."""
+        if self._latest_scan_live_details is None:
+            return None
+        # Update image count and stitch timestamp if scan is ongoing.
+        if self._scan_data is not None:
+            self._latest_scan_live_details.image_count = self._scan_data.image_count
+
+        # Use walrus operator to not need to read filesystem timestamp twice
+        if (stitch_time := self.latest_preview_stitch_time) is not None:
+            self._latest_scan_live_details.stitch_timestamp = stitch_time
+        return self._latest_scan_live_details
 
     @lt.action
     def sample_scan(self, scan_name: str = "") -> None:
@@ -306,7 +325,9 @@ class SmartScanThing(OFMThing):
 
             workflow.check_before_start(scan_name)
             self._ongoing_scan = self._scan_dir_manager.new_scan_dir(scan_name)
-            self._latest_scan_name = self.ongoing_scan.name
+            self._latest_scan_live_details = LiveScanDetails(
+                name=self.ongoing_scan.name
+            )
             self._run_scan(workflow)
             # Save any final messages.
             self._ongoing_scan.save_scan_log(self)
@@ -328,6 +349,12 @@ class SmartScanThing(OFMThing):
             # Reraise so the action reports as cancelled if stitching is cancelled.
             raise e
         finally:
+            # Save some final details but not set to the base model until after
+            # resetting valued and unlocking just in case somehow pydantic errors.
+            if self._scan_data is not None:
+                final_image_count = self._scan_data.image_count
+            if self._preview_stitcher is not None:
+                last_timestamp = self.latest_preview_stitch_time
             # However the scan finishes, unset all variables and release lock
             self._ongoing_scan = None
             self._scan_data = None
@@ -335,6 +362,11 @@ class SmartScanThing(OFMThing):
             # Ensure any PreviewStitcher created cannot be reused.
             self._preview_stitcher = None
 
+            # Set the final details so they persist until the next scan
+            self._latest_scan_live_details.image_count = final_image_count
+            self._latest_scan_live_details.stitch_timestamp = last_timestamp
+            self._latest_scan_live_details.scan_phase = "Complete"
+
             # Remove any scan folders containing zero images.
             self.purge_empty_scans()
 
@@ -449,6 +481,7 @@ class SmartScanThing(OFMThing):
                     correlation_resize=stitching_settings.correlation_resize,
                 )
 
+            self._latest_scan_live_details.scan_phase = "Scanning"
             # This is the main loop of the scan!
             self._main_scan_loop(workflow)
             self._save_final_scan_data(scan_result="success")
@@ -553,6 +586,7 @@ class SmartScanThing(OFMThing):
     @_scan_running
     def _perform_final_stitch(self) -> None:
         """Update the scan zip and perform final stitch of the data."""
+        self._latest_scan_live_details.scan_phase = "Stitching"
         if self.scan_data.image_count <= 1:
             self.logger.info("Not performing a stitch as not enough images captured")
             return
@@ -676,14 +710,16 @@ class SmartScanThing(OFMThing):
     @property
     def latest_preview_stitch_path(self) -> Optional[str]:
         """The path of the latest preview stitched image, or None if not available."""
-        if self.latest_scan_name is None:
+        if self._latest_scan_live_details is None:
             return None
 
         return self._scan_dir_manager.get_file_path_from_img_dir(
-            scan_name=self.latest_scan_name, filename="preview.jpg", check_exists=True
+            scan_name=self._latest_scan_live_details.name,
+            filename="preview.jpg",
+            check_exists=True,
         )
 
-    @lt.property
+    @property
     def latest_preview_stitch_time(self) -> Optional[float]:
         """The modification time of the latest preview image, to allow live updating.