Merge remote-tracking branch 'origin/v3' into v3-webapp

src/openflexure_microscope_server/server.py

@@ -10,6 +10,7 @@ from labthings_picamera2.thing import StreamingPiCamera2
 
 from .things.autofocus import AutofocusThing
 from .things.camera_stage_mapping import CameraStageMapper
+from .things.system_control import SystemControlThing
 from .serve_static_files import add_static_files
 import openflexure_microscope_server
 
@@ -20,6 +21,7 @@ thing_server.add_thing(StreamingPiCamera2(), "/camera/")
 thing_server.add_thing(SangaboardThing(), "/stage/")
 thing_server.add_thing(AutofocusThing(), "/autofocus/")
 thing_server.add_thing(CameraStageMapper(), "/camera_stage_mapping/")
+thing_server.add_thing(SystemControlThing(), "/system_control/")
 try:
     add_static_files(thing_server.app)
 except RuntimeError:

src/openflexure_microscope_server/things/camera_stage_mapping.py

@@ -132,11 +132,15 @@ class CameraStageMapper(Thing):
 
         result: dict = calibrate_backlash_1d(tracker, move, direction_array)
         result["move_history"] = move.history
+        result["image_resolution"] = hw.grab_image().shape[:2]
         return result
 
     @thing_action
     def calibrate_xy(self, hw: HardwareInterfaceDep) -> DenumpifyingDict:
-        """Move the microscope's stage in X and Y, to calibrate its relationship to the camera"""
+        """Move the microscope's stage in X and Y, to calibrate its relationship to the camera
+        
+        This performs two 1d calibrations in x and y, then combines their results.
+        """
         logging.info("Calibrating X axis:")
         cal_x: dict = self.calibrate_1d(hw, (1, 0, 0))
         logging.info("Calibrating Y axis:")
@@ -145,11 +149,13 @@ class CameraStageMapper(Thing):
         # Combine X and Y calibrations to make a 2D calibration
         cal_xy: dict = denumpify(image_to_stage_displacement_from_1d([cal_x, cal_y]))
         self.thing_settings.update(cal_xy)
+        self.thing_settings["image_resolution"] = cal_x["image_resolution"]
 
         data: Dict[str, dict] = {
             "camera_stage_mapping_calibration": cal_xy,
             "linear_calibration_x": cal_x,
             "linear_calibration_y": cal_y,
+            "image_resolution": cal_x["image_resolution"]
         }
 
         self.thing_settings["last_calibration"] = DenumpifyingDict(data).model_dump()
@@ -158,12 +164,34 @@ class CameraStageMapper(Thing):
 
     @thing_property
     def image_to_stage_displacement_matrix(self) -> List[List[float]]:  # 2x2 integer array
-        """A 2x2 matrix that converts displacement in image coordinates to stage coordinates."""
+        """A 2x2 matrix that converts displacement in image coordinates to stage coordinates.
+        
+        Note that this matrix is defined using "matrix coordinates", i.e. image coordinates
+        may be (y,x). This is an artifact of the way numpy, opencv, etc. define images. If
+        you are making use of this matrix in your own code, you will need to take care of
+        that conversion.
+        
+        It is often helpful to give a concrete example: to make a move in image coordinates
+        (`dy`, `dx`), where `dx` is horizontal, i.e. the longer dimension of the image, you
+        should move the stage by:
+        ```
+        stage_disp = np.dot(
+            np.array(image_to_stage_displacement_matrix),
+            np.array([dy,dx]),
+        )
+        ```
+        """
         displacement_matrix = self.thing_settings.get("image_to_stage_displacement")
         if not displacement_matrix:
             raise ValueError("The microscope has not yet been calibrated.")
         return np.array(displacement_matrix).tolist()
 
+    @thing_property
+    def last_calibration(self) -> Optional[Dict]:  # 2x2 integer array
+        """The results of the last calibration that was run
+        """
+        return self.thing_settings.get("last_calibration", None)
+    
     @thing_action
     def move_in_image_coordinates(
         self,
@@ -188,11 +216,14 @@ class CameraStageMapper(Thing):
         )
         stage.move_relative(x=relative_move[0], y=relative_move[1])
 
-    @property
+    @thing_property
     def thing_state(self) -> dict:
         """Summary metadata describing the current state of the Thing"""
         return {
             "image_to_stage_displacement_matrix": (
                 self.image_to_stage_displacement_matrix
             ),
+            "image_resolution": (
+                self.thing_settings.get("image_to_stage_displacement", None)
+            )
         }

src/openflexure_microscope_server/things/system_control.py

@@ -0,0 +1,59 @@
+"""
+OpenFlexure Microscope API extension for stage calibration
+
+This file contains the HTTP API for camera/stage calibration. It
+includes calibration functions that measure the relationship between
+stage coordinates and camera coordinates, as well as functions that
+move by a specified displacement in pixels, perform closed-loop moves,
+and return the calibration data.
+
+This module is only intended to be called from the OpenFlexure Microscope
+server, and depends on that server and its underlying LabThings library.
+"""
+import subprocess
+import os
+from labthings_fastapi.thing import Thing
+from labthings_fastapi.decorators import thing_action, thing_property
+from pydantic import BaseModel
+
+class CommandOutput(BaseModel):
+    output: str
+    error: str
+
+
+class SystemControlThing(Thing):
+    """
+    Attempt to shutdown the device 
+    """
+
+    @thing_action
+    def shutdown(self) -> CommandOutput:
+        """
+        Attempt to shutdown the device
+        """
+        p = subprocess.Popen(
+            ["sudo", "shutdown", "-h", "now"],
+            stderr=subprocess.PIPE,
+            stdout=subprocess.PIPE,
+        )
+
+        out, err = p.communicate()
+        return CommandOutput(output=out, error=err)
+
+    @thing_property
+    def is_raspberrypi() -> bool:
+        """
+        Checks if we are running on a Raspberry Pi.
+        """
+        return os.path.exists("/usr/bin/raspi-config")
+    
+    @thing_action
+    def reboot(self) -> CommandOutput:
+        """Attempt to reboot the device"""
+        p = subprocess.Popen(
+            ["sudo", "shutdown", "-r", "now"],
+            stderr=subprocess.PIPE,
+            stdout=subprocess.PIPE,
+        )
+        out, err = p.communicate()
+        return CommandOutput(output=out, error=err)