# pylint: disable = no-name-in-module,missing-class-docstring, missing-module-docstring from math import inf from unittest.mock import MagicMock, patch import fakeredis import pytest from bec_lib.bec_service import messages from bec_lib.endpoints import MessageEndpoints from bec_lib.redis_connector import RedisConnector from bec_widgets.tests.utils import DEVICES, DMMock, FakePositioner, Positioner def fake_redis_server(host, port): redis = fakeredis.FakeRedis() return redis @pytest.fixture(scope="function") def mocked_client(bec_dispatcher): connector = RedisConnector("localhost:1", redis_cls=fake_redis_server) # Create a MagicMock object client = MagicMock() # TODO change to real BECClient # Shutdown the original client bec_dispatcher.client.shutdown() # Mock the connector attribute bec_dispatcher.client = client # Mock the device_manager.devices attribute client.connector = connector client.device_manager = DMMock() client.device_manager.add_devices(DEVICES) def mock_mv(*args, relative=False): # Extracting motor and value pairs for i in range(0, len(args), 2): motor = args[i] value = args[i + 1] motor.move(value, relative=relative) return MagicMock(wait=MagicMock()) client.scans = MagicMock(mv=mock_mv) # Ensure isinstance check for Positioner passes original_isinstance = isinstance def isinstance_mock(obj, class_info): if class_info == Positioner and isinstance(obj, FakePositioner): return True return original_isinstance(obj, class_info) with patch("builtins.isinstance", new=isinstance_mock): yield client connector.shutdown() # TODO change to real BECClient ################################################## # Client Fixture with DAP ################################################## @pytest.fixture(scope="function") def dap_plugin_message(): msg = messages.AvailableResourceMessage( **{ "resource": { "GaussianModel": { "class": "LmfitService1D", "user_friendly_name": "GaussianModel", "class_doc": "A model based on a Gaussian or normal distribution lineshape.\n\n The model has three Parameters: `amplitude`, `center`, and `sigma`.\n In addition, parameters `fwhm` and `height` are included as\n constraints to report full width at half maximum and maximum peak\n height, respectively.\n\n .. math::\n\n f(x; A, \\mu, \\sigma) = \\frac{A}{\\sigma\\sqrt{2\\pi}} e^{[{-{(x-\\mu)^2}/{{2\\sigma}^2}}]}\n\n where the parameter `amplitude` corresponds to :math:`A`, `center` to\n :math:`\\mu`, and `sigma` to :math:`\\sigma`. The full width at half\n maximum is :math:`2\\sigma\\sqrt{2\\ln{2}}`, approximately\n :math:`2.3548\\sigma`.\n\n For more information, see: https://en.wikipedia.org/wiki/Normal_distribution\n\n ", "run_doc": "A model based on a Gaussian or normal distribution lineshape.\n\n The model has three Parameters: `amplitude`, `center`, and `sigma`.\n In addition, parameters `fwhm` and `height` are included as\n constraints to report full width at half maximum and maximum peak\n height, respectively.\n\n .. math::\n\n f(x; A, \\mu, \\sigma) = \\frac{A}{\\sigma\\sqrt{2\\pi}} e^{[{-{(x-\\mu)^2}/{{2\\sigma}^2}}]}\n\n where the parameter `amplitude` corresponds to :math:`A`, `center` to\n :math:`\\mu`, and `sigma` to :math:`\\sigma`. The full width at half\n maximum is :math:`2\\sigma\\sqrt{2\\ln{2}}`, approximately\n :math:`2.3548\\sigma`.\n\n For more information, see: https://en.wikipedia.org/wiki/Normal_distribution\n\n \n Args:\n scan_item (ScanItem): Scan item or scan ID\n device_x (DeviceBase | str): Device name for x\n signal_x (DeviceBase | str): Signal name for x\n device_y (DeviceBase | str): Device name for y\n signal_y (DeviceBase | str): Signal name for y\n parameters (dict): Fit parameters\n ", "run_name": "fit", "signature": [ { "name": "args", "kind": "VAR_POSITIONAL", "default": "_empty", "annotation": "_empty", }, { "name": "scan_item", "kind": "KEYWORD_ONLY", "default": None, "annotation": "ScanItem | str", }, { "name": "device_x", "kind": "KEYWORD_ONLY", "default": None, "annotation": "DeviceBase | str", }, { "name": "signal_x", "kind": "KEYWORD_ONLY", "default": None, "annotation": "DeviceBase | str", }, { "name": "device_y", "kind": "KEYWORD_ONLY", "default": None, "annotation": "DeviceBase | str", }, { "name": "signal_y", "kind": "KEYWORD_ONLY", "default": None, "annotation": "DeviceBase | str", }, { "name": "parameters", "kind": "KEYWORD_ONLY", "default": None, "annotation": "dict", }, { "name": "kwargs", "kind": "VAR_KEYWORD", "default": "_empty", "annotation": "_empty", }, ], "auto_fit_supported": True, "params": { "amplitude": { "name": "amplitude", "value": 1.0, "vary": True, "min": -inf, "max": inf, "expr": None, "brute_step": None, "user_data": None, }, "center": { "name": "center", "value": 0.0, "vary": True, "min": -inf, "max": inf, "expr": None, "brute_step": None, "user_data": None, }, "sigma": { "name": "sigma", "value": 1.0, "vary": True, "min": 0, "max": inf, "expr": None, "brute_step": None, "user_data": None, }, "fwhm": { "name": "fwhm", "value": 2.35482, "vary": False, "min": -inf, "max": inf, "expr": "2.3548200*sigma", "brute_step": None, "user_data": None, }, "height": { "name": "height", "value": 0.3989423, "vary": False, "min": -inf, "max": inf, "expr": "0.3989423*amplitude/max(1e-15, sigma)", "brute_step": None, "user_data": None, }, }, "class_args": [], "class_kwargs": {"model": "GaussianModel"}, } } } ) yield msg @pytest.fixture(scope="function") def mocked_client_with_dap(mocked_client, dap_plugin_message): dap_services = { "BECClient": messages.StatusMessage(name="BECClient", status=1, info={}), "DAPServer/LmfitService1D": messages.StatusMessage( name="LmfitService1D", status=1, info={} ), } client = mocked_client client.service_status = dap_services client.connector.set( topic=MessageEndpoints.dap_available_plugins("dap"), msg=dap_plugin_message ) # Patch the client's DAP attribute so that the available models include "GaussianModel" patched_models = {"GaussianModel": {}, "LorentzModel": {}, "SineModel": {}} client.dap._available_dap_plugins = patched_models yield client class DummyData: def __init__(self, val, timestamps): self.val = val self.timestamps = timestamps def get(self, key, default=None): if key == "val": return self.val return default def create_dummy_scan_item(): """ Helper to create a dummy scan item with both live_data and metadata/status_message info. """ dummy_live_data = { "samx": {"samx": DummyData(val=[10, 20, 30], timestamps=[100, 200, 300])}, "samy": {"samy": DummyData(val=[5, 10, 15], timestamps=[100, 200, 300])}, "bpm4i": {"bpm4i": DummyData(val=[5, 6, 7], timestamps=[101, 201, 301])}, "async_device": {"async_device": DummyData(val=[1, 2, 3], timestamps=[11, 21, 31])}, } dummy_scan = MagicMock() dummy_scan.live_data = dummy_live_data dummy_scan.metadata = { "bec": { "scan_id": "dummy", "scan_report_devices": ["samx"], "readout_priority": {"monitored": ["bpm4i"], "async": ["async_device"]}, } } dummy_scan.status_message = MagicMock() dummy_scan.status_message.info = { "readout_priority": {"monitored": ["bpm4i"], "async": ["async_device"]}, "scan_report_devices": ["samx"], } return dummy_scan