bec/ophyd_devices
1
0
Fork 0
Code Issues 9 Pull Requests 10 Actions Packages Projects Releases 126 Wiki Activity

refactor: make it easier to subclass SimPositioner

Browse Source
This commit is contained in:
guijar_m 2024-06-28 15:22:28 +02:00 committed by guijar_m
parent d3ef7df4e9
commit 9037553252
1 changed files with 51 additions and 51 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

58
ophyd_devices/sim/sim_positioner.py
View File

@ -76,6 +76,7 @@ class SimPositioner(Device, PositionerBase):
sim_init: dict = None, sim_init: dict = None,
**kwargs, **kwargs,
): ):
self.move_thread = None
self.delay = delay self.delay = delay
self.device_manager = device_manager self.device_manager = device_manager
self.precision = precision self.precision = precision
@ -139,15 +140,7 @@ class SimPositioner(Device, PositionerBase):
"""Return the simulated state of the device.""" """Return the simulated state of the device."""
return self.sim.sim_state[signal_name]["value"] return self.sim.sim_state[signal_name]["value"]
def move(self, value: float, **kwargs) -> DeviceStatus: def _update_state(self, val):
"""Change the setpoint of the simulated device, and simultaneously initiated a motion."""
self._stopped = False
self.check_value(value)
old_setpoint = self._get_sim_state(self.setpoint.name)
self._set_sim_state(self.motor_is_moving.name, 1)
self._set_sim_state(self.setpoint.name, value)
def update_state(val):
"""Update the state of the simulated device.""" """Update the state of the simulated device."""
if self._stopped: if self._stopped:
raise DeviceStop raise DeviceStop
@ -162,42 +155,47 @@ class SimPositioner(Device, PositionerBase):
timestamp=self.sim.sim_state[self.readback.name]["timestamp"], timestamp=self.sim.sim_state[self.readback.name]["timestamp"],
) )
st = DeviceStatus(device=self) def _move_and_finish(self, start_pos, stop_pos, st):
if self.delay:
def move_and_finish():
"""Move the simulated device and finish the motion.""" """Move the simulated device and finish the motion."""
success = True success = True
try: try:
move_val = self._get_sim_state( target = stop_pos + self.tolerance.get() * np.random.uniform(-1, 1)
self.setpoint.name
) + self.tolerance.get() * np.random.uniform(-1, 1)
updates = np.ceil( updates = np.ceil(
np.abs(old_setpoint - move_val) np.abs(target - start_pos) / self.velocity.get() * self.update_frequency
/ self.velocity.get()
* self.update_frequency
) )
for ii in np.linspace(old_setpoint, move_val, int(updates)): for ii in np.linspace(start_pos, target, int(updates)):
ttime.sleep(1 / self.update_frequency) ttime.sleep(1 / self.update_frequency)
update_state(ii) self._update_state(ii)
if self._stopped:
self._set_sim_state(self.motor_is_moving.name, 0) break
update_state(move_val) else:
self._update_state(target)
except DeviceStop: except DeviceStop:
success = False success = False
finally: finally:
self._stopped = False
self._done_moving(success=success) self._done_moving(success=success)
self._set_sim_state(self.motor_is_moving.name, 0) self._set_sim_state(self.motor_is_moving.name, 0)
st.set_finished() st.set_finished()
threading.Thread(target=move_and_finish, daemon=True).start() def move(self, value: float, **kwargs) -> DeviceStatus:
"""Change the setpoint of the simulated device, and simultaneously initiate a motion."""
self._stopped = False
self.check_value(value)
start_pos = self._get_sim_state(self.setpoint.name)
self._set_sim_state(self.motor_is_moving.name, 1)
self._set_sim_state(self.setpoint.name, value)
st = DeviceStatus(device=self)
if self.delay:
self.move_thread = threading.Thread(
target=self._move_and_finish, args=(start_pos, value, st)
)
self.move_thread.start()
else: else:
update_state(value) self._update_state(value)
self._done_moving() self._done_moving()
self._set_sim_state(self.motor_is_moving.name, 0) self._set_sim_state(self.motor_is_moving.name, 0)
st.set_finished() st.set_finished()
@ -205,8 +203,10 @@ class SimPositioner(Device, PositionerBase):
def stop(self, *, success=False): def stop(self, *, success=False):
"""Stop the motion of the simulated device.""" """Stop the motion of the simulated device."""
super().stop(success=success)
self._stopped = True self._stopped = True
if self.move_thread:
self.move_thread.join()
super().stop(success=success)
@property @property
def position(self) -> float: def position(self) -> float: