usov_i-org/bec
1
0
Fork 0
mirror of https://github.com/ivan-usov-org/bec.git synced 2025-04-22 02:20:02 +02:00
Code Issues Actions Packages Projects Releases Wiki Activity

Merge branch 'scan_report_devices' into 'master'

Browse Source 
feat: added scan_report_devices

See merge request bec/bec!133
This commit is contained in:
wakonig_k 2022-12-19 14:07:20 +00:00
commit 4e2c3b704e
3 changed files with 35 additions and 5 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

15
scan_server/scan_server/scan_worker.py
View File

@ -343,13 +343,20 @@ class ScanWorker(threading.Thread):
self.device_manager.devices[dev] self.device_manager.devices[dev]
for dev in instr.content["parameter"].get("primary") for dev in instr.content["parameter"].get("primary")
] ]
metadata = self.current_instruction_queue_item.active_request_block.metadata active_rb = self.current_instruction_queue_item.active_request_block
metadata = active_rb.metadata
self.current_scan_info = {**instr.metadata, **instr.content["parameter"]} self.current_scan_info = {**instr.metadata, **instr.content["parameter"]}
self.current_scan_info.update(metadata) self.current_scan_info.update(metadata)
self.current_scan_info.update({"scan_number": self.parent.scan_number}) self.current_scan_info.update(
self.current_scan_info.update({"dataset_number": self.parent.dataset_number}) {
self.current_scan_info.update({"exp_time": self._exposure_time}) "scan_number": self.parent.scan_number,
"dataset_number": self.parent.dataset_number,
"exp_time": self._exposure_time,
"scan_report_hint": active_rb.scan.scan_report_hint,
"scan_report_devices": active_rb.scan.scan_report_devices,
}
)
self.current_scan_info["scan_msgs"] = [ self.current_scan_info["scan_msgs"] = [
str(scan_msg) for scan_msg in self.current_instruction_queue_item.scan_msgs str(scan_msg) for scan_msg in self.current_instruction_queue_item.scan_msgs
] ]

12
scan_server/scan_server/scans.py
View File

@ -1,7 +1,6 @@
import ast import ast
import enum import enum
import time import time
import uuid
from abc import ABC, abstractmethod from abc import ABC, abstractmethod
import numpy as np import numpy as np
@ -173,6 +172,7 @@ class RequestBase(ABC):
self.scan_motors = [] self.scan_motors = []
self.positions = [] self.positions = []
self._pre_scan_macros = [] self._pre_scan_macros = []
self._scan_report_devices = None
self._get_scan_motors() self._get_scan_motors()
if metadata is None: if metadata is None:
self.metadata = {} self.metadata = {}
@ -180,6 +180,16 @@ class RequestBase(ABC):
producer=self.device_manager.producer, device_msg_callback=self.device_msg_metadata producer=self.device_manager.producer, device_msg_callback=self.device_msg_metadata
) )
@property
def scan_report_devices(self):
if self._scan_report_devices is None:
return self.scan_motors
return self._scan_report_devices
@scan_report_devices.setter
def scan_report_devices(self, devices: list):
self._scan_report_devices = devices
def device_msg_metadata(self): def device_msg_metadata(self):
default_metadata = {"stream": "primary", "DIID": self.DIID} default_metadata = {"stream": "primary", "DIID": self.DIID}
metadata = {**default_metadata, **self.metadata} metadata = {**default_metadata, **self.metadata}

13
scan_server/tests/test_scans.py
View File

@ -995,6 +995,19 @@ def test_pre_scan_macro():
eval_mock.assert_called_once_with("pre_scan_macro") eval_mock.assert_called_once_with("pre_scan_macro")
def test_scan_report_devices():
device_manager = DMMock()
device_manager.add_device("samx")
parameter = {
"args": {"samx": (-5, 5), "samy": (-5, 5)},
"kwargs": {"step": 3},
}
request = FermatSpiralScan(device_manager=device_manager, parameter=parameter)
assert request.scan_report_devices == ["samx", "samy"]
request.scan_report_devices = ["samx", "samz"]
assert request.scan_report_devices == ["samx", "samz"]
@pytest.mark.parametrize("in_args,reference_positions", [((5, 5, 1, 1), [[1, 0], [2, 0], [-2, 0]])]) @pytest.mark.parametrize("in_args,reference_positions", [((5, 5, 1, 1), [[1, 0], [2, 0], [-2, 0]])])
def test_round_roi_scan_positions(in_args, reference_positions): def test_round_roi_scan_positions(in_args, reference_positions):
positions = get_round_roi_scan_positions(*in_args) positions = get_round_roi_scan_positions(*in_args)