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

refactor: fix bec_lib imports

Browse Source
This commit is contained in:
usov_i
2023-11-13 08:28:09 +01:00
parent 3c173f59f7
commit d851cf6f8e
23 changed files with 74 additions and 80 deletions
Download Patch File Download Diff File Expand all files Collapse all files

8
ophyd_devices/eiger1p5m_csaxs/eiger1p5m.py
View File

@@ -2,7 +2,7 @@ import os
import time
from typing import List
from bec_lib.core import BECMessage, MessageEndpoints, bec_logger
from bec_lib import messages, MessageEndpoints, bec_logger
from ophyd import Component as Cpt
from ophyd import Device, DeviceStatus, EpicsSignal, EpicsSignalRO, Signal
@@ -81,9 +81,9 @@ class Eiger1p5MDetector(Device):
self.metadata = {}
self.username = "e20588" # TODO get from config
def _get_current_scan_msg(self) -> BECMessage.ScanStatusMessage:
def _get_current_scan_msg(self) -> messages.ScanStatusMessage:
msg = self.device_manager.producer.get(MessageEndpoints.scan_status())
return BECMessage.ScanStatusMessage.loads(msg)
return messages.ScanStatusMessage.loads(msg)
def _get_scan_dir(self, scan_bundle, scan_number, leading_zeros=None):
if leading_zeros is None:
@@ -159,7 +159,7 @@ class Eiger1p5MDetector(Device):
break
self.detector_control.put("stop")
signals = {"config": self.read(), "data": self.file_name}
msg = BECMessage.DeviceMessage(signals=signals, metadata=self.metadata)
msg = messages.DeviceMessage(signals=signals, metadata=self.metadata)
self.device_manager.producer.set_and_publish(
MessageEndpoints.device_read(self.name), msg.dumps()
)