Merge branch 'save-rom-data' into 'v3'

Save useful ROM data to data directory

See merge request openflexure/openflexure-microscope-server!590

src/openflexure_microscope_server/things/__init__.py

@@ -5,6 +5,7 @@ with other Things and including them in the LabThings-FastAPI config file.
 """
 
 import os
+from types import TracebackType
 from typing import Optional, Self
 
 import labthings_fastapi as lt
@@ -35,6 +36,19 @@ class OFMThing(lt.Thing):
         )
         return self
 
+    def __exit__(
+        self,
+        _exc_type: type[BaseException],
+        _exc_value: Optional[BaseException],
+        _traceback: Optional[TracebackType],
+    ) -> None:
+        """Close the OFMThing.
+
+        This is needed for the context manager protocol to work. Currently it doesn't
+        do anything.
+        """
+        pass
+
     @property
     def data_dir(self) -> str:
         """The data directory for this thing."""

src/openflexure_microscope_server/things/stage_measure.py

@@ -13,8 +13,11 @@ is tracked and an error is raised if it exceeds a minimum amount. Currently
 this is 10% of the expected motion is the measured axis.
 """
 
+import json
+import os
 import time
 from copy import copy
+from datetime import datetime
 from threading import Lock
 from typing import Any, Literal, Mapping, Optional, overload
 
@@ -24,6 +27,8 @@ import labthings_fastapi as lt
 from camera_stage_mapping import fft_image_tracking
 
 # Things
+from openflexure_microscope_server.things import OFMThing
+
 from .autofocus import AutofocusThing
 from .camera import BaseCamera
 from .camera_stage_mapping import CameraStageMapper
@@ -114,7 +119,7 @@ class ParasiticMotionError(lt.exceptions.InvocationError):
     """
 
 
-class RangeofMotionThing(lt.Thing):
+class RangeofMotionThing(OFMThing):
     """A class used to measure the range of motion of the stage in X and Y."""
 
     _class_settings = {"validate_properties_on_set": True}
@@ -162,6 +167,9 @@ class RangeofMotionThing(lt.Thing):
             axes: tuple[Literal["x"], Literal["y"]] = ("x", "y")
             directions: tuple[Literal[1], Literal[-1]] = (1, -1)
 
+            stage_dic = {}
+            cor_dic = {}
+
             for axis in axes:
                 # The final position of the axis under test in the given direction
                 axis_limits = []
@@ -171,6 +179,9 @@ class RangeofMotionThing(lt.Thing):
                     self._move_until_edge(axis=axis, direction=axis_dir)
                     # Append final position from tracker
                     axis_limits.append(self._rom_data.final_position[axis])
+                    axis_string = "positive" if axis_dir == 1 else "negative"
+                    stage_dic[f"{axis}_{axis_string}"] = self._rom_data.stage_coords
+                    cor_dic[f"{axis}_{axis_string}"] = self._rom_data.offsets
                 # Calculate step range.
                 step_range.append(abs(axis_limits[0] - axis_limits[1]))
 
@@ -178,7 +189,22 @@ class RangeofMotionThing(lt.Thing):
             self.logger.info(
                 f"Range of motion is {step_range[0]} x {step_range[1]} steps"
             )
-            self.calibrated_range = (step_range[0], step_range[1])
+            data = {
+                "calibrated_range": [step_range[0], step_range[1]],
+                "stage_coords": stage_dic,
+                "correlations": cor_dic,
+                "csm_matrix": self._csm.image_to_stage_displacement_matrix,
+                "time": total_time,
+            }
+
+            if not os.path.exists(self.data_dir):
+                os.makedirs(self.data_dir)
+
+            timestamp = datetime.now().strftime("%Y_%m_%d_%H_%M")
+            datafile_path = os.path.join(self.data_dir, f"rom_data_{timestamp}.json")
+            with open(datafile_path, "w", encoding="utf-8") as f:
+                json.dump(data, f, indent=4)
+
         finally:
             self._lock.release()
         return {

tests/unit_tests/test_stage_measure.py

@@ -1,6 +1,7 @@
 """File contains unit tests for stage_measure."""
 
 import logging
+import tempfile
 from copy import copy
 
 import numpy as np
@@ -148,11 +149,14 @@ def test_error_if_no_stream_res_set_when_requesting_img_coords():
 
 
 @pytest.fixture
-def rom_thing(example_rom_data, csm_matrix) -> stage_measure.RangeofMotionThing:
+def rom_thing(example_rom_data, csm_matrix, mocker) -> stage_measure.RangeofMotionThing:
     """Yield a RangeofMotionThing already populated with some example rom_data."""
     rom_thing = create_thing_without_server(
         stage_measure.RangeofMotionThing, mock_all_slots=True
     )
+    type(rom_thing._thing_server_interface).application_config = mocker.PropertyMock(
+        return_value={"data_folder": tempfile.gettempdir()}
+    )
     rom_thing._stream_resolution = [800, 600]
     rom_thing._rom_data = example_rom_data
 
@@ -586,7 +590,7 @@ def test_perform_rom_actions_locked(rom_thing):
         rom_thing.perform_recentre()
 
 
-def test_perform_rom_test_(rom_thing, mocker):
+def test_perform_rom_test(rom_thing, mocker):
     """Check that perform Rom Test runs the expected high level algorithm."""
 
     def check_and_modify_rom_data(axis: str, direction: int, **_kwargs):
@@ -605,7 +609,10 @@ def test_perform_rom_test_(rom_thing, mocker):
     mocker.patch.object(
         rom_thing._cam, "grab_as_array", return_value=np.zeros([123, 456, 3])
     )
-    final_dict = rom_thing.perform_rom_test()
+
+    # Enter the ROM thing to get the data directory.
+    with rom_thing as running_rom_thing:
+        final_dict = running_rom_thing.perform_rom_test()
 
     # This should take less than 1 sec
     assert final_dict["Time"] <= 1