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

refactor: rename infomsgmock and add docstrings

Browse Source
This commit is contained in:
appel_c 2023-11-07 19:49:05 +01:00
parent 188c8321ca
commit 8a19ce1508
1 changed files with 30 additions and 15 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

45
ophyd_devices/epics/devices/bec_scaninfo_mixin.py
View File

@ -6,7 +6,12 @@ from bec_lib.core import bec_logger
logger = bec_logger.logger
class Bec_Info_Msg_Mock:
class BECInfoMsgMock:
"""Mock BECInfoMsg class
This class is used for mocking BECInfoMsg for testing purposes
"""
def __init__(
self,
mockrid: str = "mockrid1111",
@ -46,6 +51,16 @@ class Bec_Info_Msg_Mock:
class BecScaninfoMixin:
"""BecScaninfoMixin class
Args:
device_manager (DeviceManagerBase): DeviceManagerBase object
sim_mode (bool): Simulation mode flag
bec_info_msg (dict): BECInfoMsg object
Returns:
BecScaninfoMixin: BecScaninfoMixin object
"""
def __init__(
self, device_manager: DeviceManagerBase = None, sim_mode: bool = False, bec_info_msg=None
) -> None:
@ -54,30 +69,25 @@ class BecScaninfoMixin:
self.scan_msg = None
self.scanID = None
if bec_info_msg is None:
bec_info_msg_mock = Bec_Info_Msg_Mock()
self.bec_info_msg = bec_info_msg_mock.get_bec_info_msg()
BECInfoMsgMock = BECInfoMsgMock()
self.bec_info_msg = BECInfoMsgMock.get_bec_info_msg()
else:
self.bec_info_msg = bec_info_msg
# self.bec_info_msg = {
# "RID": "mockrid",
# "queueID": "mockqueuid",
# "scan_number": 1,
# "exp_time": 12e-3,
# "num_points": 500,
# "readout_time": 3e-3,
# "scan_type": "fly",
# "num_lines": 1,
# "frames_per_trigger": 1,
# }
def get_bec_info_msg(self) -> None:
"""Get BECInfoMsg object"""
return self.bec_info_msg
def change_config(self, bec_info_msg: dict) -> None:
"""Change BECInfoMsg object"""
self.bec_info_msg = bec_info_msg
def _get_current_scan_msg(self) -> BECMessage.ScanStatusMessage:
"""Get current scan message
Returns:
BECMessage.ScanStatusMessage: BECMessage.ScanStatusMessage object
"""
if not self.sim_mode:
# TODO what if no scan info is there yet!
msg = self.device_manager.producer.get(MessageEndpoints.scan_status())
@ -90,11 +100,16 @@ class BecScaninfoMixin:
)
def get_username(self) -> str:
"""Get username"""
if not self.sim_mode:
return self.device_manager.producer.get(MessageEndpoints.account()).decode()
return os.getlogin()
def load_scan_metadata(self) -> None:
"""Load scan metadata
This function loads scan metadata from the current scan message
"""
self.scan_msg = scan_msg = self._get_current_scan_msg()
logger.info(f"{self.scan_msg}")
try: