use Autolock for manual lock as well

linien-common/linien_common/enums.py

@@ -33,6 +33,7 @@ class AutolockMode(IntEnum):
     AUTO_DETECT = 0
     ROBUST = 1
     SIMPLE = 2
+    MANUAL = 3
 
 
 class AutolockStatus(IntEnum):

linien-server/linien_server/autolock/autolock.py

@@ -61,20 +61,37 @@ def relock(self):
         self.parameters.autock_status.value = AutolockStatus.RELOCKING
         self.control.exposed_start_sweep()
         # Add a listener that listens for new spectrum data and tries to relock.
-        self.parameters.to_plot.add_callback(self.try_to_start_autolock)
+        if self.parameters.autolock_mode_preference == AutolockMode.MANUAL:
+            self.start_manual_lock()
+        else:
+            self.parameters.to_plot.add_callback(self.try_to_start_autolock)
 
     def run(
         self,
-        x0: float,
-        x1: float,
-        spectrum: np.ndarray,  # array of int
+        x0: float = 0,
+        x1: float = 0,
+        spectrum: np.ndarray = np.array([0]),  # array of int
         additional_spectra: Optional[list[np.ndarray]] = None,
     ) -> None:
+        """
+        Start the autolock to a lock point between `x0` and `x1` of a `spectrum`.
+
+        An autolock algorithm will be used depending on the `autolock_mode` parameter
+        If set `AutolockMode.AUTO_DETECT` an appropriate algorithm will be chosen based
+        on the data. If `AutolockMode.MANUAL` is set, `x0`, `x1` and `spectrum` do not
+        need to be provided. Instead, the lock will be engaged at the
+        `autolock_target_position` parameter and PID control is applied depending on the
+        `target_slope_rising` parameter.
+        """
         self.parameters.autolock_status.value = AutolockStatus.LOCKING
         self.parameters.fetch_additional_signals.value = False
         self.additional_spectra = additional_spectra or []
         self.spectrum = spectrum
 
+        if self.parameters.autolock_mode_preference == AutolockMode.MANUAL:
+            self.start_manual_lock()
+            return
+
         (
             mean_signal,
             target_slope_rising,
@@ -175,3 +192,14 @@ def try_to_start_autolock(self, plot_data: bytes) -> None:
         except Exception:
             logger.exception("Error while handling new spectrum")
             self.stop()
+
+    def start_manual_lock(self) -> None:
+        # manual lock is just the simple algorithm with a lock point determined by the
+        # `autolock_target_position` paramter
+        self.parameters.autolock_mode.value = AutolockMode.SIMPLE
+        self.control.exposed_write_registers()
+        self.control.exposed_pause_acquisition()
+        logger.info("Start manual lock.")
+        self.parameters.lock.value = True
+        self.exposed_write_registers()
+        self.exposed_continue_acquisition()

linien-server/linien_server/autolock/simple.py

@@ -66,7 +66,12 @@ def handle_new_spectrum(self, spectrum) -> None:
             round((shift * (-1)) * self.parameters.sweep_amplitude.value * 8191)
         )
         logger.debug(f"Target position is {target_position}, shift is {shift}.")
-        self.control.exposed_start_manual_lock(target_position)
+        self.control.exposed_write_registers()
+        self.control.exposed_pause_acquisition()
+        logger.info("Start lock.")
+        self.parameters.lock.value = True
+        self.exposed_write_registers()
+        self.exposed_continue_acquisition()
         self._done = True
 
     def after_lock(self):

linien-server/linien_server/parameters.py

@@ -563,8 +563,7 @@ def __init__(self):
         """Upper bound for monitor signal above which `lock_lost` is triggered."""
 
         # ------------------- AUTOLOCK PARAMETERS --------------------------------------
-        # These parameters are used internally by the optimization algorithm and usually
-        # should not be manipulated
+        # These parameters are used internally by the optimization algorithm.
         self.task = Parameter(start=None, sync=False)
         self.automatic_mode = Parameter(start=True)
         self.autolock_target_position = Parameter(start=0)

linien-server/linien_server/server.py

@@ -48,7 +48,6 @@
 from linien_server.registers import Registers
 from rpyc.core.protocol import Connection
 from rpyc.utils.server import ThreadedServer
-from typing_extensions import deprecated
 
 logger = logging.getLogger(__name__)
 
@@ -117,33 +116,6 @@ def exposed_set_parameter_log(self, param_name: str, value: bool) -> None:
     def exposed_get_parameter_log(self, param_name: str) -> bool:
         return getattr(self.parameters, param_name).log
 
-    @deprecated("Use exposed_start_manual_lock instead")
-    def exposed_start_lock(self):
-        self.exposed_start_manual_lock()
-
-    def exposed_start_manual_lock(
-        self, target_position: Optional[int] = None, slope_rising: Optional[bool] = None
-    ) -> None:
-        """
-        Engage the lock at the specified position of the sweep with the specified slope
-        sign. If one of these parameters is not specified, the previously defined values
-        of `autolock_target_position` and `target_slope_rising` are used.
-        """
-        self.parameters.fetch_additional_signals.value = False
-        self.parameters.autolock_mode.value = AutolockMode.SIMPLE
-        if target_position is not None:
-            logger.debug(f"Set target position to {target_position}.")
-            self.parameters.autolock_target_position.value = target_position
-        if slope_rising is not None:
-            logger.debug(f"Set target slope rising to {target_position}")
-            self.parameters.target_slope_rising.value = slope_rising
-        self.exposed_write_registers()
-        self.exposed_pause_acquisition()
-        logger.info("Start lock.")
-        self.parameters.lock.value = True
-        self.exposed_write_registers()
-        self.exposed_continue_acquisition()
-
     def exposed_update_influxdb_credentials(
         self, credentials: InfluxDBCredentials
     ) -> tuple[bool, str]:
@@ -277,25 +249,29 @@ def exposed_write_registers(self) -> None:
 
     def exposed_start_autolock(
         self,
-        x0: float,
-        x1: float,
-        spectrum: bytes,
+        x0: float = 0,
+        x1: float = 0,
+        spectrum: bytes = pickle.dumps(np.array([0])),
         additional_spectra: Optional[bytes] = None,
     ) -> None:
-        logger.info(f"Start autolock {x0=} {x1=}")
 
         if self.parameters.task.value is None:
             self.parameters.task.value = Autolock(self, self.parameters)
-            self.parameters.task.value.run(
-                x0,
-                x1,
-                pickle.loads(spectrum),
-                additional_spectra=(
-                    pickle.loads(additional_spectra)
-                    if additional_spectra is not None
-                    else None
-                ),
-            )
+            if self.parameters.autolock_mode_preference == AutolockMode.MANUAL:
+                logger.info("Start manual lock.")
+                self.parameters.task.value.run()
+            else:
+                logger.info(f"Start autolock with {x0=} {x1=}")
+                self.parameters.task.value.run(
+                    x0,
+                    x1,
+                    pickle.loads(spectrum),
+                    additional_spectra=(
+                        pickle.loads(additional_spectra)
+                        if additional_spectra is not None
+                        else None
+                    ),
+                )
 
     def exposed_start_optimization(self, x0, x1, spectrum):
         if self.parameters.task.value is None: