SPMMicroscope class and SPM_Scan sub-device

asyncroscopy/instruments/instrument.py

@@ -5,6 +5,7 @@
 from abc import abstractmethod, ABCMeta
 
 import tango
+import tango.server
 
 class CombinedMeta(tango.server.DeviceMeta, ABCMeta):
     """Combines Tango DeviceMeta and ABCMeta to allow abstract methods in Devices."""

asyncroscopy/instruments/scanning_probe_microscope/hardware/spm_approach.py

@@ -0,0 +1,114 @@
+"""
+SPM APPROACH Tango device.
+
+Owns the tip approach/retract sequence. Vendor-specific behaviour is
+isolated in the ``_hw_*`` hooks implemented by concrete subclasses
+(APPROACH_Jupiter, etc.).
+"""
+
+from abc import abstractmethod
+
+import tango
+from asyncroscopy.instruments.instrument import CombinedMeta
+
+class SPM_APPROACH(tango.server.Device, metaclass=CombinedMeta):
+    """Abstract SPM approach device: tip engage / retract."""
+
+    # ------------------------------------------------------------------
+    # Attributes
+    # ------------------------------------------------------------------
+
+    approached = tango.server.attribute(
+        label="Approached",
+        dtype=bool,
+        access=tango.AttrWriteType.READ,
+        doc="True when the tip is engaged on the surface. Read live from hardware.",
+    )
+
+
+    # ------------------------------------------------------------------
+    # Initialization
+    # ------------------------------------------------------------------
+
+    def init_device(self) -> None:
+        tango.server.Device.init_device(self)
+        self.set_state(tango.DevState.ON)
+        self.info_stream("SPM APPROACH device initialised")
+
+    # ------------------------------------------------------------------
+    # Attribute read
+    # ------------------------------------------------------------------
+
+    def read_approached(self) -> bool:
+        return self._hw_is_approached()
+
+        # ------------------------------------------------------------------
+    # Commands
+    # ------------------------------------------------------------------
+
+    @tango.server.command
+    def approach(self) -> None:
+        """Engage the tip on the surface. Blocks until engaged."""
+        if self.get_state() == tango.DevState.MOVING:
+            tango.Except.throw_exception(
+                "ApproachInProgress",
+                "An approach or retract is already running; call stop first.",
+                "approach()",
+            )
+        self.set_state(tango.DevState.MOVING)
+        try:
+            self._hw_approach()
+        finally:
+            self.set_state(tango.DevState.ON)
+
+    @tango.server.command
+    def retract(self) -> None:
+        """Retract the tip from the surface. Blocks until retracted."""
+        if self.get_state() == tango.DevState.MOVING:
+            tango.Except.throw_exception(
+                "ApproachInProgress",
+                "An approach or retract is already running; call stop first.",
+                "retract()",
+            )
+        self.set_state(tango.DevState.MOVING)
+        try:
+            self._hw_retract()
+        finally:
+            self.set_state(tango.DevState.ON)
+
+    @tango.server.command
+    def stop(self) -> None:
+        """Abort a running approach or retract immediately."""
+        self._hw_stop()
+        self.set_state(tango.DevState.ON)
+
+    # ------------------------------------------------------------------
+    # Abstract methods — vendor-specific
+    # ------------------------------------------------------------------
+
+    @abstractmethod
+    def _hw_approach(self) -> None:
+        """Run the approach sequence; return when the tip is engaged."""
+        pass
+
+    @abstractmethod
+    def _hw_retract(self) -> None:
+        """Retract the tip; return when clear of the surface."""
+        pass
+
+    @abstractmethod
+    def _hw_stop(self) -> None:
+        """Abort any running approach/retract motion."""
+        pass
+
+    @abstractmethod
+    def _hw_is_approached(self) -> bool:
+        """Return True if the tip is currently engaged."""
+        pass
+
+# ----------------------------------------------------------------------
+# Server entry point
+# ----------------------------------------------------------------------
+if __name__ == "__main__":
+    SPM_APPROACH.run_server()
+

asyncroscopy/instruments/scanning_probe_microscope/hardware/spm_feedback.py

@@ -0,0 +1,140 @@
+"""
+SPM FEEDBACK Tango device.
+
+Owns the Z feedback loop: setpoint, gains, and engage/withdraw.
+
+Note: the physical meaning of ``setpoint`` depends on the active SPM
+mode (deflection volts in contact, amplitude volts in AC, etc.); this
+device passes the value through without interpreting it.
+"""
+
+from abc import abstractmethod
+import tango # type: ignore
+
+from asyncroscopy.instruments.instrument import CombinedMeta
+
+class SPM_FEEDBACK(tango.server.Device, metaclass=CombinedMeta):
+    """Abstract SPM feedback device: Z loop setpoint, gains, engage state."""
+
+    # ------------------------------------------------------------------
+    # Attributes
+    # ------------------------------------------------------------------
+
+    setpoint = tango.server.attribute(
+        label="Setpoint",
+        dtype=float,
+        access=tango.AttrWriteType.READ_WRITE,
+        doc="Z feedback setpoint. Physical meaning depends on the active SPM mode "
+            "(deflection in contact, amplitude in AC).",
+    )
+
+    i_gain = tango.server.attribute(
+        label="Integral Gain",
+        dtype=float,
+        access=tango.AttrWriteType.READ_WRITE,
+        doc="Integral gain of the Z feedback loop.",
+    )
+
+    feedback_on_bool = tango.server.attribute(
+        label="Feedback On",
+        dtype=bool,
+        access=tango.AttrWriteType.READ,
+        doc="True when the Z feedback loop is engaged. Read live from hardware.",
+    )
+
+    # ------------------------------------------------------------------
+    # Initialization
+    # ------------------------------------------------------------------
+    def init_device(self) -> None:
+        tango.server.Device.init_device(self)
+        self._refresh_params()
+        self.set_state(tango.DevState.ON)
+        self.info_stream("SPM FEEDBACK device initialised")
+
+    # ------------------------------------------------------------------
+    # Internal helpers
+    # ------------------------------------------------------------------
+
+    def _refresh_params(self) -> None:
+        """Re-read all feedback parameters from hardware into local fields."""
+        params = self._hw_read_feedback_params()
+        self._setpoint: float = params["setpoint"]
+        self._i_gain: float = params["i_gain"]
+
+    def _write_param(self, name: str, value) -> None:
+        """Push one parameter to hardware, then re-read all (hardware may coerce values)."""
+        self._hw_write_feedback_param(name, value)
+        self._refresh_params()
+
+    # ------------------------------------------------------------------
+    # Attribute read / write — writes pushed to hardware
+    # ------------------------------------------------------------------
+
+    def read_setpoint(self) -> float:
+        return self._setpoint
+
+    def write_setpoint(self, value: float) -> None:
+        self._write_param("setpoint", value)
+
+    def read_i_gain(self) -> float:
+        return self._i_gain
+
+    def write_i_gain(self, value: float) -> None:
+        self._write_param("i_gain", value)
+
+    def read_feedback_on_bool(self) -> bool:
+        return self._hw_is_feedback_on()
+
+    # ------------------------------------------------------------------
+    # Commands
+    # ------------------------------------------------------------------
+
+    @tango.server.command
+    def feedback_on(self) -> None:
+        """Engage the Z feedback loop."""
+        self._hw_feedback_on()
+
+    @tango.server.command
+    def feedback_off(self) -> None:
+        """Disengage the Z feedback loop."""
+        self._hw_feedback_off()
+
+    @tango.server.command
+    def refresh_params(self) -> None:
+        """Re-read all feedback parameters from hardware, e.g. after changes in the vendor GUI."""
+        self._refresh_params()
+
+    # ------------------------------------------------------------------
+    # Abstract methods — vendor-specific
+    # ------------------------------------------------------------------
+
+    @abstractmethod
+    def _hw_read_feedback_params(self) -> dict:
+        """Read all feedback parameters from hardware. Keys must match attribute names."""
+        pass
+
+    @abstractmethod
+    def _hw_write_feedback_param(self, name: str, value) -> None:
+        """Push one feedback parameter to hardware."""
+        pass
+
+    @abstractmethod
+    def _hw_feedback_on(self) -> None:
+        """Engage the Z feedback loop."""
+        pass
+
+    @abstractmethod
+    def _hw_feedback_off(self) -> None:
+        """Disengage the Z feedback loop."""
+        pass
+
+    @abstractmethod
+    def _hw_is_feedback_on(self) -> bool:
+        """Return True if the Z feedback loop is currently engaged."""
+        pass
+
+# ----------------------------------------------------------------------
+# Server entry point
+# ----------------------------------------------------------------------
+if __name__ == "__main__":
+    SPM_FEEDBACK.run_server()