From 0a8272685dcffa5d875b6deca2af990bf39fccd3 Mon Sep 17 00:00:00 2001 From: appel_c Date: Sun, 30 Nov 2025 22:28:34 +0100 Subject: [PATCH] fix(status): cleanup and remove of old status usage --- .../ionization_chambers/ionization_chamber.py | 3 +-- debye_bec/devices/mo1_bragg/mo1_bragg.py | 7 +++---- debye_bec/devices/nidaq/nidaq.py | 5 ++--- debye_bec/devices/pilatus/pilatus.py | 19 +++++-------------- debye_bec/devices/pilatus_curtain.py | 6 ++---- tests/tests_devices/test_pilatus.py | 2 -- 6 files changed, 13 insertions(+), 29 deletions(-) diff --git a/debye_bec/devices/ionization_chambers/ionization_chamber.py b/debye_bec/devices/ionization_chambers/ionization_chamber.py index a829560..52a6a78 100644 --- a/debye_bec/devices/ionization_chambers/ionization_chamber.py +++ b/debye_bec/devices/ionization_chambers/ionization_chamber.py @@ -9,8 +9,7 @@ from ophyd import Component as Cpt from ophyd import Device from ophyd import DynamicDeviceComponent as Dcpt from ophyd import EpicsSignal, EpicsSignalRO, EpicsSignalWithRBV -from ophyd.status import DeviceStatus, SubscriptionStatus -from ophyd_devices import CompareStatus, TransitionStatus +from ophyd_devices import CompareStatus, DeviceStatus, SubscriptionStatus, TransitionStatus from ophyd_devices.interfaces.base_classes.psi_device_base import PSIDeviceBase from typeguard import typechecked diff --git a/debye_bec/devices/mo1_bragg/mo1_bragg.py b/debye_bec/devices/mo1_bragg/mo1_bragg.py index f1b22d5..5189816 100644 --- a/debye_bec/devices/mo1_bragg/mo1_bragg.py +++ b/debye_bec/devices/mo1_bragg/mo1_bragg.py @@ -9,16 +9,15 @@ used to ensure that the action is executed completely. This is believed to allow for a more stable execution of the action.""" import time -from typing import Any, Literal +from typing import Literal from bec_lib.devicemanager import ScanInfo from bec_lib.logger import bec_logger from ophyd import Component as Cpt -from ophyd import DeviceStatus, Signal, StatusBase -from ophyd.status import SubscriptionStatus, WaitTimeoutError +from ophyd import DeviceStatus, StatusBase +from ophyd.status import WaitTimeoutError from ophyd_devices import CompareStatus, ProgressSignal, TransitionStatus from ophyd_devices.interfaces.base_classes.psi_device_base import PSIDeviceBase -from ophyd_devices.utils.errors import DeviceStopError from pydantic import BaseModel, Field from typeguard import typechecked diff --git a/debye_bec/devices/nidaq/nidaq.py b/debye_bec/devices/nidaq/nidaq.py index 3b7bb93..8d6dcd9 100644 --- a/debye_bec/devices/nidaq/nidaq.py +++ b/debye_bec/devices/nidaq/nidaq.py @@ -1,12 +1,11 @@ from __future__ import annotations -import time -from typing import TYPE_CHECKING, Literal, cast +from typing import TYPE_CHECKING, Literal from bec_lib.logger import bec_logger from ophyd import Component as Cpt from ophyd import Device, DeviceStatus, EpicsSignal, EpicsSignalRO, Kind, StatusBase -from ophyd.status import SubscriptionStatus, WaitTimeoutError +from ophyd.status import WaitTimeoutError from ophyd_devices import CompareStatus, ProgressSignal, TransitionStatus from ophyd_devices.interfaces.base_classes.psi_device_base import PSIDeviceBase from ophyd_devices.sim.sim_signals import SetableSignal diff --git a/debye_bec/devices/pilatus/pilatus.py b/debye_bec/devices/pilatus/pilatus.py index d49a692..7dabb0d 100644 --- a/debye_bec/devices/pilatus/pilatus.py +++ b/debye_bec/devices/pilatus/pilatus.py @@ -17,12 +17,10 @@ from ophyd.areadetector.cam import ADBase, PilatusDetectorCam from ophyd.areadetector.plugins import HDF5Plugin_V22 as HDF5Plugin from ophyd.areadetector.plugins import ImagePlugin_V22 as ImagePlugin from ophyd.status import WaitTimeoutError -from ophyd_devices import CompareStatus, DeviceStatus, FileEventSignal, PreviewSignal +from ophyd_devices import AndStatus, CompareStatus, DeviceStatus, FileEventSignal, PreviewSignal from ophyd_devices.interfaces.base_classes.psi_device_base import PSIDeviceBase from pydantic import BaseModel, Field -from debye_bec.devices.pilatus.utils import AndStatusWithList - if TYPE_CHECKING: # pragma: no cover from bec_lib.devicemanager import ScanInfo from bec_lib.messages import DevicePreviewMessage, ScanStatusMessage @@ -360,14 +358,12 @@ class Pilatus(PSIDeviceBase, ADBase): # f"Live Mode on detector {self.name} did not stop: {content} after 10s." # ) - def check_detector_stop_running_acquisition(self) -> AndStatusWithList: + def check_detector_stop_running_acquisition(self) -> AndStatus: """Check if the detector is still running an acquisition.""" status_acquire = CompareStatus(self.cam.acquire, ACQUIREMODE.DONE.value) status_writing = CompareStatus(self.hdf.capture, ACQUIREMODE.DONE.value) status_cam_server = CompareStatus(self.cam.armed, DETECTORSTATE.UNARMED.value) - status = AndStatusWithList( - device=self, status_list=[status_acquire, status_writing, status_cam_server] - ) + status = status_acquire & status_writing & status_cam_server return status def _calculate_trigger(self, scan_msg: ScanStatusMessage): @@ -562,9 +558,7 @@ class Pilatus(PSIDeviceBase, ADBase): status_hdf = CompareStatus(self.hdf.capture, ACQUIREMODE.ACQUIRING.value) status_cam = CompareStatus(self.cam.acquire, ACQUIREMODE.ACQUIRING.value) status_cam_server = CompareStatus(self.cam.armed, DETECTORSTATE.ARMED.value) - status = AndStatusWithList( - device=self, status_list=[status_hdf, status_cam, status_cam_server] - ) + status = status_hdf & status_cam & status_cam_server self.cam.acquire.put(1) self.hdf.capture.put(1) self.cancel_on_stop(status) @@ -627,10 +621,7 @@ class Pilatus(PSIDeviceBase, ADBase): else: status_img_written = CompareStatus(self.hdf.num_captured, self.n_images) status_img_written = CompareStatus(self.hdf.num_captured, self.n_images) - status = AndStatusWithList( - device=self, - status_list=[status_hdf, status_cam, status_img_written, status_cam_server], - ) + status = status_hdf & status_cam & status_img_written & status_cam_server status.add_callback(self._complete_callback) # Callback that writing was successful self.cancel_on_stop(status) return status diff --git a/debye_bec/devices/pilatus_curtain.py b/debye_bec/devices/pilatus_curtain.py index c7f3f05..d129673 100644 --- a/debye_bec/devices/pilatus_curtain.py +++ b/debye_bec/devices/pilatus_curtain.py @@ -10,8 +10,6 @@ from ophyd import EpicsSignal, EpicsSignalRO from ophyd_devices import CompareStatus, DeviceStatus from ophyd_devices.interfaces.base_classes.psi_device_base import PSIDeviceBase -from debye_bec.devices.pilatus.utils import AndStatusWithList - if TYPE_CHECKING: from bec_lib.devicemanager import ScanInfo @@ -84,7 +82,7 @@ class PilatusCurtain(PSIDeviceBase): # TODO timeout ok? status_open = CompareStatus(self.cover_is_open, COVER.OPEN, timeout=5) status_error = CompareStatus(self.cover_error, COVER.ERROR, operation_success="!=") - status = AndStatusWithList(device=self, status_list=[status_open, status_error]) + status = status_open & status_error return status else: return None @@ -96,7 +94,7 @@ class PilatusCurtain(PSIDeviceBase): # TODO timeout ok? status_close = CompareStatus(self.cover_is_closed, COVER.CLOSED, timeout=5) status_error = CompareStatus(self.cover_error, COVER.ERROR, operation_success="!=") - status = AndStatusWithList(device=self, status_list=[status_close, status_error]) + status = status_close & status_error return status else: return None diff --git a/tests/tests_devices/test_pilatus.py b/tests/tests_devices/test_pilatus.py index 1ad3347..46228e3 100644 --- a/tests/tests_devices/test_pilatus.py +++ b/tests/tests_devices/test_pilatus.py @@ -1,10 +1,8 @@ # pylint: skip-file -import os import threading from typing import TYPE_CHECKING, Generator from unittest import mock -import numpy as np import ophyd import pytest from bec_lib.messages import ScanStatusMessage