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

fix: Adapt to FileWriter refactoring

Browse Source
This commit is contained in:
appel_c 2024-04-05 14:46:43 +02:00
parent 32b6d476ca
commit e9c626a7c8
9 changed files with 19 additions and 29 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
ophyd_devices/epics/devices/eiger9m_csaxs.py
View File

@ -205,7 +205,7 @@ class Eiger9MSetup(CustomDetectorMixin):
def prepare_data_backend(self) -> None:
"""Prepare the data backend for the scan"""
self.parent.filepath = self.parent.filewriter.compile_full_filename(
self.parent.scaninfo.scan_number, f"{self.parent.name}.h5", 1000, 5, True
f"{self.parent.name}.h5"
)
self.filepath_exists(self.parent.filepath)
self.stop_detector_backend()

2
ophyd_devices/epics/devices/falcon_csaxs.py
View File

@ -184,7 +184,7 @@ class FalconSetup(CustomDetectorMixin):
def prepare_data_backend(self) -> None:
"""Prepare data backend for acquisition"""
self.parent.filepath = self.parent.filewriter.compile_full_filename(
self.parent.scaninfo.scan_number, f"{self.parent.name}.h5", 1000, 5, True
f"{self.parent.name}.h5"
)
file_path, file_name = os.path.split(self.parent.filepath)
self.parent.hdf5.file_path.put(file_path)

4
ophyd_devices/epics/devices/pilatus_csaxs.py
View File

@ -188,9 +188,7 @@ class PilatusSetup(CustomDetectorMixin):
self.stop_detector_backend()
self.parent.filepath = self.parent.filewriter.compile_full_filename(
self.parent.scaninfo.scan_number, "pilatus_2.h5", 1000, 5, True
)
self.parent.filepath = self.parent.filewriter.compile_full_filename("pilatus_2.h5")
self.parent.cam.file_path.put("/dev/shm/zmq/")
self.parent.cam.file_name.put(
f"{self.parent.scaninfo.username}_2_{self.parent.scaninfo.scan_number:05d}"

4
ophyd_devices/epics/devices/psi_detector_base.py
View File

@ -2,7 +2,7 @@ import os
import time
from bec_lib.device import DeviceStatus
from bec_lib.file_utils import FileWriterMixin
from bec_lib.file_utils import FileWriter
from ophyd import Device
from ophyd.device import Staged
@ -235,7 +235,7 @@ class PSIDetectorBase(Device):
def _update_filewriter(self) -> None:
"""Update filewriter with service config"""
self.filewriter = FileWriterMixin(self.service_cfg)
self.filewriter = FileWriter(service_config=self.service_cfg, connector=self.connector)
def _update_scaninfo(self) -> None:
"""Update scaninfo from BecScaninfoMixing

16
ophyd_devices/ophyd_base_devices/bec_protocols.py
View File

@ -18,7 +18,7 @@ Flyers in addition, also implement the BECFlyerProtocol. Similarly, positioners
from typing import Protocol, runtime_checkable
from bec_lib.file_utils import FileWriterMixin
from bec_lib.file_utils import FileWriter
from ophyd import Component, DeviceStatus, Kind, Staged
from ophyd_devices.utils import bec_scaninfo_mixin
@ -47,11 +47,11 @@ class BECDeviceProtocol(Protocol):
"""kind setter"""
@property
def parent(self):
def parent(self) -> object:
"""Property to find the parent device"""
@property
def root(self):
def root(self) -> object:
"""Property to fint the root device"""
@property
@ -169,7 +169,7 @@ class BECSignalProtocol(Protocol):
bool: True if write access is allowed, False otherwise
"""
def check_value(self, value: float):
def check_value(self, value: float) -> None:
"""Check whether value is within limits
Args:
@ -230,7 +230,7 @@ class BECScanProtocol(BECDeviceProtocol, Protocol):
We can further publish the file location for DAQ systems
to BEC and inform BEC's file writer where data will be written to.
Stagin is not idempoent. If called twice without an unstage it should raise.
Stagin is not idempotent. If called twice without an unstage it should raise.
For ophyd devices, one may used self._staged = True to check if the device is staged.
Returns:
@ -246,7 +246,7 @@ class BECScanProtocol(BECDeviceProtocol, Protocol):
inform BEC that the file has been succesfully written, or raise upon receiving
feedback that the scan did not finish successful.
Unstaging is not idempotent. If called twice without a stage it should raise.
Unstaging is not idempotent. If called twice it should simply resolve.
It is recommended to return super().unstage() in the child class, if
the child class also inherits from ophyd repository.
"""
@ -285,7 +285,7 @@ class BECMixinProtocol(Protocol):
The stop method should set this flag to True, and i.e. stage to set it to False.
"""
filewriter: FileWriterMixin
filewriter: FileWriter
"""
The file writer mixin main purpose is to unify and centralize the creation of
file paths within BEC. Therefore, we recommend devices to use the same mixin for creation of paths.
@ -332,7 +332,7 @@ class BECPositionerProtocol(Protocol):
float: Upper limit
"""
def check_value(self, value: float):
def check_value(self, value: float) -> None:
"""Check whether value is within limits
Args:

8
tests/test_eiger9m_csaxs.py
View File

@ -27,9 +27,7 @@ def mock_det():
sim_mode = False
dm = DMMock()
with mock.patch.object(dm, "connector"):
with mock.patch(
"ophyd_devices.epics.devices.psi_detector_base.FileWriterMixin"
), mock.patch(
with mock.patch("ophyd_devices.epics.devices.psi_detector_base.FileWriter"), mock.patch(
"ophyd_devices.epics.devices.psi_detector_base.PSIDetectorBase._update_service_config"
):
with mock.patch.object(ophyd, "cl") as mock_cl:
@ -50,9 +48,7 @@ def test_init():
sim_mode = False
dm = DMMock()
with mock.patch.object(dm, "connector"):
with mock.patch(
"ophyd_devices.epics.devices.psi_detector_base.FileWriterMixin"
), mock.patch(
with mock.patch("ophyd_devices.epics.devices.psi_detector_base.FileWriter"), mock.patch(
"ophyd_devices.epics.devices.psi_detector_base.PSIDetectorBase._update_service_config"
):
with mock.patch.object(ophyd, "cl") as mock_cl:

2
tests/test_falcon_csaxs.py
View File

@ -29,7 +29,7 @@ def mock_det():
dm = DMMock()
with mock.patch.object(dm, "connector"):
with mock.patch(
"ophyd_devices.epics.devices.psi_detector_base.FileWriterMixin"
"ophyd_devices.epics.devices.psi_detector_base.FileWriter"
) as filemixin, mock.patch(
"ophyd_devices.epics.devices.psi_detector_base.PSIDetectorBase._update_service_config"
) as mock_service_config:

6
tests/test_mcs_card.py
View File

@ -34,7 +34,7 @@ def mock_det():
dm = DMMock()
with mock.patch.object(dm, "connector"):
with mock.patch(
"ophyd_devices.epics.devices.psi_detector_base.FileWriterMixin"
"ophyd_devices.epics.devices.psi_detector_base.FileWriter"
) as filemixin, mock.patch(
"ophyd_devices.epics.devices.psi_detector_base.PSIDetectorBase._update_service_config"
) as mock_service_config:
@ -54,9 +54,7 @@ def test_init():
sim_mode = False
dm = DMMock()
with mock.patch.object(dm, "connector"):
with mock.patch(
"ophyd_devices.epics.devices.psi_detector_base.FileWriterMixin"
), mock.patch(
with mock.patch("ophyd_devices.epics.devices.psi_detector_base.FileWriter"), mock.patch(
"ophyd_devices.epics.devices.psi_detector_base.PSIDetectorBase._update_service_config"
):
with mock.patch.object(ophyd, "cl") as mock_cl:

4
tests/test_pilatus_csaxs.py
View File

@ -28,9 +28,7 @@ def mock_det():
sim_mode = False
dm = DMMock()
with mock.patch.object(dm, "connector"):
with mock.patch(
"ophyd_devices.epics.devices.psi_detector_base.FileWriterMixin"
), mock.patch(
with mock.patch("ophyd_devices.epics.devices.psi_detector_base.FileWriter"), mock.patch(
"ophyd_devices.epics.devices.psi_detector_base.PSIDetectorBase._update_service_config"
):
with mock.patch.object(ophyd, "cl") as mock_cl: