bec/ophyd_devices
0
0
Fork 0
mirror of https://github.com/bec-project/ophyd_devices.git synced 2026-02-20 09:18:42 +01:00
Code Issues Packages Projects Releases Wiki Activity

wip improve status and tests

Browse Source
This commit is contained in:
appel_c
2025-09-10 10:50:20 +02:00
parent 94bc11e4d8
commit 6d0a26cc40
2 changed files with 7 additions and 3 deletions
Download Patch File Download Diff File Expand all files Collapse all files

5
ophyd_devices/interfaces/base_classes/psi_device_base.py
View File

@@ -199,8 +199,9 @@ class PSIDeviceBase(Device):
to be stopped when the device is stopped.
"""
for status in self._stoppable_status_objects:
if not status.done:
status.set_exception(DeviceStoppedError(f"Device {self.name} has been stopped"))
with status._lock:
if not status.done:
status.set_exception(DeviceStoppedError(f"Device {self.name} has been stopped"))
self._clear_stoppable_status_objects()
########################################

5
ophyd_devices/tests/utils.py
View File

@@ -30,6 +30,8 @@ def patch_dual_pvs(device):
continue
if walk.item._read_pv.pvname != walk.item._write_pv.pvname:
walk.item._read_pv = walk.item._write_pv
if walk.item.pvname != walk.item.setpoint_pvname:
walk.item._setpoint_pvname = walk.item.pvname
def patch_functions_required_for_connection(device):
@@ -288,7 +290,8 @@ class MockPV:
as_namespace=False,
):
"""Get MOCKPV data together with metadata"""
if as_string:
return {"value": str(self.mock_data)}
return {"value": self.mock_data}
def get(