From 2bfe375165ecd4626b58d334df27ac8af48b27d3 Mon Sep 17 00:00:00 2001 From: Kathryn Baker Date: Wed, 3 Jan 2024 13:48:38 +0000 Subject: [PATCH 01/15] [ADD] MQTT & EPICS PVA interfaces --- interfaces/epics_pva/README.md | 5 + interfaces/epics_pva/__init__.py | 98 +++++++++++++++++ interfaces/epics_pva/configs.yaml | 8 ++ interfaces/mqtt/README.md | 5 + interfaces/mqtt/__init__.py | 175 ++++++++++++++++++++++++++++++ interfaces/mqtt/clients.py | 71 ++++++++++++ interfaces/mqtt/configs.yaml | 8 ++ 7 files changed, 370 insertions(+) create mode 100644 interfaces/epics_pva/README.md create mode 100644 interfaces/epics_pva/__init__.py create mode 100644 interfaces/epics_pva/configs.yaml create mode 100644 interfaces/mqtt/README.md create mode 100644 interfaces/mqtt/__init__.py create mode 100644 interfaces/mqtt/clients.py create mode 100644 interfaces/mqtt/configs.yaml diff --git a/interfaces/epics_pva/README.md b/interfaces/epics_pva/README.md new file mode 100644 index 0000000..a55e3e4 --- /dev/null +++ b/interfaces/epics_pva/README.md @@ -0,0 +1,5 @@ +# EPICS Interface for Badger + +## Prerequisites + +## Usage diff --git a/interfaces/epics_pva/__init__.py b/interfaces/epics_pva/__init__.py new file mode 100644 index 0000000..8e90443 --- /dev/null +++ b/interfaces/epics_pva/__init__.py @@ -0,0 +1,98 @@ +import logging +import multiprocessing as mp +import time +from copy import deepcopy +from typing import Dict, List + +import numpy as np +from badger import interface +from joblib import Parallel, delayed +from p4p.client.thread import Context + + +class Interface(interface.Interface): + name = "epics_pva" + """Concrete interface for interacting with EPICS PVAccess PVs""" + + def __init__(self, poll_period=0.1, timeout=3): + self.poll_period = poll_period + self.timeout = timeout + super().__init__() + + def get_default_params(self) -> dict: + return {"context": "pva"} + + def get_value(self, channel: str): + context = Context("pva") + try: + return context.get(channel).raw.value + except TimeoutError as e: + logging.exception(channel, e) + raise e + finally: + context.close() + + def get_values(self, channels: List[str]) -> Dict[str, float]: + time.sleep(self.poll_period) + context = Context("pva") + values = context.get(channels) + context.close() + return dict(zip(channels, values)) + + def set_value( + self, + channel: str, + value, + validate_readback=False, + readback_pv=None, + tolerance=1e-3, + count_down=10, + offset=0, + ): + # for parallel to work, context has to be made and closed within the function + context = Context("pva") + start_time = time.time() + time_limit = deepcopy(count_down) + # always put the value to the set PV + try: + context.put(channel, value, timeout=self.timeout, get=True) + except TypeError: + context.put(channel, value.item(), timeout=self.timeout, get=True) + logging.debug(f"put value {value} to {channel}") + # then, if configured to look at a readback PV, do the check on the PV + if validate_readback: + while count_down > 0: + # replace with monitor and conditional variables + _value = context.get(readback_pv, timeout=self.timeout).raw.value + if np.isclose(_value, value + offset, rtol=tolerance): + # should we return the set value or the read value here? + end_time = time.time() + logging.debug( + f"Set var for {channel} took {end_time - start_time:5.5f}s" + ) + return _value + + time.sleep(0.1) + count_down -= 0.1 + context.close() + raise Exception( + f"PV {channel} (current: {_value}) cannot reach expected value ({value}) in designated time {time_limit}!" + ) + else: + logging.debug(f"Set var for {channel} took {end_time - start_time:5.5f}s") + context.close() + return value + + def set_values(self, channels, values, configs: Dict[str, dict], parallel=False): + start = time.time() + if parallel: + Parallel(n_jobs=mp.cpu_count())( + delayed(self.set_value)(channel, value, **configs[channel]) + for channel, value in zip(channels, values) + ) + else: + for channel, value in zip(channels, values): + config = configs[channel] + self.set_value(channel, value, **config) + end = time.time() + logging.info(f"total set time: {end - start:.5f}s") diff --git a/interfaces/epics_pva/configs.yaml b/interfaces/epics_pva/configs.yaml new file mode 100644 index 0000000..79e5fb7 --- /dev/null +++ b/interfaces/epics_pva/configs.yaml @@ -0,0 +1,8 @@ +--- +name: epics_pva +version: "0.1" +dependencies: + - numpy + - badger-opt + - p4p + - joblib \ No newline at end of file diff --git a/interfaces/mqtt/README.md b/interfaces/mqtt/README.md new file mode 100644 index 0000000..de76202 --- /dev/null +++ b/interfaces/mqtt/README.md @@ -0,0 +1,5 @@ +# MQTT/Vsystem Interface for Badger + +## Prerequisites + +## Usage diff --git a/interfaces/mqtt/__init__.py b/interfaces/mqtt/__init__.py new file mode 100644 index 0000000..2424f52 --- /dev/null +++ b/interfaces/mqtt/__init__.py @@ -0,0 +1,175 @@ +import json +import logging +import multiprocessing as mp +import random +import string +import time +from copy import deepcopy +from typing import Dict, List + +import numpy as np +from badger import interface +from joblib import Parallel, delayed + +from .clients import MQTTClient, ValidationClient + + +def pv_name_to_mqtt_topic(pvname: str, mode="get"): + topic = pvname.replace("::", ":").replace(":", "/").lower() + if mode == "get": + prefix = "values" + suffix = "" + elif mode == "set": + prefix = "set" + suffix = "/value" + return f"vista/{prefix}/{topic}{suffix}" + + +def generate_shortuuid() -> str: + """Public function for generating short UUID messages to be attached + to MQTT messages as message_id's""" + alphabet = string.ascii_lowercase + string.ascii_uppercase + string.digits + shortuiid = "".join(random.choices(alphabet, k=8)) + return shortuiid + + +class Interface(interface.Interface): + name = "mqtt" + """Concrete interface for interacting with Vsystem via MQTT messages""" + + def __init__( + self, + host="mosquitto", + port=1883, + keepalive=60, + poll_period=0.1, + timeout=3, + ): + self.host = host + self.port = port + self.keepalive = keepalive + self.poll_period = poll_period + self.timeout = timeout + super().__init__() + + def get_default_params(self) -> dict: + params = { + "url": self.host, + "port": self.port, + "keepalive": self.keepalive, + } + return params + + def set_value( + self, + channel: str, + value, + validate_readback=False, + readback_pv=None, + tolerance=1e-3, + count_down=10, + offset=0, + ): + client = MQTTClient() + start_time = time.time() + time_limit = deepcopy(count_down) + # always put the value to the set PV + set_topic = pv_name_to_mqtt_topic(channel, mode="set") + if isinstance(value, np.ndarray): + value = value.item() + + payload = { + "timestamp": time.time(), + "channel": channel.lower(), + "value": value, + "messageid": generate_shortuuid(), + } + client.connect(self.host, self.port, self.keepalive) + client.publish(topic=set_topic, payload=json.dumps(payload)) + logging.debug(f"Published value {value} to {set_topic}") + # then, if configured to look at a readback PV, do the check on the PV + if validate_readback: + readback_topic = pv_name_to_mqtt_topic(readback_pv, mode="get") + validation_client = ValidationClient( + monitor_topic=readback_topic, + ) + validation_client.connect(self.host, self.port, self.keepalive) + validation_client.loop_start() + set_correct = False + while count_down > 0: + # set_correct = False + _value = validation_client.current_value + if _value is not None: + if np.isclose(_value, value + offset, rtol=tolerance): + # should we return the set value or the read value here? + end_time = time.time() + logging.info( + f"Set var for {channel} took {end_time - start_time:5.5f}s" + ) + set_correct = True + break + + time.sleep(0.1) + count_down -= 0.1 + # always stop the client once the validation is complete + validation_client.loop_stop() + validation_client.disconnect() + client.loop_stop() + client.disconnect() + if set_correct is False: + raise Exception( + f"PV {channel} (current: {_value}) cannot reach expected value ({value}) in designated time {time_limit}!" + ) + else: + return _value + else: + client.loop_stop() + client.disconnect() + end_time = time.time() + logging.info(f"Set var for {channel} took {end_time - start_time:5.5f}s") + return value + + def set_values(self, channels, values, configs: Dict[str, dict], parallel=False): + start = time.time() + if parallel: + Parallel(n_jobs=mp.cpu_count())( + delayed(self.set_value)(channel, value, **configs[channel]) + for channel, value in zip(channels, values) + ) + else: + for channel, value in zip(channels, values): + config = configs[channel] + self.set_value(channel, value, **config) + end = time.time() + logging.info(f"total set time: {end - start:.5f}s") + + def get_value(self, channel: str): + # we don't use this here because it's a stream of data instead of a request + return self.get_values([channel])[channel] + + def get_values(self, channels: List[str]) -> Dict[str, float]: + # in order to get values from MQTT, we need to start the loop and build + # up a message bank that we can then parse afterwards + client = MQTTClient(self.host, self.port, self.keepalive) + + client.initialise_messages(channels) + self._read_system(client, channels) + results = {} + for channel, value in client.messages.items(): + if len(value) == 0: + results[channel] = np.nan + else: + results[channel] = value[-1] + client.dump_messages() + + return results + + def _read_system(self, client, channels: List[str]): + client.connect(self.host, self.port, self.keepalive) + topics = [pv_name_to_mqtt_topic(channel, mode="get") for channel in channels] + client.subscribe([(topic, 0) for topic in topics]) + client.loop_start() + time.sleep(self.poll_period) + client.loop_stop() + client.unsubscribe(topics) + client.disconnect() diff --git a/interfaces/mqtt/clients.py b/interfaces/mqtt/clients.py new file mode 100644 index 0000000..a9c7aa7 --- /dev/null +++ b/interfaces/mqtt/clients.py @@ -0,0 +1,71 @@ +import json +import logging +from typing import List + +from paho.mqtt.client import Client + + +class MQTTClient(Client): + def __init__( + self, + *args, + **kwargs, + ) -> None: + super().__init__(*args, **kwargs) + + def on_connect(self, client, userdata, flags, rc): + logging.info(f"Connected with result code:{str(rc)}") + + def on_disconnect(self, client, userdata, rc): + logging.info(f"Disconnected with result code:{str(rc)}") + + def on_publish(self, client, userdata, mid): + logging.debug(f"Message published! ID: {mid}") + + def on_log(self, client, userdata, level, buf): + if level > 20: + # we only want to log messages over the INFO level + logging.info(level=level, msg=buf) + + def on_message(self, client, userdata, message): + message_byte = message.payload + message_dict = json.loads(message_byte.decode("utf-8")) + channel_name = message_dict["channel"] + + if channel_name.upper() in list(self.messages.keys()): + self.messages[channel_name.upper()].append(message_dict["value"]) + else: + self.messages[channel_name.upper()] = [message_dict["value"]] + + def dump_messages(self): + self.messages = {} + + def initialise_messages(self, channels: List[str]): + self.messages = {channel.upper(): [] for channel in channels} + + def subscribe(self, topic, qos=0, options=None, properties=None) -> tuple[int, int]: + logging.debug(f"subscribed to {topic}") + return super().subscribe(topic, qos, options, properties) + + +class ValidationClient(Client): + def __init__(self, monitor_topic, *args, **kwargs) -> None: + super().__init__(*args, **kwargs) + self.monitor_topic = monitor_topic + self.current_value = None + + def on_connect(self, client, userdata, flags, rc): + self.subscribe(self.monitor_topic) + logging.info( + f"\nValidation client connected with result code:{str(rc)}, subscribed to {self.monitor_topic}" + ) + + def on_disconnect(self, client, userdata, rc): + self.unsubscribe(self.monitor_topic) + logging.info(f"\nValidation client disconnected with result code:{str(rc)}") + + def on_message(self, client, userdata, message): + logging.debug("message received!") + message_byte = message.payload + message_dict = json.loads(message_byte.decode("utf-8")) + self.current_value = message_dict["value"] diff --git a/interfaces/mqtt/configs.yaml b/interfaces/mqtt/configs.yaml new file mode 100644 index 0000000..115ec2b --- /dev/null +++ b/interfaces/mqtt/configs.yaml @@ -0,0 +1,8 @@ +--- +name: mqtt +version: "0.1" +dependencies: + - numpy + - badger-opt + - paho-mqtt + - joblib From d79fd18e839138bbce1ae4d32c4efb50b8eb468b Mon Sep 17 00:00:00 2001 From: Kathryn Baker Date: Wed, 3 Jan 2024 17:19:03 +0000 Subject: [PATCH 02/15] [CHANGE] moved parallel into interface constructor --- interfaces/epics_pva/__init__.py | 9 +++++---- interfaces/mqtt/__init__.py | 6 ++++-- 2 files changed, 9 insertions(+), 6 deletions(-) diff --git a/interfaces/epics_pva/__init__.py b/interfaces/epics_pva/__init__.py index 8e90443..0f1a4aa 100644 --- a/interfaces/epics_pva/__init__.py +++ b/interfaces/epics_pva/__init__.py @@ -14,9 +14,10 @@ class Interface(interface.Interface): name = "epics_pva" """Concrete interface for interacting with EPICS PVAccess PVs""" - def __init__(self, poll_period=0.1, timeout=3): + def __init__(self, poll_period=0.1, timeout=3, parallel=False): self.poll_period = poll_period self.timeout = timeout + self.parallel = parallel super().__init__() def get_default_params(self) -> dict: @@ -37,7 +38,7 @@ def get_values(self, channels: List[str]) -> Dict[str, float]: context = Context("pva") values = context.get(channels) context.close() - return dict(zip(channels, values)) + return dict(zip(channels, [value.raw.value for value in values])) def set_value( self, @@ -83,9 +84,9 @@ def set_value( context.close() return value - def set_values(self, channels, values, configs: Dict[str, dict], parallel=False): + def set_values(self, channels, values, configs: Dict[str, dict]): start = time.time() - if parallel: + if self.parallel: Parallel(n_jobs=mp.cpu_count())( delayed(self.set_value)(channel, value, **configs[channel]) for channel, value in zip(channels, values) diff --git a/interfaces/mqtt/__init__.py b/interfaces/mqtt/__init__.py index 2424f52..bc1e55b 100644 --- a/interfaces/mqtt/__init__.py +++ b/interfaces/mqtt/__init__.py @@ -44,12 +44,14 @@ def __init__( keepalive=60, poll_period=0.1, timeout=3, + parallel=False, ): self.host = host self.port = port self.keepalive = keepalive self.poll_period = poll_period self.timeout = timeout + self.parallel = parallel super().__init__() def get_default_params(self) -> dict: @@ -129,9 +131,9 @@ def set_value( logging.info(f"Set var for {channel} took {end_time - start_time:5.5f}s") return value - def set_values(self, channels, values, configs: Dict[str, dict], parallel=False): + def set_values(self, channels, values, configs: Dict[str, dict]): start = time.time() - if parallel: + if self.parallel: Parallel(n_jobs=mp.cpu_count())( delayed(self.set_value)(channel, value, **configs[channel]) for channel, value in zip(channels, values) From 4a1f440b316f1cfadf3281e3b62565ecb01aa9ba Mon Sep 17 00:00:00 2001 From: Kathryn Baker Date: Thu, 4 Jan 2024 16:38:56 +0000 Subject: [PATCH 03/15] [ADD] unit tests for interfaces --- interfaces/epics_pva/__init__.py | 46 +++-- interfaces/epics_pva/tests.py | 166 ++++++++++++++++ interfaces/mqtt/__init__.py | 30 +-- interfaces/mqtt/clients.py | 1 + interfaces/mqtt/tests.py | 316 +++++++++++++++++++++++++++++++ 5 files changed, 531 insertions(+), 28 deletions(-) create mode 100644 interfaces/epics_pva/tests.py create mode 100644 interfaces/mqtt/tests.py diff --git a/interfaces/epics_pva/__init__.py b/interfaces/epics_pva/__init__.py index 0f1a4aa..4e8ba70 100644 --- a/interfaces/epics_pva/__init__.py +++ b/interfaces/epics_pva/__init__.py @@ -28,32 +28,40 @@ def get_value(self, channel: str): try: return context.get(channel).raw.value except TimeoutError as e: - logging.exception(channel, e) - raise e + # TODO - decide whether we should return a NaN value here + logging.exception(f"{channel}: {e}") + return np.nan + # raise e finally: context.close() def get_values(self, channels: List[str]) -> Dict[str, float]: time.sleep(self.poll_period) context = Context("pva") - values = context.get(channels) + try: + values = [value.raw.value for value in context.get(channels)] + except TimeoutError: + logging.exception(f"Timeout error from {channels}, retrying individually") + # if we get a timeout error one even one of them, we retry to get + # the individual values + values = [self.get_value(channel) for channel in channels] context.close() - return dict(zip(channels, [value.raw.value for value in values])) + return dict(zip(channels, values)) def set_value( self, channel: str, value, - validate_readback=False, - readback_pv=None, - tolerance=1e-3, - count_down=10, - offset=0, + set_config=None, + # validate_readback=False, + # readback_pv=None, + # tolerance=1e-3, + # count_down=10, + # offset=0, ): # for parallel to work, context has to be made and closed within the function context = Context("pva") start_time = time.time() - time_limit = deepcopy(count_down) # always put the value to the set PV try: context.put(channel, value, timeout=self.timeout, get=True) @@ -61,7 +69,12 @@ def set_value( context.put(channel, value.item(), timeout=self.timeout, get=True) logging.debug(f"put value {value} to {channel}") # then, if configured to look at a readback PV, do the check on the PV - if validate_readback: + if set_config is not None and set_config.get("validate_readback", False): + readback_pv = set_config.get("readback_pv") + tolerance = set_config.get("tolerance", 1e-3) + count_down = set_config.get("count_down", 10) + time_limit = deepcopy(count_down) + offset = set_config.get("offset", 0) while count_down > 0: # replace with monitor and conditional variables _value = context.get(readback_pv, timeout=self.timeout).raw.value @@ -71,15 +84,16 @@ def set_value( logging.debug( f"Set var for {channel} took {end_time - start_time:5.5f}s" ) + context.close() return _value time.sleep(0.1) count_down -= 0.1 - context.close() - raise Exception( + logging.exception( f"PV {channel} (current: {_value}) cannot reach expected value ({value}) in designated time {time_limit}!" ) else: + end_time = time.time() logging.debug(f"Set var for {channel} took {end_time - start_time:5.5f}s") context.close() return value @@ -88,12 +102,12 @@ def set_values(self, channels, values, configs: Dict[str, dict]): start = time.time() if self.parallel: Parallel(n_jobs=mp.cpu_count())( - delayed(self.set_value)(channel, value, **configs[channel]) + delayed(self.set_value)(channel, value, configs.get(channel)) for channel, value in zip(channels, values) ) else: for channel, value in zip(channels, values): - config = configs[channel] - self.set_value(channel, value, **config) + # TODO - what should we do if we can't get to the right value? + self.set_value(channel, value, configs.get(channel)) end = time.time() logging.info(f"total set time: {end - start:.5f}s") diff --git a/interfaces/epics_pva/tests.py b/interfaces/epics_pva/tests.py new file mode 100644 index 0000000..51a0450 --- /dev/null +++ b/interfaces/epics_pva/tests.py @@ -0,0 +1,166 @@ +import sys +from pathlib import Path + +current_file = Path(__file__) +root_dir = current_file.parents[1] +sys.path.insert(0, str(root_dir)) +from unittest.mock import MagicMock, patch + +import numpy as np +import pytest +from p4p.client.thread import TimeoutError + +channels = ["test::channel:1", "test::channel:2"] +values = [5.0, 6.0] + +from . import Interface + + +class MockRaw: + def __init__(self, value) -> None: + self.value = value + + +class p4pValue: + """Mock class designed to replicate the p4p Value class returned from `get` calls""" + + def __init__(self, value) -> None: + self.raw = MockRaw(value) + + +@pytest.fixture +def mock_context(): + with patch("epics_pva.Context") as ctx: + yield ctx + + +class TestEPICSInterface: + def test_epics_get_values(self, mock_context): + mock_context.return_value.get.return_value = [ + p4pValue(values[0]), + p4pValue(values[1]), + ] + interface = Interface(poll_period=0.1, timeout=1, parallel=False) + result = interface.get_values(channels) + assert result == { + "test::channel:1": values[0], + "test::channel:2": values[1], + } + mock_context.return_value.close.assert_called_once_with() + + def test_epics_get_values_timeout(self, caplog, mock_context): + # we get a timeout error first, then we get the individual values + mock_context.return_value.get.side_effect = [ + TimeoutError("original error"), + TimeoutError("individual timeout"), + p4pValue(values[1]), + ] + interface = Interface(poll_period=0.1, timeout=1, parallel=False) + result = interface.get_values(channels) + + # first check the returned values + assert np.isnan(result["test::channel:1"]) + assert result["test::channel:2"] == values[1] + + # then check the logs + assert len(caplog.records) == 2 + assert ( + caplog.records[0].getMessage() + == f"Timeout error from {channels}, retrying individually" + ) + assert caplog.records[1].getMessage() == f"{channels[0]}: individual timeout" + + assert len(mock_context.return_value.close.call_args_list) == len(channels) + 1 + + @pytest.mark.parametrize( + "test_input", + [ + ([value for value in values]), + ([np.array(value) for value in values]), + ], + ) + def test_epics_set_values_no_validation(self, mock_context, test_input): + mock_context.return_value.put = MagicMock() + interface = Interface(poll_period=0.1, timeout=1, parallel=False) + interface.set_values(channels, test_input, configs={}) + assert mock_context.return_value.put.call_args_list[0][0] == ( + channels[0], + values[0], + ) + assert mock_context.return_value.put.call_args_list[1][0] == ( + channels[1], + values[1], + ) + assert len(mock_context.return_value.close.call_args_list) == len(channels) + + def test_epics_set_values_with_validation(self, mock_context): + mock_context.return_value.put = MagicMock() + # we only configure one of the values to have validation so we only need to + # set values of the mock here + mock_context.return_value.get = MagicMock( + side_effect=[ + p4pValue(7), + p4pValue(values[0] - 0.2 + np.random.uniform(-0.05, 0.05)), + ] + ) + configs = { + "test::channel:1": { + "validate_readback": True, + "readback_pv": "test::channel:1:read", + "tolerance": 0.1, + "count_down": 0.11, + "offset": 0.2, + } + } + interface = Interface(poll_period=0.1, timeout=1, parallel=False) + interface.set_values(channels, values, configs=configs) + # check that the correct put values were called + assert mock_context.return_value.put.call_args_list[0][0] == ( + channels[0], + values[0], + ) + assert mock_context.return_value.put.call_args_list[1][0] == ( + channels[1], + values[1], + ) + # then check that the correct readback PV was used were called + for mock_call in mock_context.return_value.get.call_args_list: + assert mock_call[0][0] == "test::channel:1:read" + assert len(mock_context.return_value.close.call_args_list) == len(channels) + + def test_epics_set_values_validation_timeout(self, caplog, mock_context): + mock_context.return_value.put = MagicMock() + # we only configure one of the values to have validation so we only need to + # set values of the mock here + mock_context.return_value.get = MagicMock( + # in this case, neither value satisfies the conditions we've placed + side_effect=[p4pValue(7), p4pValue(7.01)] + ) + configs = { + "test::channel:1": { + "validate_readback": True, + "readback_pv": "test::channel:1:read", + "tolerance": 0.1, + "count_down": 0.11, + "offset": 0.2, + } + } + interface = Interface(poll_period=0.1, timeout=1, parallel=False) + interface.set_values(channels, values, configs=configs) + + # check that the correct put values were called + assert mock_context.return_value.put.call_args_list[0][0] == ( + channels[0], + values[0], + ) + assert mock_context.return_value.put.call_args_list[1][0] == ( + channels[1], + values[1], + ) + # check the error is logged correctly + assert len(caplog.records) == 1 + assert ( + caplog.records[0].getMessage() + == f"PV {channels[0]} (current: 7.01) cannot reach expected value ({float(values[0])}) in designated time 0.11!" + ) + assert len(mock_context.return_value.close.call_args_list) == len(channels) diff --git a/interfaces/mqtt/__init__.py b/interfaces/mqtt/__init__.py index bc1e55b..4d4b36f 100644 --- a/interfaces/mqtt/__init__.py +++ b/interfaces/mqtt/__init__.py @@ -66,15 +66,15 @@ def set_value( self, channel: str, value, - validate_readback=False, - readback_pv=None, - tolerance=1e-3, - count_down=10, - offset=0, + set_config=None, + # validate_readback=False, + # readback_pv=None, + # tolerance=1e-3, + # count_down=10, + # offset=0, ): client = MQTTClient() start_time = time.time() - time_limit = deepcopy(count_down) # always put the value to the set PV set_topic = pv_name_to_mqtt_topic(channel, mode="set") if isinstance(value, np.ndarray): @@ -90,8 +90,14 @@ def set_value( client.publish(topic=set_topic, payload=json.dumps(payload)) logging.debug(f"Published value {value} to {set_topic}") # then, if configured to look at a readback PV, do the check on the PV - if validate_readback: + if set_config is not None and set_config.get("validate_readback", False): + readback_pv = set_config.get("readback_pv") + tolerance = set_config.get("tolerance", 1e-3) + count_down = set_config.get("count_down", 10) + time_limit = deepcopy(count_down) + offset = set_config.get("offset", 0) readback_topic = pv_name_to_mqtt_topic(readback_pv, mode="get") + # set up the new client to store the incoming messages validation_client = ValidationClient( monitor_topic=readback_topic, ) @@ -119,7 +125,7 @@ def set_value( client.loop_stop() client.disconnect() if set_correct is False: - raise Exception( + logging.exception( f"PV {channel} (current: {_value}) cannot reach expected value ({value}) in designated time {time_limit}!" ) else: @@ -135,19 +141,19 @@ def set_values(self, channels, values, configs: Dict[str, dict]): start = time.time() if self.parallel: Parallel(n_jobs=mp.cpu_count())( - delayed(self.set_value)(channel, value, **configs[channel]) + delayed(self.set_value)(channel, value, configs.get(channel)) for channel, value in zip(channels, values) ) else: for channel, value in zip(channels, values): - config = configs[channel] - self.set_value(channel, value, **config) + # config = configs[channel] + self.set_value(channel, value, configs.get(channel)) end = time.time() logging.info(f"total set time: {end - start:.5f}s") def get_value(self, channel: str): # we don't use this here because it's a stream of data instead of a request - return self.get_values([channel])[channel] + return self.get_values([channel])[pv_name_to_mqtt_topic(channel, mode="get")] def get_values(self, channels: List[str]) -> Dict[str, float]: # in order to get values from MQTT, we need to start the loop and build diff --git a/interfaces/mqtt/clients.py b/interfaces/mqtt/clients.py index a9c7aa7..da981f7 100644 --- a/interfaces/mqtt/clients.py +++ b/interfaces/mqtt/clients.py @@ -11,6 +11,7 @@ def __init__( *args, **kwargs, ) -> None: + self.messages = {} super().__init__(*args, **kwargs) def on_connect(self, client, userdata, flags, rc): diff --git a/interfaces/mqtt/tests.py b/interfaces/mqtt/tests.py new file mode 100644 index 0000000..d0edefe --- /dev/null +++ b/interfaces/mqtt/tests.py @@ -0,0 +1,316 @@ +import sys +from pathlib import Path + +current_file = Path(__file__) +root_dir = current_file.parents[1] +sys.path.insert(0, str(root_dir)) +import json +from unittest.mock import MagicMock, patch + +import numpy as np +import pytest + +channels = ["test::channel:1", "test::channel:2", "TEST::CHANNEL:3"] +values = [5.0, 6.0, 7.0] + +from . import Interface, MQTTClient, ValidationClient + + +class MockMessage: + def __init__(self, topic, channel, value) -> None: + self._topic = topic + self._payload = { + "channel": channel, + "value": value, + "timestamp": 123.456, + "messageid": "test1234", + } + + @property + def topic(self): + return self._topic + + @property + def payload(self): + return json.dumps(self._payload).encode("utf-8") + + +def test_mqtt_client_messages(): + client = MQTTClient() + + client.initialise_messages(["test::channel:1", "TEST::CHANNEL:2"]) + assert client.messages == {"TEST::CHANNEL:1": [], "TEST::CHANNEL:2": []} + client.dump_messages() + assert client.messages == {} + + +def test_mqtt_client_on_messages(): + client = MQTTClient() + + test_messages = [ + ("vista/values/test/channel/1", "test::channel:1", 1.0), + ("vista/values/test/channel/2", "test::channel:2", 2.0), + ("vista/values/test/channel/1", "test::channel:1", 2.0), + ] + for topic, channel, value in test_messages: + client.on_message( + client, + MagicMock(), + MockMessage(topic, channel, value), + ) + + assert client.messages == {"TEST::CHANNEL:1": [1.0, 2.0], "TEST::CHANNEL:2": [2.0]} + + +def test_validation_client_on_messages(): + client = ValidationClient("vista/values/test/channel/1") + + test_messages = [ + ("vista/values/test/channel/1", "test::channel:1", 1.0), + ("vista/values/test/channel/1", "test::channel:2", 2.0), + ("vista/values/test/channel/1", "test::channel:1", 2.0), + ] + for topic, channel, value in test_messages: + client.on_message( + client, + MagicMock(), + MockMessage(topic, channel, value), + ) + assert client.current_value == value + + +@pytest.fixture +def mock_client(): + with patch("mqtt.MQTTClient") as client: + yield client.return_value + + +class TestMQTTInterface: + def test_MQTT_get_values(self, mock_client): + interface = Interface( + host="testhost", + port=1883, + keepalive=60, + poll_period=0.1, + timeout=1, + parallel=False, + ) + + # the on_message function on the client won't work as a mock because it needs threading + # so instead we have to fake what the messages might say + mock_client.messages = { + "vista/values/test/channel/1": [ + values[0], + values[0] + 1, + ], # an example of multiple values + "vista/values/test/channel/2": [values[1]], # an example of just one value + "vista/values/test/channel/3": [], # an example where no values are received + } + result = interface.get_values(channels) + + # check all the calls + mock_client.initialise_messages.assert_called_once_with(channels) + mock_client.loop_start.assert_called_once_with() + mock_client.loop_stop.assert_called_once_with() + mock_client.connect.assert_called_once_with("testhost", 1883, 60) + mock_client.disconnect.assert_called_once_with() + mock_client.dump_messages.assert_called_once_with() + + mock_client.subscribe.assert_called_once_with( + [ + ("vista/values/test/channel/1", 0), + ("vista/values/test/channel/2", 0), + ("vista/values/test/channel/3", 0), + ] + ) + mock_client.unsubscribe.assert_called_once_with( + [ + "vista/values/test/channel/1", + "vista/values/test/channel/2", + "vista/values/test/channel/3", + ] + ) + + assert result == { + "vista/values/test/channel/1": values[0] + 1, + "vista/values/test/channel/2": values[1], + "vista/values/test/channel/3": np.nan, + } + + def test_MQTT_get_value(self, mock_client): + # when we call get_value on it's own, we use the same procedure as with multiple + # channels but on it's own + interface = Interface( + host="testhost", + port=1883, + keepalive=60, + poll_period=0.1, + timeout=1, + parallel=False, + ) + channel_name = "test::channel:1" + # the on_message function on the client won't work as a mock because it needs threading + # so instead we have to fake what the messages might say + mock_client.messages = { + "vista/values/test/channel/1": [ + values[0], + values[0] + 1, + ], # an example of multiple values + } + result = interface.get_value("test::channel:1") + + # check all the calls + mock_client.initialise_messages.assert_called_once_with([channel_name]) + mock_client.loop_start.assert_called_once_with() + mock_client.loop_stop.assert_called_once_with() + mock_client.connect.assert_called_once_with("testhost", 1883, 60) + mock_client.disconnect.assert_called_once_with() + mock_client.dump_messages.assert_called_once_with() + + mock_client.subscribe.assert_called_once_with( + [ + ("vista/values/test/channel/1", 0), + ] + ) + mock_client.unsubscribe.assert_called_once_with( + [ + "vista/values/test/channel/1", + ] + ) + + assert result == values[0] + 1 + + @pytest.mark.parametrize( + "test_input", + [ + ([value for value in values]), + ([np.array(value) for value in values]), + ], + ) + def test_MQTT_set_values_no_validation(self, mock_client, test_input): + interface = Interface( + host="testhost", + port=1883, + keepalive=60, + poll_period=0.1, + timeout=1, + parallel=False, + ) + + interface.set_values(channels, test_input, configs={}) + + # check all the calls + assert len(mock_client.connect.call_args_list) == len(channels) + assert len(mock_client.disconnect.call_args_list) == len(channels) + + # check the contents of the publish message + for i in range(3): + assert ( + json.loads(mock_client.publish.call_args_list[i][1]["payload"])["value"] + == values[i] + ) + assert ( + json.loads(mock_client.publish.call_args_list[i][1]["payload"])[ + "channel" + ] + == f"test::channel:{i+1}" + ) + assert ( + mock_client.publish.call_args_list[i][1]["topic"] + == f"vista/set/test/channel/{i+1}/value" + ) + + def test_MQTT_set_values_with_validation(self, caplog, mock_client): + interface = Interface( + host="testhost", + port=1883, + keepalive=60, + poll_period=0.1, + timeout=1, + parallel=False, + ) + configs = { + "test::channel:1": { + "validate_readback": True, + "readback_pv": "test::channel:1:read", + "tolerance": 0.1, + "count_down": 0.11, + "offset": 0.2, + } + } + + with patch("mqtt.ValidationClient") as val_client: + # with a value that is within the tolerance, we shouldn'get any + # errors raised + val_client.return_value.current_value = ( + values[0] - 0.2 + np.random.uniform(-0.05, 0.05) + ) + interface.set_values(channels, values, configs=configs) + + # first check all the calls to the validation client - as we + # have only configured one PV to be validated, these should + # all only be called once + val_client.return_value.connect.assert_called_once_with( + "testhost", 1883, 60 + ) + val_client.return_value.disconnect.assert_called_once_with() + val_client.return_value.loop_start.assert_called_once_with() + val_client.return_value.loop_stop.assert_called_once_with() + + # if everything set correctly and within tolerance (as is in current_value) + # then we should get no logged errors + assert len(caplog.records) == 0 + + # then check the calls to the publish client - here we should have as + # many calls as there are channels + assert len(mock_client.connect.call_args_list) == len(channels) + assert len(mock_client.loop_stop.call_args_list) == len(channels) + assert len(mock_client.disconnect.call_args_list) == len(channels) + + def test_MQTT_set_values_with_validation_timeout(self, caplog, mock_client): + interface = Interface( + host="testhost", + port=1883, + keepalive=60, + poll_period=0.1, + timeout=1, + parallel=False, + ) + configs = { + "test::channel:1": { + "validate_readback": True, + "readback_pv": "test::channel:1:read", + "tolerance": 0.1, + "count_down": 0.11, + "offset": 0.2, + } + } + + with patch("mqtt.ValidationClient") as val_client: + # with a value that is within the tolerance, we shouldn'get any + # errors raised + val_client.return_value.current_value = 7 + interface.set_values(channels, values, configs=configs) + + # first check all the calls to the validation client - as we + # have only configured one PV to be validated, these should + # all only be called once + val_client.return_value.connect.assert_called_once_with( + "testhost", 1883, 60 + ) + val_client.return_value.disconnect.assert_called_once_with() + val_client.return_value.loop_start.assert_called_once_with() + val_client.return_value.loop_stop.assert_called_once_with() + + # although all the calls are correct, we should get an error logged that + # the value wasn't set correctly + assert len(caplog.records) == 1 + assert ( + caplog.records[0].getMessage() + == "PV test::channel:1 (current: 7) cannot reach expected value (5.0) in designated time 0.11!" + ) + # then check the calls to the publish client - here we should have as + # many calls as there are channels + assert len(mock_client.connect.call_args_list) == len(channels) + assert len(mock_client.loop_stop.call_args_list) == len(channels) + assert len(mock_client.disconnect.call_args_list) == len(channels) From ee46350746fbe787086a27ff4647cfa426314e02 Mon Sep 17 00:00:00 2001 From: Kathryn Baker Date: Wed, 24 Jan 2024 18:46:43 +0000 Subject: [PATCH 04/15] [TIDY] removed redundant params --- interfaces/epics_pva/__init__.py | 5 ----- interfaces/mqtt/__init__.py | 5 ----- 2 files changed, 10 deletions(-) diff --git a/interfaces/epics_pva/__init__.py b/interfaces/epics_pva/__init__.py index 4e8ba70..23b82a5 100644 --- a/interfaces/epics_pva/__init__.py +++ b/interfaces/epics_pva/__init__.py @@ -53,11 +53,6 @@ def set_value( channel: str, value, set_config=None, - # validate_readback=False, - # readback_pv=None, - # tolerance=1e-3, - # count_down=10, - # offset=0, ): # for parallel to work, context has to be made and closed within the function context = Context("pva") diff --git a/interfaces/mqtt/__init__.py b/interfaces/mqtt/__init__.py index 4d4b36f..5f66ebc 100644 --- a/interfaces/mqtt/__init__.py +++ b/interfaces/mqtt/__init__.py @@ -67,11 +67,6 @@ def set_value( channel: str, value, set_config=None, - # validate_readback=False, - # readback_pv=None, - # tolerance=1e-3, - # count_down=10, - # offset=0, ): client = MQTTClient() start_time = time.time() From e5295896d1cf5be05766121161565d573c4bb189 Mon Sep 17 00:00:00 2001 From: Kathryn Baker Date: Thu, 25 Jan 2024 10:04:37 +0000 Subject: [PATCH 05/15] [CHANGE] read-only flag in interfaces --- interfaces/epics_pva/__init__.py | 114 ++++++++++++---------- interfaces/epics_pva/tests.py | 35 ++++++- interfaces/mqtt/__init__.py | 157 +++++++++++++++++-------------- interfaces/mqtt/tests.py | 45 +++++++++ 4 files changed, 230 insertions(+), 121 deletions(-) diff --git a/interfaces/epics_pva/__init__.py b/interfaces/epics_pva/__init__.py index 23b82a5..14908ef 100644 --- a/interfaces/epics_pva/__init__.py +++ b/interfaces/epics_pva/__init__.py @@ -14,14 +14,18 @@ class Interface(interface.Interface): name = "epics_pva" """Concrete interface for interacting with EPICS PVAccess PVs""" - def __init__(self, poll_period=0.1, timeout=3, parallel=False): + def __init__(self, poll_period=0.1, timeout=3, parallel=False, read_only=False): self.poll_period = poll_period self.timeout = timeout self.parallel = parallel + self.read_only = read_only super().__init__() def get_default_params(self) -> dict: - return {"context": "pva"} + return { + "context": "pva", + "read-only": self.read_only, + } def get_value(self, channel: str): context = Context("pva") @@ -54,55 +58,67 @@ def set_value( value, set_config=None, ): - # for parallel to work, context has to be made and closed within the function - context = Context("pva") - start_time = time.time() - # always put the value to the set PV - try: - context.put(channel, value, timeout=self.timeout, get=True) - except TypeError: - context.put(channel, value.item(), timeout=self.timeout, get=True) - logging.debug(f"put value {value} to {channel}") - # then, if configured to look at a readback PV, do the check on the PV - if set_config is not None and set_config.get("validate_readback", False): - readback_pv = set_config.get("readback_pv") - tolerance = set_config.get("tolerance", 1e-3) - count_down = set_config.get("count_down", 10) - time_limit = deepcopy(count_down) - offset = set_config.get("offset", 0) - while count_down > 0: - # replace with monitor and conditional variables - _value = context.get(readback_pv, timeout=self.timeout).raw.value - if np.isclose(_value, value + offset, rtol=tolerance): - # should we return the set value or the read value here? - end_time = time.time() - logging.debug( - f"Set var for {channel} took {end_time - start_time:5.5f}s" - ) - context.close() - return _value + if not self.read_only: + # for parallel to work, context has to be made and closed within the function + context = Context("pva") + start_time = time.time() + # always put the value to the set PV + try: + context.put(channel, value, timeout=self.timeout, get=True) + except TypeError: + context.put(channel, value.item(), timeout=self.timeout, get=True) + logging.debug(f"put value {value} to {channel}") + # then, if configured to look at a readback PV, do the check on the PV + if set_config is not None and set_config.get("validate_readback", False): + readback_pv = set_config.get("readback_pv") + tolerance = set_config.get("tolerance", 1e-3) + count_down = set_config.get("count_down", 10) + time_limit = deepcopy(count_down) + offset = set_config.get("offset", 0) + while count_down > 0: + # replace with monitor and conditional variables + _value = context.get(readback_pv, timeout=self.timeout).raw.value + if np.isclose(_value, value + offset, rtol=tolerance): + # should we return the set value or the read value here? + end_time = time.time() + logging.debug( + f"Set var for {channel} took {end_time - start_time:5.5f}s" + ) + context.close() + return _value - time.sleep(0.1) - count_down -= 0.1 - logging.exception( - f"PV {channel} (current: {_value}) cannot reach expected value ({value}) in designated time {time_limit}!" - ) + time.sleep(0.1) + count_down -= 0.1 + logging.exception( + f"PV {channel} (current: {_value}) cannot reach expected value ({value}) in designated time {time_limit}!" + ) + else: + end_time = time.time() + logging.debug( + f"Set var for {channel} took {end_time - start_time:5.5f}s" + ) + context.close() + return value else: - end_time = time.time() - logging.debug(f"Set var for {channel} took {end_time - start_time:5.5f}s") - context.close() - return value + logging.info( + f"Interface is set to read-only mode, cannot set value {value} to {channel}" + ) def set_values(self, channels, values, configs: Dict[str, dict]): - start = time.time() - if self.parallel: - Parallel(n_jobs=mp.cpu_count())( - delayed(self.set_value)(channel, value, configs.get(channel)) - for channel, value in zip(channels, values) - ) + if not self.read_only: + start = time.time() + if self.parallel: + Parallel(n_jobs=mp.cpu_count())( + delayed(self.set_value)(channel, value, configs.get(channel)) + for channel, value in zip(channels, values) + ) + else: + for channel, value in zip(channels, values): + # TODO - what should we do if we can't get to the right value? + self.set_value(channel, value, configs.get(channel)) + end = time.time() + logging.info(f"total set time: {end - start:.5f}s") else: - for channel, value in zip(channels, values): - # TODO - what should we do if we can't get to the right value? - self.set_value(channel, value, configs.get(channel)) - end = time.time() - logging.info(f"total set time: {end - start:.5f}s") + logging.info( + f"Interface is set to read-only mode, cannot set values {values} to {channels}" + ) diff --git a/interfaces/epics_pva/tests.py b/interfaces/epics_pva/tests.py index 51a0450..4461065 100644 --- a/interfaces/epics_pva/tests.py +++ b/interfaces/epics_pva/tests.py @@ -4,6 +4,7 @@ current_file = Path(__file__) root_dir = current_file.parents[1] sys.path.insert(0, str(root_dir)) +import logging from unittest.mock import MagicMock, patch import numpy as np @@ -13,7 +14,7 @@ channels = ["test::channel:1", "test::channel:2"] values = [5.0, 6.0] -from . import Interface +from epics_pva import Interface class MockRaw: @@ -164,3 +165,35 @@ def test_epics_set_values_validation_timeout(self, caplog, mock_context): == f"PV {channels[0]} (current: 7.01) cannot reach expected value ({float(values[0])}) in designated time 0.11!" ) assert len(mock_context.return_value.close.call_args_list) == len(channels) + + def test_epics_set_values_read_only(self, caplog, mock_context): + caplog.set_level(logging.INFO) + mock_context.return_value.put = MagicMock() + interface = Interface( + poll_period=0.1, timeout=1, parallel=False, read_only=True + ) + interface.set_values(channels, values, configs={}) + + # check the log message is sent correctly and that context.put is not called + mock_context.return_value.put.assert_not_called() + assert len(caplog.records) == 1 + assert ( + caplog.records[0].getMessage() + == f"Interface is set to read-only mode, cannot set values {values} to {channels}" + ) + + def test_epics_set_value_read_only(self, caplog, mock_context): + caplog.set_level(logging.INFO) + mock_context.return_value.put = MagicMock() + interface = Interface( + poll_period=0.1, timeout=1, parallel=False, read_only=True + ) + interface.set_value(channels[0], values[0], set_config={}) + + # check the log message is sent correctly and that context.put is not called + mock_context.return_value.put.assert_not_called() + assert len(caplog.records) == 1 + assert ( + caplog.records[0].getMessage() + == f"Interface is set to read-only mode, cannot set value {values[0]} to {channels[0]}" + ) diff --git a/interfaces/mqtt/__init__.py b/interfaces/mqtt/__init__.py index 5f66ebc..d5f1667 100644 --- a/interfaces/mqtt/__init__.py +++ b/interfaces/mqtt/__init__.py @@ -45,6 +45,7 @@ def __init__( poll_period=0.1, timeout=3, parallel=False, + read_only=False, ): self.host = host self.port = port @@ -52,6 +53,7 @@ def __init__( self.poll_period = poll_period self.timeout = timeout self.parallel = parallel + self.read_only = read_only super().__init__() def get_default_params(self) -> dict: @@ -59,6 +61,7 @@ def get_default_params(self) -> dict: "url": self.host, "port": self.port, "keepalive": self.keepalive, + "read-only": self.read_only, } return params @@ -68,83 +71,95 @@ def set_value( value, set_config=None, ): - client = MQTTClient() - start_time = time.time() - # always put the value to the set PV - set_topic = pv_name_to_mqtt_topic(channel, mode="set") - if isinstance(value, np.ndarray): - value = value.item() - - payload = { - "timestamp": time.time(), - "channel": channel.lower(), - "value": value, - "messageid": generate_shortuuid(), - } - client.connect(self.host, self.port, self.keepalive) - client.publish(topic=set_topic, payload=json.dumps(payload)) - logging.debug(f"Published value {value} to {set_topic}") - # then, if configured to look at a readback PV, do the check on the PV - if set_config is not None and set_config.get("validate_readback", False): - readback_pv = set_config.get("readback_pv") - tolerance = set_config.get("tolerance", 1e-3) - count_down = set_config.get("count_down", 10) - time_limit = deepcopy(count_down) - offset = set_config.get("offset", 0) - readback_topic = pv_name_to_mqtt_topic(readback_pv, mode="get") - # set up the new client to store the incoming messages - validation_client = ValidationClient( - monitor_topic=readback_topic, - ) - validation_client.connect(self.host, self.port, self.keepalive) - validation_client.loop_start() - set_correct = False - while count_down > 0: - # set_correct = False - _value = validation_client.current_value - if _value is not None: - if np.isclose(_value, value + offset, rtol=tolerance): - # should we return the set value or the read value here? - end_time = time.time() - logging.info( - f"Set var for {channel} took {end_time - start_time:5.5f}s" - ) - set_correct = True - break - - time.sleep(0.1) - count_down -= 0.1 - # always stop the client once the validation is complete - validation_client.loop_stop() - validation_client.disconnect() - client.loop_stop() - client.disconnect() - if set_correct is False: - logging.exception( - f"PV {channel} (current: {_value}) cannot reach expected value ({value}) in designated time {time_limit}!" + if not self.read_only: + client = MQTTClient() + start_time = time.time() + # always put the value to the set PV + set_topic = pv_name_to_mqtt_topic(channel, mode="set") + if isinstance(value, np.ndarray): + value = value.item() + + payload = { + "timestamp": time.time(), + "channel": channel.lower(), + "value": value, + "messageid": generate_shortuuid(), + } + client.connect(self.host, self.port, self.keepalive) + client.publish(topic=set_topic, payload=json.dumps(payload)) + logging.debug(f"Published value {value} to {set_topic}") + # then, if configured to look at a readback PV, do the check on the PV + if set_config is not None and set_config.get("validate_readback", False): + readback_pv = set_config.get("readback_pv") + tolerance = set_config.get("tolerance", 1e-3) + count_down = set_config.get("count_down", 10) + time_limit = deepcopy(count_down) + offset = set_config.get("offset", 0) + readback_topic = pv_name_to_mqtt_topic(readback_pv, mode="get") + # set up the new client to store the incoming messages + validation_client = ValidationClient( + monitor_topic=readback_topic, ) + validation_client.connect(self.host, self.port, self.keepalive) + validation_client.loop_start() + set_correct = False + while count_down > 0: + # set_correct = False + _value = validation_client.current_value + if _value is not None: + if np.isclose(_value, value + offset, rtol=tolerance): + # should we return the set value or the read value here? + end_time = time.time() + logging.info( + f"Set var for {channel} took {end_time - start_time:5.5f}s" + ) + set_correct = True + break + + time.sleep(0.1) + count_down -= 0.1 + # always stop the client once the validation is complete + validation_client.loop_stop() + validation_client.disconnect() + client.loop_stop() + client.disconnect() + if set_correct is False: + logging.exception( + f"PV {channel} (current: {_value}) cannot reach expected value ({value}) in designated time {time_limit}!" + ) + else: + return _value else: - return _value + client.loop_stop() + client.disconnect() + end_time = time.time() + logging.info( + f"Set var for {channel} took {end_time - start_time:5.5f}s" + ) + return value else: - client.loop_stop() - client.disconnect() - end_time = time.time() - logging.info(f"Set var for {channel} took {end_time - start_time:5.5f}s") - return value + logging.info( + f"Interface is set to read-only mode, cannot set value {value} to {channel}" + ) def set_values(self, channels, values, configs: Dict[str, dict]): - start = time.time() - if self.parallel: - Parallel(n_jobs=mp.cpu_count())( - delayed(self.set_value)(channel, value, configs.get(channel)) - for channel, value in zip(channels, values) - ) + if not self.read_only: + start = time.time() + if self.parallel: + Parallel(n_jobs=mp.cpu_count())( + delayed(self.set_value)(channel, value, configs.get(channel)) + for channel, value in zip(channels, values) + ) + else: + for channel, value in zip(channels, values): + # config = configs[channel] + self.set_value(channel, value, configs.get(channel)) + end = time.time() + logging.info(f"total set time: {end - start:.5f}s") else: - for channel, value in zip(channels, values): - # config = configs[channel] - self.set_value(channel, value, configs.get(channel)) - end = time.time() - logging.info(f"total set time: {end - start:.5f}s") + logging.info( + f"Interface is set to read-only mode, cannot set values {values} to {channels}" + ) def get_value(self, channel: str): # we don't use this here because it's a stream of data instead of a request diff --git a/interfaces/mqtt/tests.py b/interfaces/mqtt/tests.py index d0edefe..d932148 100644 --- a/interfaces/mqtt/tests.py +++ b/interfaces/mqtt/tests.py @@ -5,6 +5,7 @@ root_dir = current_file.parents[1] sys.path.insert(0, str(root_dir)) import json +import logging from unittest.mock import MagicMock, patch import numpy as np @@ -314,3 +315,47 @@ def test_MQTT_set_values_with_validation_timeout(self, caplog, mock_client): assert len(mock_client.connect.call_args_list) == len(channels) assert len(mock_client.loop_stop.call_args_list) == len(channels) assert len(mock_client.disconnect.call_args_list) == len(channels) + + def test_mqtt_set_values_read_only(self, caplog, mock_client): + caplog.set_level(logging.INFO) + mock_client.connect = MagicMock() + interface = Interface( + host="testhost", + port=1883, + keepalive=60, + poll_period=0.1, + timeout=1, + parallel=False, + read_only=True, + ) + interface.set_values(channels, values, configs={}) + + # check the log message is sent correctly and that context.put is not called + mock_client.connect.assert_not_called() + assert len(caplog.records) == 1 + assert ( + caplog.records[0].getMessage() + == f"Interface is set to read-only mode, cannot set values {values} to {channels}" + ) + + def test_mqtt_set_value_read_only(self, caplog, mock_client): + caplog.set_level(logging.INFO) + mock_client.connect = MagicMock() + interface = Interface( + host="testhost", + port=1883, + keepalive=60, + poll_period=0.1, + timeout=1, + parallel=False, + read_only=True, + ) + interface.set_value(channels[0], values[0], set_config={}) + + # check the log message is sent correctly and that context.put is not called + mock_client.connect.assert_not_called() + assert len(caplog.records) == 1 + assert ( + caplog.records[0].getMessage() + == f"Interface is set to read-only mode, cannot set value {values[0]} to {channels[0]}" + ) From 8ac3ad7f39c0f3155a1b044c634a11227dca476f Mon Sep 17 00:00:00 2001 From: Kathryn Baker Date: Thu, 1 Feb 2024 16:51:13 +0000 Subject: [PATCH 06/15] [FIX] hot fix for machine physics in feb accounting for Enums, I/O errors and TimeoutErrors --- interfaces/epics_pva/__init__.py | 73 +++++++++++++++++++++++--------- 1 file changed, 53 insertions(+), 20 deletions(-) diff --git a/interfaces/epics_pva/__init__.py b/interfaces/epics_pva/__init__.py index 14908ef..d825b8f 100644 --- a/interfaces/epics_pva/__init__.py +++ b/interfaces/epics_pva/__init__.py @@ -40,10 +40,12 @@ def get_value(self, channel: str): context.close() def get_values(self, channels: List[str]) -> Dict[str, float]: - time.sleep(self.poll_period) + # time.sleep(self.poll_period) context = Context("pva") try: - values = [value.raw.value for value in context.get(channels)] + # using real allows us to quickly extract the number from both + # int/floats as well as enum types + values = [value.real for value in context.get(channels)] except TimeoutError: logging.exception(f"Timeout error from {channels}, retrying individually") # if we get a timeout error one even one of them, we retry to get @@ -51,6 +53,11 @@ def get_values(self, channels: List[str]) -> Dict[str, float]: values = [self.get_value(channel) for channel in channels] context.close() return dict(zip(channels, values)) + def _put(self, context, value,channel): + try: + context.put(channel, value, timeout=self.timeout, get=True) + except TypeError: + context.put(channel, value.item(), timeout=self.timeout, get=True) def set_value( self, @@ -64,9 +71,21 @@ def set_value( start_time = time.time() # always put the value to the set PV try: - context.put(channel, value, timeout=self.timeout, get=True) - except TypeError: - context.put(channel, value.item(), timeout=self.timeout, get=True) + self._put(context, channel, value) + # context.put(channel, value, timeout=self.timeout, get=True) + # except TypeError: + # context.put(channel, value.item(), timeout=self.timeout, get=True) + except TimeoutError: + logging.exception(f'Timeout Error on {channel}') + # we try again! + try: + self._put(context, channel, value) + except TimeoutError as e: + # give it some time and see if it fixes itself + time.sleep(1) + logging.exception(f'Timeout on {channel}: {e} after 2 attempts and a 1 second delay') + # raise TimeoutError(f'Timeout on {channel}: {e} after 2 tries and ') + # raise e logging.debug(f"put value {value} to {channel}") # then, if configured to look at a readback PV, do the check on the PV if set_config is not None and set_config.get("validate_readback", False): @@ -77,18 +96,23 @@ def set_value( offset = set_config.get("offset", 0) while count_down > 0: # replace with monitor and conditional variables - _value = context.get(readback_pv, timeout=self.timeout).raw.value - if np.isclose(_value, value + offset, rtol=tolerance): - # should we return the set value or the read value here? - end_time = time.time() - logging.debug( - f"Set var for {channel} took {end_time - start_time:5.5f}s" - ) - context.close() - return _value + _value = context.get(readback_pv, timeout=self.timeout) + severity = _value.severity + _value = _value.real + if severity != 3: + if np.isclose(_value, value + offset, atol=tolerance): + # should we return the set value or the read value here? + end_time = time.time() + logging.debug( + f"Set var for {channel} took {end_time - start_time:5.5f}s" + ) + context.close() + return _value - time.sleep(0.1) - count_down -= 0.1 + time.sleep(0.1) + count_down -= 0.1 + else: + logging.warning(f'readback PV {readback_pv} is in an I/O error state, validation will continue once it is out of this state') logging.exception( f"PV {channel} (current: {_value}) cannot reach expected value ({value}) in designated time {time_limit}!" ) @@ -108,10 +132,19 @@ def set_values(self, channels, values, configs: Dict[str, dict]): if not self.read_only: start = time.time() if self.parallel: - Parallel(n_jobs=mp.cpu_count())( - delayed(self.set_value)(channel, value, configs.get(channel)) - for channel, value in zip(channels, values) - ) + try: + Parallel(n_jobs=mp.cpu_count())( + delayed(self.set_value)(channel, value, configs.get(channel)) + for channel, value in zip(channels, values) + ) + except TimeoutError as e: + # if we get a TimeoutError, for now we want to just continue with + # the update and hope that getting the values after will calrify + # what the value actually was + # TODO this will have to be dealt with differently for a more + # generalisable solution across facilities as we want to make sure + # that the values going into the Xopt model are correct + logging.exception(e) else: for channel, value in zip(channels, values): # TODO - what should we do if we can't get to the right value? From 5b0c549865d6081452b021da7c9a5c2ec086f08f Mon Sep 17 00:00:00 2001 From: Kathryn Baker Date: Wed, 20 Mar 2024 17:03:36 +0000 Subject: [PATCH 07/15] [CHANGE] using validation function in set_value --- interfaces/epics_pva/__init__.py | 150 ++++++++++++++----------------- 1 file changed, 66 insertions(+), 84 deletions(-) diff --git a/interfaces/epics_pva/__init__.py b/interfaces/epics_pva/__init__.py index d825b8f..0b6830d 100644 --- a/interfaces/epics_pva/__init__.py +++ b/interfaces/epics_pva/__init__.py @@ -10,6 +10,36 @@ from p4p.client.thread import Context +def timeit(func): + def wrapper_timeit(*args, **kwargs): + start_time = time.time() + func(*args, **kwargs) + end_time = time.time() + logging.info(f"total set time: {end_time - start_time:.5f}s") + + return wrapper_timeit + + +def retry_on_timeout(func): + def wrapper_retry(*args, **kwargs): + channel = args[1] + try: + func(*args, **kwargs) + except TimeoutError: + # we try again! + # give it some time and see if it fixes itself + time.sleep(1) + try: + func(*args, **kwargs) + except TimeoutError as e: + # TODO - decide whether we should re-raise or not + raise TimeoutError( + f"Timeout on {channel}: {e} after 2 attempts and a 1 second delay" + ) + + return wrapper_retry + + class Interface(interface.Interface): name = "epics_pva" """Concrete interface for interacting with EPICS PVAccess PVs""" @@ -30,17 +60,15 @@ def get_default_params(self) -> dict: def get_value(self, channel: str): context = Context("pva") try: - return context.get(channel).raw.value + return context.get(channel).real except TimeoutError as e: # TODO - decide whether we should return a NaN value here logging.exception(f"{channel}: {e}") return np.nan - # raise e finally: context.close() def get_values(self, channels: List[str]) -> Dict[str, float]: - # time.sleep(self.poll_period) context = Context("pva") try: # using real allows us to quickly extract the number from both @@ -51,106 +79,60 @@ def get_values(self, channels: List[str]) -> Dict[str, float]: # if we get a timeout error one even one of them, we retry to get # the individual values values = [self.get_value(channel) for channel in channels] - context.close() + finally: + context.close() return dict(zip(channels, values)) - def _put(self, context, value,channel): + + @retry_on_timeout + def _put(self, context, channel, value): try: - context.put(channel, value, timeout=self.timeout, get=True) + context.put(channel, value, timeout=self.timeout) except TypeError: - context.put(channel, value.item(), timeout=self.timeout, get=True) - - def set_value( - self, - channel: str, - value, - set_config=None, - ): + context.put(channel, value.item(), timeout=self.timeout) + logging.debug(f"put value {value} to {channel}") + + def set_value(self, channel: str, value, validation_function=None): if not self.read_only: # for parallel to work, context has to be made and closed within the function context = Context("pva") - start_time = time.time() # always put the value to the set PV try: self._put(context, channel, value) - # context.put(channel, value, timeout=self.timeout, get=True) - # except TypeError: - # context.put(channel, value.item(), timeout=self.timeout, get=True) - except TimeoutError: - logging.exception(f'Timeout Error on {channel}') - # we try again! - try: - self._put(context, channel, value) - except TimeoutError as e: - # give it some time and see if it fixes itself - time.sleep(1) - logging.exception(f'Timeout on {channel}: {e} after 2 attempts and a 1 second delay') - # raise TimeoutError(f'Timeout on {channel}: {e} after 2 tries and ') - # raise e - logging.debug(f"put value {value} to {channel}") - # then, if configured to look at a readback PV, do the check on the PV - if set_config is not None and set_config.get("validate_readback", False): - readback_pv = set_config.get("readback_pv") - tolerance = set_config.get("tolerance", 1e-3) - count_down = set_config.get("count_down", 10) - time_limit = deepcopy(count_down) - offset = set_config.get("offset", 0) - while count_down > 0: - # replace with monitor and conditional variables - _value = context.get(readback_pv, timeout=self.timeout) - severity = _value.severity - _value = _value.real - if severity != 3: - if np.isclose(_value, value + offset, atol=tolerance): - # should we return the set value or the read value here? - end_time = time.time() - logging.debug( - f"Set var for {channel} took {end_time - start_time:5.5f}s" - ) - context.close() - return _value - - time.sleep(0.1) - count_down -= 0.1 - else: - logging.warning(f'readback PV {readback_pv} is in an I/O error state, validation will continue once it is out of this state') - logging.exception( - f"PV {channel} (current: {_value}) cannot reach expected value ({value}) in designated time {time_limit}!" - ) - else: - end_time = time.time() - logging.debug( - f"Set var for {channel} took {end_time - start_time:5.5f}s" - ) - context.close() - return value + if validation_function is not None: + try: + validation_function( + set_pv=channel, + set_value=value, + context=context, + timeout=self.timeout, + ) + except ValueError as e: + logging.warning(e) + # context.close() + # return value + except TimeoutError as e: + logging.exception(e) + finally: + context.close() else: logging.info( - f"Interface is set to read-only mode, cannot set value {value} to {channel}" + "Interface is set to read-only mode, cannot set value %s to %s", + value, + channel, ) + @timeit def set_values(self, channels, values, configs: Dict[str, dict]): if not self.read_only: - start = time.time() if self.parallel: - try: - Parallel(n_jobs=mp.cpu_count())( - delayed(self.set_value)(channel, value, configs.get(channel)) - for channel, value in zip(channels, values) - ) - except TimeoutError as e: - # if we get a TimeoutError, for now we want to just continue with - # the update and hope that getting the values after will calrify - # what the value actually was - # TODO this will have to be dealt with differently for a more - # generalisable solution across facilities as we want to make sure - # that the values going into the Xopt model are correct - logging.exception(e) + Parallel(n_jobs=mp.cpu_count())( + delayed(self.set_value)(channel, value, configs.get(channel)) + for channel, value in zip(channels, values) + ) + else: for channel, value in zip(channels, values): - # TODO - what should we do if we can't get to the right value? self.set_value(channel, value, configs.get(channel)) - end = time.time() - logging.info(f"total set time: {end - start:.5f}s") else: logging.info( f"Interface is set to read-only mode, cannot set values {values} to {channels}" From 41487839330ebe18d3ed605d2bc236716e35136d Mon Sep 17 00:00:00 2001 From: Kathryn Baker Date: Fri, 22 Mar 2024 11:26:37 +0000 Subject: [PATCH 08/15] [TIDY] small mods to logging messages --- interfaces/epics_pva/__init__.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/interfaces/epics_pva/__init__.py b/interfaces/epics_pva/__init__.py index 0b6830d..f329fa2 100644 --- a/interfaces/epics_pva/__init__.py +++ b/interfaces/epics_pva/__init__.py @@ -22,7 +22,7 @@ def wrapper_timeit(*args, **kwargs): def retry_on_timeout(func): def wrapper_retry(*args, **kwargs): - channel = args[1] + channel = args[2] try: func(*args, **kwargs) except TimeoutError: @@ -111,7 +111,7 @@ def set_value(self, channel: str, value, validation_function=None): # context.close() # return value except TimeoutError as e: - logging.exception(e) + logging.warning(e) finally: context.close() else: From d9958f42474e6abf95e129fcd5bf424742f3a878 Mon Sep 17 00:00:00 2001 From: Kathryn Baker Date: Mon, 25 Mar 2024 17:28:45 +0000 Subject: [PATCH 09/15] [CHANGE] removed MQTT interface code not tested and not a relevant interface for now --- interfaces/mqtt/README.md | 5 - interfaces/mqtt/__init__.py | 193 ------------------- interfaces/mqtt/clients.py | 72 ------- interfaces/mqtt/configs.yaml | 8 - interfaces/mqtt/tests.py | 361 ----------------------------------- 5 files changed, 639 deletions(-) delete mode 100644 interfaces/mqtt/README.md delete mode 100644 interfaces/mqtt/__init__.py delete mode 100644 interfaces/mqtt/clients.py delete mode 100644 interfaces/mqtt/configs.yaml delete mode 100644 interfaces/mqtt/tests.py diff --git a/interfaces/mqtt/README.md b/interfaces/mqtt/README.md deleted file mode 100644 index de76202..0000000 --- a/interfaces/mqtt/README.md +++ /dev/null @@ -1,5 +0,0 @@ -# MQTT/Vsystem Interface for Badger - -## Prerequisites - -## Usage diff --git a/interfaces/mqtt/__init__.py b/interfaces/mqtt/__init__.py deleted file mode 100644 index d5f1667..0000000 --- a/interfaces/mqtt/__init__.py +++ /dev/null @@ -1,193 +0,0 @@ -import json -import logging -import multiprocessing as mp -import random -import string -import time -from copy import deepcopy -from typing import Dict, List - -import numpy as np -from badger import interface -from joblib import Parallel, delayed - -from .clients import MQTTClient, ValidationClient - - -def pv_name_to_mqtt_topic(pvname: str, mode="get"): - topic = pvname.replace("::", ":").replace(":", "/").lower() - if mode == "get": - prefix = "values" - suffix = "" - elif mode == "set": - prefix = "set" - suffix = "/value" - return f"vista/{prefix}/{topic}{suffix}" - - -def generate_shortuuid() -> str: - """Public function for generating short UUID messages to be attached - to MQTT messages as message_id's""" - alphabet = string.ascii_lowercase + string.ascii_uppercase + string.digits - shortuiid = "".join(random.choices(alphabet, k=8)) - return shortuiid - - -class Interface(interface.Interface): - name = "mqtt" - """Concrete interface for interacting with Vsystem via MQTT messages""" - - def __init__( - self, - host="mosquitto", - port=1883, - keepalive=60, - poll_period=0.1, - timeout=3, - parallel=False, - read_only=False, - ): - self.host = host - self.port = port - self.keepalive = keepalive - self.poll_period = poll_period - self.timeout = timeout - self.parallel = parallel - self.read_only = read_only - super().__init__() - - def get_default_params(self) -> dict: - params = { - "url": self.host, - "port": self.port, - "keepalive": self.keepalive, - "read-only": self.read_only, - } - return params - - def set_value( - self, - channel: str, - value, - set_config=None, - ): - if not self.read_only: - client = MQTTClient() - start_time = time.time() - # always put the value to the set PV - set_topic = pv_name_to_mqtt_topic(channel, mode="set") - if isinstance(value, np.ndarray): - value = value.item() - - payload = { - "timestamp": time.time(), - "channel": channel.lower(), - "value": value, - "messageid": generate_shortuuid(), - } - client.connect(self.host, self.port, self.keepalive) - client.publish(topic=set_topic, payload=json.dumps(payload)) - logging.debug(f"Published value {value} to {set_topic}") - # then, if configured to look at a readback PV, do the check on the PV - if set_config is not None and set_config.get("validate_readback", False): - readback_pv = set_config.get("readback_pv") - tolerance = set_config.get("tolerance", 1e-3) - count_down = set_config.get("count_down", 10) - time_limit = deepcopy(count_down) - offset = set_config.get("offset", 0) - readback_topic = pv_name_to_mqtt_topic(readback_pv, mode="get") - # set up the new client to store the incoming messages - validation_client = ValidationClient( - monitor_topic=readback_topic, - ) - validation_client.connect(self.host, self.port, self.keepalive) - validation_client.loop_start() - set_correct = False - while count_down > 0: - # set_correct = False - _value = validation_client.current_value - if _value is not None: - if np.isclose(_value, value + offset, rtol=tolerance): - # should we return the set value or the read value here? - end_time = time.time() - logging.info( - f"Set var for {channel} took {end_time - start_time:5.5f}s" - ) - set_correct = True - break - - time.sleep(0.1) - count_down -= 0.1 - # always stop the client once the validation is complete - validation_client.loop_stop() - validation_client.disconnect() - client.loop_stop() - client.disconnect() - if set_correct is False: - logging.exception( - f"PV {channel} (current: {_value}) cannot reach expected value ({value}) in designated time {time_limit}!" - ) - else: - return _value - else: - client.loop_stop() - client.disconnect() - end_time = time.time() - logging.info( - f"Set var for {channel} took {end_time - start_time:5.5f}s" - ) - return value - else: - logging.info( - f"Interface is set to read-only mode, cannot set value {value} to {channel}" - ) - - def set_values(self, channels, values, configs: Dict[str, dict]): - if not self.read_only: - start = time.time() - if self.parallel: - Parallel(n_jobs=mp.cpu_count())( - delayed(self.set_value)(channel, value, configs.get(channel)) - for channel, value in zip(channels, values) - ) - else: - for channel, value in zip(channels, values): - # config = configs[channel] - self.set_value(channel, value, configs.get(channel)) - end = time.time() - logging.info(f"total set time: {end - start:.5f}s") - else: - logging.info( - f"Interface is set to read-only mode, cannot set values {values} to {channels}" - ) - - def get_value(self, channel: str): - # we don't use this here because it's a stream of data instead of a request - return self.get_values([channel])[pv_name_to_mqtt_topic(channel, mode="get")] - - def get_values(self, channels: List[str]) -> Dict[str, float]: - # in order to get values from MQTT, we need to start the loop and build - # up a message bank that we can then parse afterwards - client = MQTTClient(self.host, self.port, self.keepalive) - - client.initialise_messages(channels) - self._read_system(client, channels) - results = {} - for channel, value in client.messages.items(): - if len(value) == 0: - results[channel] = np.nan - else: - results[channel] = value[-1] - client.dump_messages() - - return results - - def _read_system(self, client, channels: List[str]): - client.connect(self.host, self.port, self.keepalive) - topics = [pv_name_to_mqtt_topic(channel, mode="get") for channel in channels] - client.subscribe([(topic, 0) for topic in topics]) - client.loop_start() - time.sleep(self.poll_period) - client.loop_stop() - client.unsubscribe(topics) - client.disconnect() diff --git a/interfaces/mqtt/clients.py b/interfaces/mqtt/clients.py deleted file mode 100644 index da981f7..0000000 --- a/interfaces/mqtt/clients.py +++ /dev/null @@ -1,72 +0,0 @@ -import json -import logging -from typing import List - -from paho.mqtt.client import Client - - -class MQTTClient(Client): - def __init__( - self, - *args, - **kwargs, - ) -> None: - self.messages = {} - super().__init__(*args, **kwargs) - - def on_connect(self, client, userdata, flags, rc): - logging.info(f"Connected with result code:{str(rc)}") - - def on_disconnect(self, client, userdata, rc): - logging.info(f"Disconnected with result code:{str(rc)}") - - def on_publish(self, client, userdata, mid): - logging.debug(f"Message published! ID: {mid}") - - def on_log(self, client, userdata, level, buf): - if level > 20: - # we only want to log messages over the INFO level - logging.info(level=level, msg=buf) - - def on_message(self, client, userdata, message): - message_byte = message.payload - message_dict = json.loads(message_byte.decode("utf-8")) - channel_name = message_dict["channel"] - - if channel_name.upper() in list(self.messages.keys()): - self.messages[channel_name.upper()].append(message_dict["value"]) - else: - self.messages[channel_name.upper()] = [message_dict["value"]] - - def dump_messages(self): - self.messages = {} - - def initialise_messages(self, channels: List[str]): - self.messages = {channel.upper(): [] for channel in channels} - - def subscribe(self, topic, qos=0, options=None, properties=None) -> tuple[int, int]: - logging.debug(f"subscribed to {topic}") - return super().subscribe(topic, qos, options, properties) - - -class ValidationClient(Client): - def __init__(self, monitor_topic, *args, **kwargs) -> None: - super().__init__(*args, **kwargs) - self.monitor_topic = monitor_topic - self.current_value = None - - def on_connect(self, client, userdata, flags, rc): - self.subscribe(self.monitor_topic) - logging.info( - f"\nValidation client connected with result code:{str(rc)}, subscribed to {self.monitor_topic}" - ) - - def on_disconnect(self, client, userdata, rc): - self.unsubscribe(self.monitor_topic) - logging.info(f"\nValidation client disconnected with result code:{str(rc)}") - - def on_message(self, client, userdata, message): - logging.debug("message received!") - message_byte = message.payload - message_dict = json.loads(message_byte.decode("utf-8")) - self.current_value = message_dict["value"] diff --git a/interfaces/mqtt/configs.yaml b/interfaces/mqtt/configs.yaml deleted file mode 100644 index 115ec2b..0000000 --- a/interfaces/mqtt/configs.yaml +++ /dev/null @@ -1,8 +0,0 @@ ---- -name: mqtt -version: "0.1" -dependencies: - - numpy - - badger-opt - - paho-mqtt - - joblib diff --git a/interfaces/mqtt/tests.py b/interfaces/mqtt/tests.py deleted file mode 100644 index d932148..0000000 --- a/interfaces/mqtt/tests.py +++ /dev/null @@ -1,361 +0,0 @@ -import sys -from pathlib import Path - -current_file = Path(__file__) -root_dir = current_file.parents[1] -sys.path.insert(0, str(root_dir)) -import json -import logging -from unittest.mock import MagicMock, patch - -import numpy as np -import pytest - -channels = ["test::channel:1", "test::channel:2", "TEST::CHANNEL:3"] -values = [5.0, 6.0, 7.0] - -from . import Interface, MQTTClient, ValidationClient - - -class MockMessage: - def __init__(self, topic, channel, value) -> None: - self._topic = topic - self._payload = { - "channel": channel, - "value": value, - "timestamp": 123.456, - "messageid": "test1234", - } - - @property - def topic(self): - return self._topic - - @property - def payload(self): - return json.dumps(self._payload).encode("utf-8") - - -def test_mqtt_client_messages(): - client = MQTTClient() - - client.initialise_messages(["test::channel:1", "TEST::CHANNEL:2"]) - assert client.messages == {"TEST::CHANNEL:1": [], "TEST::CHANNEL:2": []} - client.dump_messages() - assert client.messages == {} - - -def test_mqtt_client_on_messages(): - client = MQTTClient() - - test_messages = [ - ("vista/values/test/channel/1", "test::channel:1", 1.0), - ("vista/values/test/channel/2", "test::channel:2", 2.0), - ("vista/values/test/channel/1", "test::channel:1", 2.0), - ] - for topic, channel, value in test_messages: - client.on_message( - client, - MagicMock(), - MockMessage(topic, channel, value), - ) - - assert client.messages == {"TEST::CHANNEL:1": [1.0, 2.0], "TEST::CHANNEL:2": [2.0]} - - -def test_validation_client_on_messages(): - client = ValidationClient("vista/values/test/channel/1") - - test_messages = [ - ("vista/values/test/channel/1", "test::channel:1", 1.0), - ("vista/values/test/channel/1", "test::channel:2", 2.0), - ("vista/values/test/channel/1", "test::channel:1", 2.0), - ] - for topic, channel, value in test_messages: - client.on_message( - client, - MagicMock(), - MockMessage(topic, channel, value), - ) - assert client.current_value == value - - -@pytest.fixture -def mock_client(): - with patch("mqtt.MQTTClient") as client: - yield client.return_value - - -class TestMQTTInterface: - def test_MQTT_get_values(self, mock_client): - interface = Interface( - host="testhost", - port=1883, - keepalive=60, - poll_period=0.1, - timeout=1, - parallel=False, - ) - - # the on_message function on the client won't work as a mock because it needs threading - # so instead we have to fake what the messages might say - mock_client.messages = { - "vista/values/test/channel/1": [ - values[0], - values[0] + 1, - ], # an example of multiple values - "vista/values/test/channel/2": [values[1]], # an example of just one value - "vista/values/test/channel/3": [], # an example where no values are received - } - result = interface.get_values(channels) - - # check all the calls - mock_client.initialise_messages.assert_called_once_with(channels) - mock_client.loop_start.assert_called_once_with() - mock_client.loop_stop.assert_called_once_with() - mock_client.connect.assert_called_once_with("testhost", 1883, 60) - mock_client.disconnect.assert_called_once_with() - mock_client.dump_messages.assert_called_once_with() - - mock_client.subscribe.assert_called_once_with( - [ - ("vista/values/test/channel/1", 0), - ("vista/values/test/channel/2", 0), - ("vista/values/test/channel/3", 0), - ] - ) - mock_client.unsubscribe.assert_called_once_with( - [ - "vista/values/test/channel/1", - "vista/values/test/channel/2", - "vista/values/test/channel/3", - ] - ) - - assert result == { - "vista/values/test/channel/1": values[0] + 1, - "vista/values/test/channel/2": values[1], - "vista/values/test/channel/3": np.nan, - } - - def test_MQTT_get_value(self, mock_client): - # when we call get_value on it's own, we use the same procedure as with multiple - # channels but on it's own - interface = Interface( - host="testhost", - port=1883, - keepalive=60, - poll_period=0.1, - timeout=1, - parallel=False, - ) - channel_name = "test::channel:1" - # the on_message function on the client won't work as a mock because it needs threading - # so instead we have to fake what the messages might say - mock_client.messages = { - "vista/values/test/channel/1": [ - values[0], - values[0] + 1, - ], # an example of multiple values - } - result = interface.get_value("test::channel:1") - - # check all the calls - mock_client.initialise_messages.assert_called_once_with([channel_name]) - mock_client.loop_start.assert_called_once_with() - mock_client.loop_stop.assert_called_once_with() - mock_client.connect.assert_called_once_with("testhost", 1883, 60) - mock_client.disconnect.assert_called_once_with() - mock_client.dump_messages.assert_called_once_with() - - mock_client.subscribe.assert_called_once_with( - [ - ("vista/values/test/channel/1", 0), - ] - ) - mock_client.unsubscribe.assert_called_once_with( - [ - "vista/values/test/channel/1", - ] - ) - - assert result == values[0] + 1 - - @pytest.mark.parametrize( - "test_input", - [ - ([value for value in values]), - ([np.array(value) for value in values]), - ], - ) - def test_MQTT_set_values_no_validation(self, mock_client, test_input): - interface = Interface( - host="testhost", - port=1883, - keepalive=60, - poll_period=0.1, - timeout=1, - parallel=False, - ) - - interface.set_values(channels, test_input, configs={}) - - # check all the calls - assert len(mock_client.connect.call_args_list) == len(channels) - assert len(mock_client.disconnect.call_args_list) == len(channels) - - # check the contents of the publish message - for i in range(3): - assert ( - json.loads(mock_client.publish.call_args_list[i][1]["payload"])["value"] - == values[i] - ) - assert ( - json.loads(mock_client.publish.call_args_list[i][1]["payload"])[ - "channel" - ] - == f"test::channel:{i+1}" - ) - assert ( - mock_client.publish.call_args_list[i][1]["topic"] - == f"vista/set/test/channel/{i+1}/value" - ) - - def test_MQTT_set_values_with_validation(self, caplog, mock_client): - interface = Interface( - host="testhost", - port=1883, - keepalive=60, - poll_period=0.1, - timeout=1, - parallel=False, - ) - configs = { - "test::channel:1": { - "validate_readback": True, - "readback_pv": "test::channel:1:read", - "tolerance": 0.1, - "count_down": 0.11, - "offset": 0.2, - } - } - - with patch("mqtt.ValidationClient") as val_client: - # with a value that is within the tolerance, we shouldn'get any - # errors raised - val_client.return_value.current_value = ( - values[0] - 0.2 + np.random.uniform(-0.05, 0.05) - ) - interface.set_values(channels, values, configs=configs) - - # first check all the calls to the validation client - as we - # have only configured one PV to be validated, these should - # all only be called once - val_client.return_value.connect.assert_called_once_with( - "testhost", 1883, 60 - ) - val_client.return_value.disconnect.assert_called_once_with() - val_client.return_value.loop_start.assert_called_once_with() - val_client.return_value.loop_stop.assert_called_once_with() - - # if everything set correctly and within tolerance (as is in current_value) - # then we should get no logged errors - assert len(caplog.records) == 0 - - # then check the calls to the publish client - here we should have as - # many calls as there are channels - assert len(mock_client.connect.call_args_list) == len(channels) - assert len(mock_client.loop_stop.call_args_list) == len(channels) - assert len(mock_client.disconnect.call_args_list) == len(channels) - - def test_MQTT_set_values_with_validation_timeout(self, caplog, mock_client): - interface = Interface( - host="testhost", - port=1883, - keepalive=60, - poll_period=0.1, - timeout=1, - parallel=False, - ) - configs = { - "test::channel:1": { - "validate_readback": True, - "readback_pv": "test::channel:1:read", - "tolerance": 0.1, - "count_down": 0.11, - "offset": 0.2, - } - } - - with patch("mqtt.ValidationClient") as val_client: - # with a value that is within the tolerance, we shouldn'get any - # errors raised - val_client.return_value.current_value = 7 - interface.set_values(channels, values, configs=configs) - - # first check all the calls to the validation client - as we - # have only configured one PV to be validated, these should - # all only be called once - val_client.return_value.connect.assert_called_once_with( - "testhost", 1883, 60 - ) - val_client.return_value.disconnect.assert_called_once_with() - val_client.return_value.loop_start.assert_called_once_with() - val_client.return_value.loop_stop.assert_called_once_with() - - # although all the calls are correct, we should get an error logged that - # the value wasn't set correctly - assert len(caplog.records) == 1 - assert ( - caplog.records[0].getMessage() - == "PV test::channel:1 (current: 7) cannot reach expected value (5.0) in designated time 0.11!" - ) - # then check the calls to the publish client - here we should have as - # many calls as there are channels - assert len(mock_client.connect.call_args_list) == len(channels) - assert len(mock_client.loop_stop.call_args_list) == len(channels) - assert len(mock_client.disconnect.call_args_list) == len(channels) - - def test_mqtt_set_values_read_only(self, caplog, mock_client): - caplog.set_level(logging.INFO) - mock_client.connect = MagicMock() - interface = Interface( - host="testhost", - port=1883, - keepalive=60, - poll_period=0.1, - timeout=1, - parallel=False, - read_only=True, - ) - interface.set_values(channels, values, configs={}) - - # check the log message is sent correctly and that context.put is not called - mock_client.connect.assert_not_called() - assert len(caplog.records) == 1 - assert ( - caplog.records[0].getMessage() - == f"Interface is set to read-only mode, cannot set values {values} to {channels}" - ) - - def test_mqtt_set_value_read_only(self, caplog, mock_client): - caplog.set_level(logging.INFO) - mock_client.connect = MagicMock() - interface = Interface( - host="testhost", - port=1883, - keepalive=60, - poll_period=0.1, - timeout=1, - parallel=False, - read_only=True, - ) - interface.set_value(channels[0], values[0], set_config={}) - - # check the log message is sent correctly and that context.put is not called - mock_client.connect.assert_not_called() - assert len(caplog.records) == 1 - assert ( - caplog.records[0].getMessage() - == f"Interface is set to read-only mode, cannot set value {values[0]} to {channels[0]}" - ) From b956e4a1d73a682f02d2596f66f567257899b8f1 Mon Sep 17 00:00:00 2001 From: Kathryn Baker Date: Wed, 10 Apr 2024 15:06:39 +0100 Subject: [PATCH 10/15] [CHANGE] updated to use pydantic fields --- interfaces/epics_pva/__init__.py | 42 ++++++++++++++++++++------------ 1 file changed, 26 insertions(+), 16 deletions(-) diff --git a/interfaces/epics_pva/__init__.py b/interfaces/epics_pva/__init__.py index f329fa2..7119a72 100644 --- a/interfaces/epics_pva/__init__.py +++ b/interfaces/epics_pva/__init__.py @@ -2,12 +2,14 @@ import multiprocessing as mp import time from copy import deepcopy -from typing import Dict, List +from functools import partial +from typing import Any, Dict, List import numpy as np from badger import interface from joblib import Parallel, delayed from p4p.client.thread import Context +from pydantic import Field def timeit(func): @@ -41,24 +43,30 @@ def wrapper_retry(*args, **kwargs): class Interface(interface.Interface): - name = "epics_pva" """Concrete interface for interacting with EPICS PVAccess PVs""" - def __init__(self, poll_period=0.1, timeout=3, parallel=False, read_only=False): - self.poll_period = poll_period - self.timeout = timeout - self.parallel = parallel - self.read_only = read_only - super().__init__() + name: str = "epics_pva" + context_str: str = "pva" + timeout: float = Field( + default=3.0, description="Number of seconds to try connecting to a PV" + ) + parallel: bool = Field( + default=False, + description="Flog indicating whether all variables should be set at once or in series.", + ) + read_only: bool = Field( + default=False, + description="Flag to indicate whether the interface should allow values to be set or not.", + ) def get_default_params(self) -> dict: return { - "context": "pva", + "context": self.context_str, "read-only": self.read_only, } def get_value(self, channel: str): - context = Context("pva") + context = Context(self.context_str) try: return context.get(channel).real except TimeoutError as e: @@ -69,7 +77,9 @@ def get_value(self, channel: str): context.close() def get_values(self, channels: List[str]) -> Dict[str, float]: - context = Context("pva") + if isinstance(channels, str): + channels = [channels] + context = Context(self.context_str) try: # using real allows us to quickly extract the number from both # int/floats as well as enum types @@ -94,7 +104,7 @@ def _put(self, context, channel, value): def set_value(self, channel: str, value, validation_function=None): if not self.read_only: # for parallel to work, context has to be made and closed within the function - context = Context("pva") + context = Context(self.context_str) # always put the value to the set PV try: self._put(context, channel, value) @@ -122,18 +132,18 @@ def set_value(self, channel: str, value, validation_function=None): ) @timeit - def set_values(self, channels, values, configs: Dict[str, dict]): + def set_values(self, channel_inputs: Dict[str, Any], configs: Dict[str, partial]): if not self.read_only: if self.parallel: Parallel(n_jobs=mp.cpu_count())( delayed(self.set_value)(channel, value, configs.get(channel)) - for channel, value in zip(channels, values) + for channel, value in channel_inputs.items() ) else: - for channel, value in zip(channels, values): + for channel, value in channel_inputs.items(): self.set_value(channel, value, configs.get(channel)) else: logging.info( - f"Interface is set to read-only mode, cannot set values {values} to {channels}" + f"Interface is set to read-only mode, cannot set {channel_inputs}" ) From 5a5242de8f74a9bc788dd501f1cbfb382b6d1e8c Mon Sep 17 00:00:00 2001 From: Kathryn Baker Date: Fri, 12 Apr 2024 11:51:37 +0100 Subject: [PATCH 11/15] [CHANGE] updated tests for badger-opt v1.0 (pydantic v2) --- interfaces/epics_pva/__init__.py | 4 +- interfaces/epics_pva/tests.py | 145 ++++++++++++++++++------------- 2 files changed, 87 insertions(+), 62 deletions(-) diff --git a/interfaces/epics_pva/__init__.py b/interfaces/epics_pva/__init__.py index 7119a72..c8749fd 100644 --- a/interfaces/epics_pva/__init__.py +++ b/interfaces/epics_pva/__init__.py @@ -17,7 +17,7 @@ def wrapper_timeit(*args, **kwargs): start_time = time.time() func(*args, **kwargs) end_time = time.time() - logging.info(f"total set time: {end_time - start_time:.5f}s") + logging.debug(f"total set time: {end_time - start_time:.5f}s") return wrapper_timeit @@ -97,9 +97,9 @@ def get_values(self, channels: List[str]) -> Dict[str, float]: def _put(self, context, channel, value): try: context.put(channel, value, timeout=self.timeout) + logging.debug(f"put value {value} to {channel}") except TypeError: context.put(channel, value.item(), timeout=self.timeout) - logging.debug(f"put value {value} to {channel}") def set_value(self, channel: str, value, validation_function=None): if not self.read_only: diff --git a/interfaces/epics_pva/tests.py b/interfaces/epics_pva/tests.py index 4461065..94ea8e8 100644 --- a/interfaces/epics_pva/tests.py +++ b/interfaces/epics_pva/tests.py @@ -28,6 +28,10 @@ class p4pValue: def __init__(self, value) -> None: self.raw = MockRaw(value) + @property + def real(self): + return self.raw.value + @pytest.fixture def mock_context(): @@ -41,7 +45,7 @@ def test_epics_get_values(self, mock_context): p4pValue(values[0]), p4pValue(values[1]), ] - interface = Interface(poll_period=0.1, timeout=1, parallel=False) + interface = Interface(timeout=1, parallel=False, read_only=False) result = interface.get_values(channels) assert result == { "test::channel:1": values[0], @@ -56,7 +60,7 @@ def test_epics_get_values_timeout(self, caplog, mock_context): TimeoutError("individual timeout"), p4pValue(values[1]), ] - interface = Interface(poll_period=0.1, timeout=1, parallel=False) + interface = Interface(timeout=1, parallel=False, read_only=False) result = interface.get_values(channels) # first check the returned values @@ -82,8 +86,8 @@ def test_epics_get_values_timeout(self, caplog, mock_context): ) def test_epics_set_values_no_validation(self, mock_context, test_input): mock_context.return_value.put = MagicMock() - interface = Interface(poll_period=0.1, timeout=1, parallel=False) - interface.set_values(channels, test_input, configs={}) + interface = Interface(timeout=1, parallel=False, read_only=False) + interface.set_values(dict(zip(channels, test_input)), configs={}) assert mock_context.return_value.put.call_args_list[0][0] == ( channels[0], values[0], @@ -96,25 +100,13 @@ def test_epics_set_values_no_validation(self, mock_context, test_input): def test_epics_set_values_with_validation(self, mock_context): mock_context.return_value.put = MagicMock() - # we only configure one of the values to have validation so we only need to - # set values of the mock here - mock_context.return_value.get = MagicMock( - side_effect=[ - p4pValue(7), - p4pValue(values[0] - 0.2 + np.random.uniform(-0.05, 0.05)), - ] - ) + configs = { - "test::channel:1": { - "validate_readback": True, - "readback_pv": "test::channel:1:read", - "tolerance": 0.1, - "count_down": 0.11, - "offset": 0.2, - } + # in reality this would be a partial function instead of a MagicMock + "test::channel:1": MagicMock() } - interface = Interface(poll_period=0.1, timeout=1, parallel=False) - interface.set_values(channels, values, configs=configs) + interface = Interface(timeout=1, parallel=False, read_only=False) + interface.set_values(dict(zip(channels, values)), configs=configs) # check that the correct put values were called assert mock_context.return_value.put.call_args_list[0][0] == ( channels[0], @@ -124,71 +116,104 @@ def test_epics_set_values_with_validation(self, mock_context): channels[1], values[1], ) - # then check that the correct readback PV was used were called - for mock_call in mock_context.return_value.get.call_args_list: - assert mock_call[0][0] == "test::channel:1:read" + # check the validation function was called as expected + configs["test::channel:1"].assert_called_once_with( + set_pv="test::channel:1", + set_value=5.0, + context=mock_context.return_value, + timeout=1, + ) + assert len(configs["test::channel:1"].call_args_list) == 1 + # check that the context is closed correctly() assert len(mock_context.return_value.close.call_args_list) == len(channels) - def test_epics_set_values_validation_timeout(self, caplog, mock_context): + @pytest.mark.parametrize( + "error", + [(ValueError("validation function failed")), (TimeoutError("timeout error"))], + ) + def test_epics_set_values_with_validation_fails(self, caplog, mock_context, error): mock_context.return_value.put = MagicMock() - # we only configure one of the values to have validation so we only need to - # set values of the mock here - mock_context.return_value.get = MagicMock( - # in this case, neither value satisfies the conditions we've placed - side_effect=[p4pValue(7), p4pValue(7.01)] - ) + configs = { - "test::channel:1": { - "validate_readback": True, - "readback_pv": "test::channel:1:read", - "tolerance": 0.1, - "count_down": 0.11, - "offset": 0.2, - } + # in reality this would be a partial function instead of a MagicMock + "test::channel:1": MagicMock(side_effect=error) } - interface = Interface(poll_period=0.1, timeout=1, parallel=False) - interface.set_values(channels, values, configs=configs) + interface = Interface(timeout=1, parallel=False, read_only=False) + interface.set_values(dict(zip(channels, values)), configs=configs) - # check that the correct put values were called - assert mock_context.return_value.put.call_args_list[0][0] == ( - channels[0], - values[0], - ) - assert mock_context.return_value.put.call_args_list[1][0] == ( - channels[1], - values[1], + assert len(caplog.records) == 1 + assert caplog.records[0].getMessage() == str(error) + + @patch("time.sleep") + def test_epics_set_values_put_timeout_retry_fails( + self, mock_time, caplog, mock_context + ): + # here we test whether the put is called twice + mock_context.return_value.put = MagicMock( + side_effect=TimeoutError("timeout error") ) - # check the error is logged correctly + + interface = Interface(timeout=1, parallel=False, read_only=False) + + interface.set_values({channels[0]: values[0]}, configs={}) + + # check that the put was attempted twice + assert len(mock_context.return_value.put.call_args_list) == 2 + # make sure we waited in between + mock_time.assert_called_once_with(1) + + # finally we log a message if both attempts were unsuccessful assert len(caplog.records) == 1 assert ( caplog.records[0].getMessage() - == f"PV {channels[0]} (current: 7.01) cannot reach expected value ({float(values[0])}) in designated time 0.11!" + == "Timeout on test::channel:1: timeout error after 2 attempts and a 1 second delay" + ) + + @patch("time.sleep") + def test_epics_set_values_put_timeout_retry_successful( + self, mock_time, caplog, mock_context + ): + mock_context.return_value.put = MagicMock( + side_effect=[TimeoutError("timeout error"), None] ) - assert len(mock_context.return_value.close.call_args_list) == len(channels) + + interface = Interface(timeout=1, parallel=False, read_only=False) + + with caplog.at_level(logging.DEBUG): + interface.set_values({channels[0]: values[0]}, configs={}) + + # check that the put was attempted twice + assert len(mock_context.return_value.put.call_args_list) == 2 + # make sure we waited in between + mock_time.assert_called_once_with(1) + + # finally we log a message if both attempts were unsuccessful and we should + # also get a message about how long it takes to set the values + assert len(caplog.records) == 2 + assert caplog.records[0].getMessage() == "put value 5.0 to test::channel:1" def test_epics_set_values_read_only(self, caplog, mock_context): caplog.set_level(logging.INFO) mock_context.return_value.put = MagicMock() - interface = Interface( - poll_period=0.1, timeout=1, parallel=False, read_only=True - ) - interface.set_values(channels, values, configs={}) + interface = Interface(timeout=1, parallel=False, read_only=True) + set_vals = dict(zip(channels, values)) + with caplog.at_level(logging.INFO): + interface.set_values(set_vals, configs={}) # check the log message is sent correctly and that context.put is not called mock_context.return_value.put.assert_not_called() assert len(caplog.records) == 1 assert ( caplog.records[0].getMessage() - == f"Interface is set to read-only mode, cannot set values {values} to {channels}" + == f"Interface is set to read-only mode, cannot set {set_vals}" ) def test_epics_set_value_read_only(self, caplog, mock_context): caplog.set_level(logging.INFO) mock_context.return_value.put = MagicMock() - interface = Interface( - poll_period=0.1, timeout=1, parallel=False, read_only=True - ) - interface.set_value(channels[0], values[0], set_config={}) + interface = Interface(timeout=1, parallel=False, read_only=True) + with caplog.at_level(logging.INFO): + interface.set_value(channels[0], values[0]) # check the log message is sent correctly and that context.put is not called mock_context.return_value.put.assert_not_called() From debbf27078e45a2334a793e25bca546608abb435 Mon Sep 17 00:00:00 2001 From: Kathryn Baker Date: Fri, 12 Apr 2024 11:58:39 +0100 Subject: [PATCH 12/15] [FIX] typo in Field description --- interfaces/epics_pva/__init__.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/interfaces/epics_pva/__init__.py b/interfaces/epics_pva/__init__.py index c8749fd..fd141b9 100644 --- a/interfaces/epics_pva/__init__.py +++ b/interfaces/epics_pva/__init__.py @@ -52,7 +52,7 @@ class Interface(interface.Interface): ) parallel: bool = Field( default=False, - description="Flog indicating whether all variables should be set at once or in series.", + description="Flag indicating whether all variables should be set at once or in series.", ) read_only: bool = Field( default=False, From fefc707eed8f52a17c9663a7798e3a578f01d49e Mon Sep 17 00:00:00 2001 From: Kathryn Baker Date: Fri, 12 Apr 2024 12:31:22 +0100 Subject: [PATCH 13/15] update README --- interfaces/epics_pva/README.md | 66 +++++++++++++++++++++++++++++++++- 1 file changed, 65 insertions(+), 1 deletion(-) diff --git a/interfaces/epics_pva/README.md b/interfaces/epics_pva/README.md index a55e3e4..7ae17ce 100644 --- a/interfaces/epics_pva/README.md +++ b/interfaces/epics_pva/README.md @@ -1,5 +1,69 @@ -# EPICS Interface for Badger +# EPICS PVAccess Interface for Badger +Interface for setting and reading EPICS PVAccess PV values utilising the [p4p](https://github.com/mdavidsaver/p4p) library. ## Prerequisites ## Usage +Some general notes on usage: + +### Read Only Mode +In some cases we may want to be testing our software stack on a live system but not want to be changing values. For these occasions we utilise a `read_only` flag configured in the constructor. Instead of actually setting values in the control system, the `set_values()` call will log a message to indicate the values that would have been set. `get_values()` will still return valid observations. + +```python +interface = Interface(read_only=True) +``` + +### Retry on Timeout +In the case of a TimeoutError on a PV when doing a `put` call, we retry the call twice, waiting one second between. If this second attempt to set the PV fails, an exception is thrown. + +### Validation Functions +In the case of some PVs we may want to validate that the PV has in fact been set, for example by checking an associated readback PV. For these cases, we pass a `validation_function` to the `set_value()` call. It is expected that this function would be configured as a `functools.partial` function and multiple validation functions can be passed to the `set_values()` call as a dictionary with the key as the set PV name. An example is provided below: + +Validation function: +```python +def wait_for_readback( + set_pv: str, set_value: float, readback_pv: str, tolerance: float, count_down: int, offset: float, context: p4p.client.thread.Context, timeout: int +): + time_limit = deepcopy(count_down) + while count_down > 0: + _value = context.get(readback_pv, timeout=timeout) + if np.isclose(_value.real, set_value + offset,atol=tolerance): + return _value.real + else: + time.sleep(0.1) + count_down -= 0.1 + raise ValueError( + f"PV {set_pv} (current {readback_pv}: {_value.real}) cannot reach expected value ({set_value + offset}) in designated time {time_limit}!" + ) +``` + +Expected usage: +```python +class Environment(badger.Environment): + ... + _configs: ClassVar[Dict[str, partial]] = { + "EXAMPLE:PV:SET": partial( + wait_for_readback, + readback_pv="EXAMPLE:PV:READ", + tolerance=0.05, + count_down=15, + offset=0, + ), + } + def set_variables(self, variable_inputs): + self.interface.set_values( + variable_inputs, + configs=self._configs, + ) +``` + + +### Parallel Execution +When using these validation function, setting values on multiple PVs may cause long delays, espcially if the update rate of the READ PV is slow. Therefore we provide the option to run all of the `set` commands in parallel. This can be configured using the `parallel` flag in the constructor: +```python +interface = Interface(parallel=False) +``` + + +## Testing +Initial tests for the interface are available in the `tests.py` file and can be run locally using `pytest tests.py` or `pytest --cov=. tests.py` from within the `epics_pva` directory. \ No newline at end of file From e1a2842504211ea04ce4220f48406d36ab276b54 Mon Sep 17 00:00:00 2001 From: Kathryn Baker Date: Mon, 9 Jun 2025 13:48:03 +0100 Subject: [PATCH 14/15] updated tests for new, cleaner handling of TImeoutErrors --- interfaces/epics_pva/__init__.py | 155 +++++++++++++++++-------------- interfaces/epics_pva/tests.py | 107 +++++++++++++-------- 2 files changed, 156 insertions(+), 106 deletions(-) diff --git a/interfaces/epics_pva/__init__.py b/interfaces/epics_pva/__init__.py index fd141b9..1cd32fe 100644 --- a/interfaces/epics_pva/__init__.py +++ b/interfaces/epics_pva/__init__.py @@ -8,7 +8,7 @@ import numpy as np from badger import interface from joblib import Parallel, delayed -from p4p.client.thread import Context +from p4p.client.thread import Cancelled, Context, Disconnected, RemoteError, TimeoutError from pydantic import Field @@ -22,34 +22,12 @@ def wrapper_timeit(*args, **kwargs): return wrapper_timeit -def retry_on_timeout(func): - def wrapper_retry(*args, **kwargs): - channel = args[2] - try: - func(*args, **kwargs) - except TimeoutError: - # we try again! - # give it some time and see if it fixes itself - time.sleep(1) - try: - func(*args, **kwargs) - except TimeoutError as e: - # TODO - decide whether we should re-raise or not - raise TimeoutError( - f"Timeout on {channel}: {e} after 2 attempts and a 1 second delay" - ) - - return wrapper_retry - - class Interface(interface.Interface): """Concrete interface for interacting with EPICS PVAccess PVs""" name: str = "epics_pva" context_str: str = "pva" - timeout: float = Field( - default=3.0, description="Number of seconds to try connecting to a PV" - ) + timeout: float = Field(default=3.0, description="Number of seconds to try connecting to a PV") parallel: bool = Field( default=False, description="Flag indicating whether all variables should be set at once or in series.", @@ -67,63 +45,104 @@ def get_default_params(self) -> dict: def get_value(self, channel: str): context = Context(self.context_str) - try: - return context.get(channel).real - except TimeoutError as e: - # TODO - decide whether we should return a NaN value here - logging.exception(f"{channel}: {e}") - return np.nan - finally: - context.close() + + result = context.get(channel, throw=False) + if isinstance(result, (Disconnected, TimeoutError, RemoteError, Cancelled)): + logging.warning("Could not retrieve value of %s due to %s", channel, type(result).__name__) + result = np.nan + else: + result = result.real + context.close() + return result def get_values(self, channels: List[str]) -> Dict[str, float]: if isinstance(channels, str): channels = [channels] context = Context(self.context_str) - try: - # using real allows us to quickly extract the number from both - # int/floats as well as enum types - values = [value.real for value in context.get(channels)] - except TimeoutError: - logging.exception(f"Timeout error from {channels}, retrying individually") - # if we get a timeout error one even one of them, we retry to get - # the individual values - values = [self.get_value(channel) for channel in channels] - finally: - context.close() - return dict(zip(channels, values)) + book = {} + + results = context.get(channels, throw=False) + failures = [] + + for i, result in enumerate(results): + if isinstance(result, (Disconnected, TimeoutError, RemoteError, Cancelled)): + failures.append(channels[i]) + logging.warning( + "Could not retrieve value of %s due to %s. Retrying individually.", + channels[i], + type(result).__name__, + ) + else: + book[channels[i]] = result.real + + # then if there are any failures, retry the failed ones in the same way + if len(failures) >= 1: + results = context.get(failures, throw=False) + + for i, result in enumerate(results): + if isinstance(result, (Disconnected, TimeoutError, RemoteError, Cancelled)): + logging.warning("Could not retrieve value of %s due to %s.", failures[i], type(result).__name__) + book[failures[i]] = np.nan + else: + book[failures[i]] = result.real + context.close() + return book - @retry_on_timeout def _put(self, context, channel, value): + success = False try: - context.put(channel, value, timeout=self.timeout) - logging.debug(f"put value {value} to {channel}") + result = context.put(channel, value, timeout=self.timeout, throw=False) except TypeError: - context.put(channel, value.item(), timeout=self.timeout) + value = value.item() + result = context.put(channel, value, timeout=self.timeout, throw=False) + if result is None: + logging.debug( + "Put value %d to %s", + value, + channel, + ) + success = True + else: + logging.info("Could not put value %d to %s due to %s. Retrying...", value, channel, type(result).__name__) + # retry after a short sleep + time.sleep(1) + result = context.put(channel, value, timeout=self.timeout, throw=False) + if result is None: + logging.debug( + "Put value %d to %s", + value, + channel, + ) + success = True + else: + logging.warning( + "Could not put value %d to %s (second attempt) due to %s after a 1 second delay.", + value, + channel, + type(result).__name__, + ) + return success def set_value(self, channel: str, value, validation_function=None): if not self.read_only: # for parallel to work, context has to be made and closed within the function context = Context(self.context_str) # always put the value to the set PV - try: - self._put(context, channel, value) - if validation_function is not None: - try: - validation_function( - set_pv=channel, - set_value=value, - context=context, - timeout=self.timeout, - ) - except ValueError as e: - logging.warning(e) - # context.close() - # return value - except TimeoutError as e: - logging.warning(e) - finally: - context.close() + + success = self._put(context, channel, value) + # if we weren't successful in setting the PV, there's no need to validate + # against the readback + if success and validation_function is not None: + try: + validation_function( + set_pv=channel, + set_value=value, + context=context, + timeout=self.timeout, + ) + except (ValueError, TimeoutError) as e: + logging.warning(e) + context.close() else: logging.info( "Interface is set to read-only mode, cannot set value %s to %s", @@ -144,6 +163,4 @@ def set_values(self, channel_inputs: Dict[str, Any], configs: Dict[str, partial] for channel, value in channel_inputs.items(): self.set_value(channel, value, configs.get(channel)) else: - logging.info( - f"Interface is set to read-only mode, cannot set {channel_inputs}" - ) + logging.info("Interface is set to read-only mode, cannot set %s", channel_inputs) diff --git a/interfaces/epics_pva/tests.py b/interfaces/epics_pva/tests.py index 94ea8e8..c582752 100644 --- a/interfaces/epics_pva/tests.py +++ b/interfaces/epics_pva/tests.py @@ -53,29 +53,51 @@ def test_epics_get_values(self, mock_context): } mock_context.return_value.close.assert_called_once_with() - def test_epics_get_values_timeout(self, caplog, mock_context): + def test_epics_get_values_timeout_retry(self, caplog, mock_context): # we get a timeout error first, then we get the individual values mock_context.return_value.get.side_effect = [ - TimeoutError("original error"), - TimeoutError("individual timeout"), - p4pValue(values[1]), + [ + TimeoutError("original error"), + p4pValue(values[1]), + ], + [p4pValue(values[0])], + ] + interface = Interface(timeout=1, parallel=False, read_only=False) + result = interface.get_values(channels) + + # check the mock calls + assert len(mock_context.return_value.get.call_args_list) == 2 + + # first check the returned values + assert result["test::channel:1"] == values[0] + assert result["test::channel:2"] == values[1] + + # then check the logs + assert len(caplog.records) == 1 + assert len(mock_context.return_value.close.call_args_list) == 1 + + def test_epics_get_values_timeout_retry_fails(self, caplog, mock_context): + # we get a timeout error first, then we get the individual values + mock_context.return_value.get.side_effect = [ + [ + TimeoutError("original error"), + p4pValue(values[1]), + ], + [TimeoutError("retry")], ] interface = Interface(timeout=1, parallel=False, read_only=False) result = interface.get_values(channels) + # check the mock calls + assert len(mock_context.return_value.get.call_args_list) == 2 + # first check the returned values assert np.isnan(result["test::channel:1"]) assert result["test::channel:2"] == values[1] # then check the logs assert len(caplog.records) == 2 - assert ( - caplog.records[0].getMessage() - == f"Timeout error from {channels}, retrying individually" - ) - assert caplog.records[1].getMessage() == f"{channels[0]}: individual timeout" - - assert len(mock_context.return_value.close.call_args_list) == len(channels) + 1 + assert len(mock_context.return_value.close.call_args_list) == 1 @pytest.mark.parametrize( "test_input", @@ -85,9 +107,10 @@ def test_epics_get_values_timeout(self, caplog, mock_context): ], ) def test_epics_set_values_no_validation(self, mock_context, test_input): - mock_context.return_value.put = MagicMock() + mock_context.return_value.put = MagicMock(return_value=None) interface = Interface(timeout=1, parallel=False, read_only=False) interface.set_values(dict(zip(channels, test_input)), configs={}) + assert mock_context.return_value.put.call_args_list[0][0] == ( channels[0], values[0], @@ -98,8 +121,30 @@ def test_epics_set_values_no_validation(self, mock_context, test_input): ) assert len(mock_context.return_value.close.call_args_list) == len(channels) + def test_epics_set_values_no_validation_timeout(self, caplog, mock_context): + caplog.set_level(logging.INFO) + mock_context.return_value.put = MagicMock(return_value=TimeoutError("timeout")) + interface = Interface(timeout=1, parallel=False, read_only=False) + interface.set_values(dict(zip(channels, [value for value in values])), configs={}) + + # here we check that the retry works + assert len(mock_context.return_value.put.call_args_list) == 4 + assert mock_context.return_value.put.call_args_list[0][0] == ( + channels[0], + values[0], + ) + assert mock_context.return_value.put.call_args_list[1][0] == ( + channels[0], + values[0], + ) + assert len(mock_context.return_value.close.call_args_list) == len(channels) + + assert len(caplog.records) == 4 + assert str(caplog.records[0].getMessage()).endswith("Retrying...") + assert str(caplog.records[-1].getMessage()).endswith("after a 1 second delay.") + def test_epics_set_values_with_validation(self, mock_context): - mock_context.return_value.put = MagicMock() + mock_context.return_value.put = MagicMock(return_value=None) configs = { # in reality this would be a partial function instead of a MagicMock @@ -131,8 +176,8 @@ def test_epics_set_values_with_validation(self, mock_context): "error", [(ValueError("validation function failed")), (TimeoutError("timeout error"))], ) - def test_epics_set_values_with_validation_fails(self, caplog, mock_context, error): - mock_context.return_value.put = MagicMock() + def test_epics_set_values_with_validation_errors(self, caplog, mock_context, error): + mock_context.return_value.put = MagicMock(return_value=None) configs = { # in reality this would be a partial function instead of a MagicMock @@ -145,13 +190,10 @@ def test_epics_set_values_with_validation_fails(self, caplog, mock_context, erro assert caplog.records[0].getMessage() == str(error) @patch("time.sleep") - def test_epics_set_values_put_timeout_retry_fails( - self, mock_time, caplog, mock_context - ): + def test_epics_set_values_put_timeout_retry_fails(self, mock_time, caplog, mock_context): # here we test whether the put is called twice - mock_context.return_value.put = MagicMock( - side_effect=TimeoutError("timeout error") - ) + caplog.set_level = logging.INFO + mock_context.return_value.put = MagicMock(return_value=TimeoutError("timeout error")) interface = Interface(timeout=1, parallel=False, read_only=False) @@ -164,18 +206,12 @@ def test_epics_set_values_put_timeout_retry_fails( # finally we log a message if both attempts were unsuccessful assert len(caplog.records) == 1 - assert ( - caplog.records[0].getMessage() - == "Timeout on test::channel:1: timeout error after 2 attempts and a 1 second delay" - ) + assert str(caplog.records[0].getMessage()).endswith("after a 1 second delay.") + @pytest.mark.skip(reason="Unsure how to test when side_effect will raise TimeoutError instead of return") @patch("time.sleep") - def test_epics_set_values_put_timeout_retry_successful( - self, mock_time, caplog, mock_context - ): - mock_context.return_value.put = MagicMock( - side_effect=[TimeoutError("timeout error"), None] - ) + def test_epics_set_values_put_timeout_retry_successful(self, mock_time, caplog, mock_context): + mock_context.return_value.put = MagicMock(side_effect=[TimeoutError("timeout error"), None]) interface = Interface(timeout=1, parallel=False, read_only=False) @@ -189,8 +225,8 @@ def test_epics_set_values_put_timeout_retry_successful( # finally we log a message if both attempts were unsuccessful and we should # also get a message about how long it takes to set the values - assert len(caplog.records) == 2 - assert caplog.records[0].getMessage() == "put value 5.0 to test::channel:1" + assert len(caplog.records) == 1 + assert caplog.records[0].getMessage() == "Retrying..." def test_epics_set_values_read_only(self, caplog, mock_context): caplog.set_level(logging.INFO) @@ -203,10 +239,7 @@ def test_epics_set_values_read_only(self, caplog, mock_context): # check the log message is sent correctly and that context.put is not called mock_context.return_value.put.assert_not_called() assert len(caplog.records) == 1 - assert ( - caplog.records[0].getMessage() - == f"Interface is set to read-only mode, cannot set {set_vals}" - ) + assert caplog.records[0].getMessage() == f"Interface is set to read-only mode, cannot set {set_vals}" def test_epics_set_value_read_only(self, caplog, mock_context): caplog.set_level(logging.INFO) From 0f6210dbee1e64cbd80a525f1f5affb0995136e2 Mon Sep 17 00:00:00 2001 From: Kathryn Baker Date: Mon, 9 Jun 2025 16:17:51 +0100 Subject: [PATCH 15/15] added new mock interface --- interfaces/mock/README.md | 5 +++++ interfaces/mock/__init__.py | 26 ++++++++++++++++++++++++++ interfaces/mock/configs.yaml | 6 ++++++ 3 files changed, 37 insertions(+) create mode 100644 interfaces/mock/README.md create mode 100644 interfaces/mock/__init__.py create mode 100644 interfaces/mock/configs.yaml diff --git a/interfaces/mock/README.md b/interfaces/mock/README.md new file mode 100644 index 0000000..bde9fc4 --- /dev/null +++ b/interfaces/mock/README.md @@ -0,0 +1,5 @@ +# Mock Interface for Badger + +## Prerequisites + +## Usage diff --git a/interfaces/mock/__init__.py b/interfaces/mock/__init__.py new file mode 100644 index 0000000..c145ade --- /dev/null +++ b/interfaces/mock/__init__.py @@ -0,0 +1,26 @@ +from badger import interface + +from numpy import random + + +class Interface(interface.Interface): + name = "mock" + + @staticmethod + def get_default_params(): + return None + + def get_value(self, channel: str): + print("Called get_value for channel: {}.".format(channel)) + return random.random() + + def set_value(self, channel: str, value): + print("Called set_value for channel: {}, with value: {}".format(channel, value)) + + def get_values(self, channels: list): + return {channel: self.get_value(channel) for channel in channels} + + def set_values(self, channel_inputs: dict, *args, **kwargs): + print(f"Called set_values with args:{args}, kwargs:{kwargs}") + for channel, val in channel_inputs.items(): + self.set_value(channel, val) diff --git a/interfaces/mock/configs.yaml b/interfaces/mock/configs.yaml new file mode 100644 index 0000000..3242482 --- /dev/null +++ b/interfaces/mock/configs.yaml @@ -0,0 +1,6 @@ +--- +name: mock +version: "0.1" +dependencies: + - badger-opt + - numpy