refactor: Fix thread leak in SimPositioner, improve callback in SimMonitor

Author: cappel89

ophyd_devices/sim/sim_data.py

@@ -108,7 +108,7 @@ def __init__(self, *args, parent=None, **kwargs) -> None:
         self.parent = parent
         self.sim_state = defaultdict(dict)
         self.registered_proxies = getattr(self.parent, "registered_proxies", {})
-        self._model = {}
+        self._model = None
         self._model_params = None
         self._params = {}
 
@@ -329,6 +329,7 @@ def __init__(self, *args, parent=None, **kwargs) -> None:
         self._model_lookup = self.init_lmfit_models()
         super().__init__(*args, parent=parent, **kwargs)
         self.bit_depth = self.parent.BIT_DEPTH
+        self._cb_registered = False
         self._init_default()
 
     @SimulatedDataBase.params.setter
@@ -341,9 +342,22 @@ def _add_callback_to_motor(self) -> None:
         mot_name = self.params.get("ref_motor", "")
         if not hasattr(self.parent, "device_manager"):
             return
+        if not hasattr(self.parent.device_manager, "devices"):
+            return
         if mot_name in self.parent.device_manager.devices:
             if hasattr(self.parent, "setup_readback_monitor"):
                 self.parent.setup_readback_monitor(mot_name)
+            # pylint: disable=protected-access
+            if (
+                hasattr(self.parent, "_registered_callback")
+                and self.parent._registered_callback is not None
+                and self.parent._registered_callback.motor == mot_name
+            ):
+                # If the callback was registered, keep track of it
+                self._cb_registered = True
+            else:
+                # If no callback was registered, this should also be reflected in self._cb_registered
+                self._cb_registered = False
 
     def select_model(self, model: str) -> None:
         """
@@ -396,6 +410,8 @@ def get_params_for_model_cls(self) -> dict:
             dict: {name: value} for the active simulation model.
         """
         rtr = {}
+        if not isinstance(self._model, Model):
+            return rtr
         params = self._model.make_params()
         for name, parameter in params.items():
             if name in DEFAULT_PARAMS_LMFIT:
@@ -465,6 +481,10 @@ def _compute(self, *args, **kwargs) -> int:
         mot_name = self.params.get("ref_motor", "")
         if self.parent.device_manager and mot_name in self.parent.device_manager.devices:
             motor_pos = self.parent.device_manager.devices[mot_name].obj.read()[mot_name]["value"]
+            # It can happen that the SimMonitor was created before the motor was available in the device manager,
+            # therefore we have to check if a callback is registered and update it if not.
+            if not self._cb_registered:
+                self._add_callback_to_motor()
         else:
             motor_pos = 0
         method = self._model

ophyd_devices/sim/sim_positioner.py

@@ -85,6 +85,7 @@ def __init__(
         self.precision = precision
         self.sim_init = sim_init
         self._registered_proxies = {}
+        self._lock = threading.RLock()
 
         self.update_frequency = update_frequency
         self._stopped = False
@@ -186,21 +187,25 @@ def _move_to_setpoint(self) -> None:
                     raise DeviceStopError(f"{self.name} was stopped")
                 ttime.sleep(1 / self.update_frequency)
             self._update_state(self.readback.get())
-            for status in self._status_list:
-                status.set_finished()
         # pylint: disable=broad-except
         except Exception as exc:
             content = traceback.format_exc()
             logger.warning(
-                f"Error in on_complete call in device {self.name}. Error traceback: {content}"
+                f"Error in _move_to_setpoint call in device {self.name}. Error traceback: {content}"
             )
-            for status in self._status_list:
-                status.set_exception(exc=exc)
+            with self._lock:
+                for status in self._status_list:
+                    status.set_exception(exc=exc)
+                self._status_list = []
         finally:
-            self.motor_is_moving.put(0)
-            if not self._stopped:
-                self._update_state(self.readback.get())
-            self._status_list = []
+            with self._lock:
+                self.motor_is_moving.put(0)
+                if not self._stopped:
+                    self._update_state(self.readback.get())
+                for status in self._status_list:
+                    if not status.done:
+                        status.set_finished()
+                self._status_list = []
 
     def move(self, value: float, **kwargs) -> DeviceStatus:
         """Change the setpoint of the simulated device, and simultaneously initiate a motion."""
@@ -210,7 +215,8 @@ def move(self, value: float, **kwargs) -> DeviceStatus:
         self.setpoint.put(value)
 
         st = DeviceStatus(device=self)
-        self._status_list.append(st)
+        with self._lock:
+            self._status_list.append(st)
         if self.delay:
             if self.move_thread is None or not self.move_thread.is_alive():
                 self.move_thread = threading.Thread(target=self._move_to_setpoint)

ophyd_devices/sim/sim_signals.py

@@ -80,7 +80,8 @@ def get(self, **kwargs) -> Any:
         """
         old_value = self._readback
         self._readback = self._value = self._get_value()
-        self._run_subs(sub_type=self.SUB_VALUE, old_value=old_value, value=self._readback)
+        if old_value != self._readback:  # only run subs if the value has changed
+            self._run_subs(sub_type=self.SUB_VALUE, old_value=old_value, value=self._readback)
         return self._readback
 
     # pylint: disable=arguments-differ
@@ -89,9 +90,11 @@ def put(self, value) -> None:
 
         Core function for signal.
         """
+        old_value = self._value
         self.check_value(value)
         self._update_sim_state(value)
         self._value = value
+        self._run_subs(sub_type=self.SUB_VALUE, old_value=old_value, value=self._value)
         super().put(value)
 
     def set(self, value):

ophyd_devices/sim/sim_waveform.py

@@ -129,9 +129,9 @@ def __init__(
         self._trigger_received = 0
         self.scan_info = scan_info
         self._delay_slice_update = False
+        self._slice_index = 0
         if self.sim_init:
             self.sim.set_init(self.sim_init)
-        self._slice_index = 0
 
     @property
     def delay_slice_update(self) -> bool:

ophyd_devices/utils/static_device_test.py

@@ -286,7 +286,7 @@ def run(
             if device_manager is not None:  # Only possible if bec-server is installed
                 obj = self.construct_device_obj(name, conf)
                 if obj is None:  # construction failed, skip connection test
-                    return_value += 1
+                    return_val += 1
                 elif obj is not None and connect:
                     return_val += self.connect_device(
                         name,

tests/test_simulation.py

@@ -48,6 +48,7 @@ def waveform(name="waveform"):
     """Fixture for SimWaveform."""
     dm = DMMock()
     wave = SimWaveform(name=name, device_manager=dm)
+    wave.wait_for_connection()
     yield wave
 
 
@@ -59,10 +60,21 @@ def signal(name="signal"):
 
 
 @pytest.fixture(scope="function")
-def monitor(name="monitor"):
-    """Fixture for SimMonitor."""
+def samx(name="samx"):
+    """Fixture for SimPositioner."""
     dm = DMMock()
+    pos = SimPositioner(name=name, device_manager=dm)
+    yield pos
+
+
+@pytest.fixture(scope="function")
+def monitor(samx, name="monitor"):
+    """Fixture for SimMonitor."""
+    dm = samx.device_manager
+    samx.obj = samx  # Set obj attribute to itself for proxy lookup
+    dm.devices["samx"] = samx
     mon = SimMonitor(name=name, device_manager=dm)
+    mon.wait_for_connection()
     yield mon
 
 
@@ -97,6 +109,7 @@ def async_monitor(name="async_monitor"):
     """Fixture for SimMonitorAsync."""
     dm = DMMock()
     mon = SimMonitorAsync(name=name, device_manager=dm)
+    mon.wait_for_connection()
     yield mon
 
 
@@ -168,6 +181,7 @@ def test_monitor_with_sim_init():
     """Test to see if the sim init parameters are passed to the device"""
     dm = DMMock()
     sim = SimMonitor(name="sim", device_manager=dm)
+    sim.wait_for_connection()
     assert sim.sim._model._name == "constant"
     model = "GaussianModel"
     params = {
@@ -179,6 +193,7 @@ def test_monitor_with_sim_init():
         "ref_motor": "samy",
     }
     sim = SimMonitor(name="sim", device_manager=dm, sim_init={"model": model, "params": params})
+    sim.wait_for_connection()
     assert sim.sim._model._name == model.strip("Model").lower()
     diff_keys = set(sim.sim.params.keys()) - set(params.keys())
     for k in params:
@@ -224,9 +239,7 @@ def test_init_async_monitor(async_monitor):
 def test_monitor_readback(monitor, center, positioner):
     """Test the readback method of SimMonitor."""
     motor_pos = 0
-    samx = SimPositioner(name="samx", device_manager=monitor.device_manager)
-    setattr(samx, "obj", samx)  # Set obj attribute to itself for proxy lookup
-    monitor.device_manager.devices["samx"] = samx
+    samx = monitor.device_manager.devices.get("samx", None)
     for model_name in monitor.sim.get_models():
         monitor.sim.select_model(model_name)
         monitor.sim.params["noise_multipler"] = 10