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

refactor: ScanItem attributes bec and callback made private

Browse Source
This commit is contained in:
appel_c 2024-07-21 20:42:28 +02:00
parent 46738ad259
commit a70af8f58c
2 changed files with 11 additions and 11 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

10
bec_lib/bec_lib/scan_items.py
View File

@ -8,7 +8,6 @@ from __future__ import annotations
import builtins
import datetime
import importlib
import sys
import threading
import time
from collections import defaultdict, deque
@ -50,6 +49,7 @@ class ScanItem:
self.scan_number = scan_number
self.scan_id = scan_id
self.status = status
self.status_message = None
self.data = ScanData()
self.async_data = {}
self.baseline = ScanData()
@ -61,8 +61,8 @@ class ScanItem:
self.start_time = None
self.end_time = None
self.scan_report_instructions = []
self.callbacks = []
self.bec = builtins.__dict__.get("bec")
self._callbacks = []
self._bec = builtins.__dict__.get("bec")
@property
def queue(self):
@ -70,11 +70,11 @@ class ScanItem:
return self.scan_manager.queue_storage.find_queue_item_by_ID(self._queue_id)
def emit_data(self, scan_msg: messages.ScanMessage) -> None:
self.bec.callbacks.run("scan_segment", scan_msg.content, scan_msg.metadata)
self._bec.callbacks.run("scan_segment", scan_msg.content, scan_msg.metadata)
self._run_request_callbacks("scan_segment", scan_msg.content, scan_msg.metadata)
def emit_status(self, scan_status: messages.ScanStatusMessage) -> None:
self.bec.callbacks.run("scan_status", scan_status.content, scan_status.metadata)
self._bec.callbacks.run("scan_status", scan_status.content, scan_status.metadata)
self._run_request_callbacks("scan_status", scan_status.content, scan_status.metadata)
def _run_request_callbacks(self, event_type: str, data: dict, metadata: dict):

12
bec_lib/tests/test_scan_items.py
View File

@ -141,24 +141,24 @@ def test_scan_item_str_plain(scan_item):
def test_emit_data(scan_item):
scan_item.bec = mock.Mock()
scan_item.bec.callbacks = mock.Mock()
scan_item._bec = mock.Mock()
scan_item._bec.callbacks = mock.Mock()
scan_item._run_request_callbacks = mock.Mock()
msg = messages.ScanMessage(point_id=0, scan_id="scan_id", data={"samx": {"value": 1}})
scan_item.emit_data(msg)
scan_item.bec.callbacks.run.assert_called_once_with("scan_segment", msg.content, msg.metadata)
scan_item._bec.callbacks.run.assert_called_once_with("scan_segment", msg.content, msg.metadata)
scan_item._run_request_callbacks.assert_called_once_with(
"scan_segment", msg.content, msg.metadata
)
def test_emit_status(scan_item):
scan_item.bec = mock.Mock()
scan_item.bec.callbacks = mock.Mock()
scan_item._bec = mock.Mock()
scan_item._bec.callbacks = mock.Mock()
scan_item._run_request_callbacks = mock.Mock()
msg = messages.ScanStatusMessage(scan_id="scan_id", status="open", info={"info": "info"})
scan_item.emit_status(msg)
scan_item.bec.callbacks.run.assert_called_once_with("scan_status", msg.content, msg.metadata)
scan_item._bec.callbacks.run.assert_called_once_with("scan_status", msg.content, msg.metadata)
scan_item._run_request_callbacks.assert_called_once_with(
"scan_status", msg.content, msg.metadata
)