diff --git a/.gitignore b/.gitignore index ed7e36d..f8a3fdf 100644 --- a/.gitignore +++ b/.gitignore @@ -16,3 +16,7 @@ relPaths.sh /doc/ *_info_positions.req *_info_settings.req +__pycache__/ +*.py[cod] +*$py.class +test-reports/ diff --git a/Makefile b/Makefile index bab6eff..9e4fe0e 100644 --- a/Makefile +++ b/Makefile @@ -12,3 +12,6 @@ DIRS += $(wildcard *[Aa]pp) DIRS += $(wildcard ioc[Bb]oot) include $(TOP)/configure/RULES_TOP + +ioctests: + .\system_tests\run_tests.bat diff --git a/system_tests/__init__.py b/system_tests/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/system_tests/lewis_emulators/__init__.py b/system_tests/lewis_emulators/__init__.py new file mode 100644 index 0000000..ad92558 --- /dev/null +++ b/system_tests/lewis_emulators/__init__.py @@ -0,0 +1,2 @@ +# DO NOT DELETE THIS FILE - LEWIS FRAMEWORK REQUIRES THE DIRECTORY TO BE IMPORTABLE +from __future__ import absolute_import diff --git a/system_tests/lewis_emulators/lewis_versions.py b/system_tests/lewis_emulators/lewis_versions.py new file mode 100644 index 0000000..0f6c45a --- /dev/null +++ b/system_tests/lewis_emulators/lewis_versions.py @@ -0,0 +1,2 @@ +LEWIS_1_3_0 = "1.3.0" +LEWIS_LATEST = LEWIS_1_3_0 diff --git a/system_tests/lewis_emulators/triton/__init__.py b/system_tests/lewis_emulators/triton/__init__.py new file mode 100644 index 0000000..c68adeb --- /dev/null +++ b/system_tests/lewis_emulators/triton/__init__.py @@ -0,0 +1,5 @@ +from .device import SimulatedTriton +from ..lewis_versions import LEWIS_LATEST + +framework_version = LEWIS_LATEST +__all__ = ['SimulatedTriton'] diff --git a/system_tests/lewis_emulators/triton/device.py b/system_tests/lewis_emulators/triton/device.py new file mode 100644 index 0000000..016e2a3 --- /dev/null +++ b/system_tests/lewis_emulators/triton/device.py @@ -0,0 +1,153 @@ +from collections import OrderedDict +from .states import DefaultState +from lewis.devices import StateMachineDevice + + +HEATER_NAME = "H1" + + +class TemperatureStage(object): + """ + Class representing a temperature stage. + """ + def __init__(self, name): + self.name = name + self.temperature = 1 + self.enabled = True + + self.resistance = 0 + + self.excitation_type = "VOLT" + self.excitation = 10 + + self.pause = 10 + self.dwell = 3 + + +class PressureSensor(object): + """ + Class to represent a pressure sensor. + + Having this as a class makes it more extensible in future, as the triton driver is still in flux. + """ + def __init__(self): + self.pressure = 0 + + +class SimulatedTriton(StateMachineDevice): + + def _initialize_data(self): + """ + Initialize all of the device's attributes. + """ + self.heater_range = 0 + self.heater_power = 1 + self.heater_resistance = 0 + + self.temperature_setpoint = 0 + self.p = 0 + self.i = 0 + self.d = 0 + self.closed_loop = False + + self.status = "This is a device status message." + self.automation = "This is the automation status" + + self.pressure_sensors = {"P{}".format(i): PressureSensor() for i in range(1, 6)} + + self.temperature_stages = { + "T1": TemperatureStage("STIL"), + "T2": TemperatureStage("PT1"), + "T3": TemperatureStage("PT2"), + "T4": TemperatureStage("SORB"), + "T5": TemperatureStage("MC"), + "T6": TemperatureStage("unknown"), + } + + self.sample_channel = "T5" + assert self.sample_channel in self.temperature_stages + + def find_temperature_channel(self, name): + + for k, v in self.temperature_stages.items(): + if v.name == name: + return k + else: + raise KeyError("{} not found".format(name)) + + def set_temperature_backdoor(self, stage_name, new_temp): + self.temperature_stages[self.find_temperature_channel(stage_name)].temperature = new_temp + + def set_pressure_backdoor(self, sensor, newpressure): + self.pressure_sensors["P{}".format(sensor)].pressure = float(newpressure) + + def set_sensor_property_backdoor(self, sensor, property, value): + # In older versions of the software, there was an off-by-one error here in the OI software. + # This has now been fixed by O.I. so no longer need to adjust by one here. + setattr(self.temperature_stages["T{}".format(sensor)], property, value) + + def _get_state_handlers(self): + return {'default': DefaultState()} + + def _get_initial_state(self): + return 'default' + + def _get_transition_handlers(self): + return OrderedDict([]) + + def get_closed_loop_mode(self): + return self.closed_loop + + def set_closed_loop_mode(self, mode): + self.closed_loop = mode + + def set_p(self, value): + self.p = value + + def set_i(self, value): + self.i = value + + def set_d(self, value): + self.d = value + + def get_p(self): + return self.p + + def get_i(self): + return self.i + + def get_d(self): + return self.d + + def get_temperature_setpoint(self): + return self.temperature_setpoint + + def set_temperature_setpoint(self, value): + self.temperature_setpoint = value + + def get_heater_range(self): + return self.heater_range + + def set_heater_range(self, value): + self.heater_range = value + + def is_channel_enabled(self, chan): + try: + return self.temperature_stages[chan].enabled + except KeyError: + return False + + def set_channel_enabled(self, chan, newstate): + self.temperature_stages[chan].enabled = newstate + + def get_status(self): + return self.status + + def get_automation(self): + return self.automation + + def get_pressure(self, sensor): + return self.pressure_sensors[sensor].pressure + + def get_temp(self, stage): + return self.temperature_stages[stage].temperature diff --git a/system_tests/lewis_emulators/triton/interfaces/__init__.py b/system_tests/lewis_emulators/triton/interfaces/__init__.py new file mode 100644 index 0000000..605c573 --- /dev/null +++ b/system_tests/lewis_emulators/triton/interfaces/__init__.py @@ -0,0 +1,3 @@ +from .stream_interface import TritonStreamInterface + +__all__ = ['TritonStreamInterface'] diff --git a/system_tests/lewis_emulators/triton/interfaces/stream_interface.py b/system_tests/lewis_emulators/triton/interfaces/stream_interface.py new file mode 100644 index 0000000..2bf2ded --- /dev/null +++ b/system_tests/lewis_emulators/triton/interfaces/stream_interface.py @@ -0,0 +1,218 @@ +from datetime import datetime + +from lewis.adapters.stream import StreamInterface +from lewis.core.logging import has_log +from lewis.utils.command_builder import CmdBuilder +from lewis_emulators.triton.device import HEATER_NAME + + +@has_log +class TritonStreamInterface(StreamInterface): + + # Commands that we expect via serial during normal operation + commands = { + # ID + CmdBuilder("get_idn").escape("*IDN?").eos().build(), + + # UIDs + CmdBuilder("get_uid").escape("READ:SYS:DR:CHAN:").arg("[A-Z0-9]+").eos().build(), + + # PID setpoints + CmdBuilder("set_p").escape("SET:DEV:").arg("T[0-9]+").escape(":TEMP:LOOP:P:").float().eos().build(), + CmdBuilder("set_i").escape("SET:DEV:").arg("T[0-9]+").escape(":TEMP:LOOP:I:").float().eos().build(), + CmdBuilder("set_d").escape("SET:DEV:").arg("T[0-9]+").escape(":TEMP:LOOP:D:").float().eos().build(), + + # PID readbacks + CmdBuilder("get_p").escape("READ:DEV:").arg("T[0-9]+").escape(":TEMP:LOOP:P").eos().build(), + CmdBuilder("get_i").escape("READ:DEV:").arg("T[0-9]+").escape(":TEMP:LOOP:I").eos().build(), + CmdBuilder("get_d").escape("READ:DEV:").arg("T[0-9]+").escape(":TEMP:LOOP:D").eos().build(), + + # Setpoint temperature + CmdBuilder("set_temperature_setpoint").escape("SET:DEV:").arg("T[0-9]+").escape(":TEMP:LOOP:TSET:").float().eos().build(), + CmdBuilder("get_temperature_setpoint").escape("READ:DEV:").arg("T[0-9]+").escape(":TEMP:LOOP:TSET").eos().build(), + + # Temperature + CmdBuilder("get_temp").escape("READ:DEV:").arg("T[0-9]+").escape(":TEMP:SIG:TEMP").eos().build(), + + # Heater range + CmdBuilder("set_heater_range").escape("SET:DEV:").arg("T[0-9]+").escape(":TEMP:LOOP:RANGE:").float().eos().build(), + CmdBuilder("get_heater_range").escape("READ:DEV:").arg("T[0-9]+").escape(":TEMP:LOOP:RANGE").eos().build(), + + # Heater type + CmdBuilder("get_heater_type").escape("READ:DEV:").arg("T[0-9]+").escape(":TEMP:LOOP:HTR").eos().build(), + + # Get heater power + CmdBuilder("get_heater_power").escape("READ:DEV:{}:HTR:SIG:POWR".format(HEATER_NAME)).eos().build(), + + # Get heater resistance + CmdBuilder("get_heater_resistance").escape("READ:DEV:{}:HTR:RES".format(HEATER_NAME)).eos().build(), + + # Heater control sensor + CmdBuilder("get_heater_control_sensor").escape("READ:DEV:{}:HTR:LOOP".format(HEATER_NAME)).eos().build(), + + # Loop mode + CmdBuilder("get_closed_loop_mode").escape("READ:DEV:").arg("T[0-9]+").escape(":TEMP:LOOP:MODE").eos().build(), + CmdBuilder("set_closed_loop_mode").escape("SET:DEV:").arg("T[0-9]+").escape(":TEMP:LOOP:MODE:").any().eos().build(), + + # Channel enablement + CmdBuilder("get_channel_enabled").escape("READ:DEV:").arg("T[0-9]+").escape(":TEMP:MEAS:ENAB").eos().build(), + CmdBuilder("set_channel_enabled").escape("SET:DEV:").arg("T[0-9]+").escape(":TEMP:MEAS:ENAB:").any().eos().build(), + + # Status + CmdBuilder("get_status").escape("READ:SYS:DR:STATUS").eos().build(), + CmdBuilder("get_automation").escape("READ:SYS:DR:ACTN").eos().build(), + + # Pressures + CmdBuilder("get_pressure").escape("READ:DEV:").arg("P[0-9]+").escape(":PRES:SIG:PRES").eos().build(), + + # System + CmdBuilder("get_time").escape("READ:SYS:TIME").eos().build(), + + # Sensor info + CmdBuilder("get_sig").escape("READ:DEV:").arg("T[0-9]+").escape(":TEMP:SIG").eos().build(), + CmdBuilder("get_excitation").escape("READ:DEV:").arg("T[0-9]+").escape(":TEMP:EXCT").eos().build(), + CmdBuilder("get_meas").escape("READ:DEV:").arg("T[0-9]+").escape(":TEMP:MEAS").eos().build(), + } + + in_terminator = "\r\n" + out_terminator = "\n" + + def handle_error(self, request, error): + err_string = "command was: {}, error was: {}: {}\n".format(request, error.__class__.__name__, error) + print(err_string) + self.log.error(err_string) + return err_string + + def raise_if_channel_is_not_sample_channel(self, chan): + if str(chan) != self.device.sample_channel: + raise ValueError("Channel should have been sample channel") + + def get_idn(self): + return "This is the IDN of this device" + + def get_uid(self, chan): + return "STAT:SYS:DR:CHAN:{}:{}".format(chan, self.device.find_temperature_channel(chan)) + + def set_p(self, stage, value): + self.raise_if_channel_is_not_sample_channel(stage) + self.device.set_p(float(value)) + return "ok" + + def set_i(self, stage, value): + self.raise_if_channel_is_not_sample_channel(stage) + self.device.set_i(float(value)) + return "ok" + + def set_d(self, stage, value): + self.raise_if_channel_is_not_sample_channel(stage) + self.device.set_d(float(value)) + return "ok" + + def get_p(self, stage): + self.raise_if_channel_is_not_sample_channel(stage) + return "STAT:DEV:{}:TEMP:LOOP:P:{}".format(stage, self.device.get_p()) + + def get_i(self, stage): + self.raise_if_channel_is_not_sample_channel(stage) + return "STAT:DEV:{}:TEMP:LOOP:I:{}".format(stage, self.device.get_i()) + + def get_d(self, stage): + self.raise_if_channel_is_not_sample_channel(stage) + return "STAT:DEV:{}:TEMP:LOOP:D:{}".format(stage, self.device.get_d()) + + def set_temperature_setpoint(self, chan, value): + self.raise_if_channel_is_not_sample_channel(chan) + self.device.set_temperature_setpoint(float(value)) + return "ok" + + def get_temperature_setpoint(self, chan): + self.raise_if_channel_is_not_sample_channel(chan) + return "STAT:DEV:{}:TEMP:LOOP:TSET:{}K" .format(chan, self.device.get_temperature_setpoint()) + + def set_heater_range(self, chan, value): + self.raise_if_channel_is_not_sample_channel(chan) + self.device.set_heater_range(float(value)) + return "ok" + + def get_heater_range(self, chan): + self.raise_if_channel_is_not_sample_channel(chan) + return "STAT:DEV:{}:TEMP:LOOP:RANGE:{}mA".format(chan, self.device.get_heater_range()) + + def get_heater_type(self, chan): + self.raise_if_channel_is_not_sample_channel(chan) + return "STAT:DEV:{}:TEMP:LOOP:HTR:{}".format(chan, HEATER_NAME) + + def get_heater_power(self): + return "STAT:DEV:{}:HTR:SIG:POWR:{}uW".format(HEATER_NAME, self.device.heater_power) + + def get_heater_resistance(self): + return "STAT:DEV:{}:HTR:RES:{}Ohm".format(HEATER_NAME, self.device.heater_resistance) + + def get_heater_current(self): + return "STAT:DEV:{}:HTR:SIG:CURR:{}mA".format(HEATER_NAME, self.device.heater_current) + + def get_closed_loop_mode(self, chan): + self.raise_if_channel_is_not_sample_channel(chan) + return "STAT:DEV:{}:TEMP:LOOP:MODE:{}".format(chan, "ON" if self.device.get_closed_loop_mode() else "OFF") + + def set_closed_loop_mode(self, chan, mode): + self.raise_if_channel_is_not_sample_channel(chan) + + if mode not in ["ON", "OFF"]: + raise ValueError("Invalid mode") + + self.device.set_closed_loop_mode(mode == "ON") + return "STAT:SET:DEV:{}:TEMP:LOOP:MODE:{}:VALID".format(chan, mode) + + def get_channel_enabled(self, channel): + return "STAT:DEV:{}:TEMP:MEAS:ENAB:{}"\ + .format(channel, "ON" if self.device.is_channel_enabled(channel) else "OFF") + + def set_channel_enabled(self, channel, newstate): + newstate = str(newstate) + + if newstate not in ["ON", "OFF"]: + raise ValueError("New state '{}' not valid.".format(newstate)) + + self.device.set_channel_enabled(channel, newstate == "ON") + return "ok" + + def get_status(self): + return "STAT:SYS:DR:STATUS:{}".format(self.device.get_status()) + + def get_automation(self): + return "STAT:SYS:DR:ACTN:{}".format(self.device.get_automation()) + + def get_temp(self, stage): + return "STAT:DEV:{}:TEMP:SIG:TEMP:{}K".format(stage, self.device.get_temp(str(stage))) + + def get_pressure(self, sensor): + return "STAT:DEV:{}:PRES:SIG:PRES:{}mB".format(sensor, self.device.get_pressure(sensor)) + + def get_time(self): + return datetime.now().strftime("STAT:SYS:TIME:%H:%M:%S") + + def get_heater_control_sensor(self): + # Always assume heater controls sample. This is true so far at ISIS + return "STAT:DEV:{}:HTR:LOOP:SENS:{}".format(HEATER_NAME, self.device.sample_channel) + + def get_sig(self, chan): + return "STAT:DEV:{}:TEMP:SIG:TEMP:{}K:RES:{}Ohm".format( + chan, + self.device.temperature_stages[chan].temperature, + self.device.temperature_stages[chan].resistance, + ) + + def get_excitation(self, chan): + return "STAT:DEV:{}:TEMP:EXCT:TYPE:{}:MAG:{}V".format( + chan, + self.device.temperature_stages[chan].excitation_type, + self.device.temperature_stages[chan].excitation, + ) + + def get_meas(self, chan): + return "STAT:DEV:{}:TEMP:MEAS:PAUS:{}s:DWEL:{}s:ENAB:ON".format( + chan, + self.device.temperature_stages[chan].pause, + self.device.temperature_stages[chan].dwell, + ) diff --git a/system_tests/lewis_emulators/triton/states.py b/system_tests/lewis_emulators/triton/states.py new file mode 100644 index 0000000..e4ca48e --- /dev/null +++ b/system_tests/lewis_emulators/triton/states.py @@ -0,0 +1,5 @@ +from lewis.core.statemachine import State + + +class DefaultState(State): + pass diff --git a/system_tests/run_tests.bat b/system_tests/run_tests.bat new file mode 100644 index 0000000..65ab1d6 --- /dev/null +++ b/system_tests/run_tests.bat @@ -0,0 +1,10 @@ +@echo off +setlocal +REM Run this directory's tests using the IOC Testing Framework + +call "%~dp0..\..\..\..\config_env.bat" + +REM Command line arguments always passed to the test script +SET ARGS=--test_and_emulator %~dp0 +call %PYTHON3% -u "%EPICS_KIT_ROOT%\support\IocTestFramework\master\run_tests.py" %ARGS% %* +IF %ERRORLEVEL% NEQ 0 EXIT /b %ERRORLEVEL% diff --git a/system_tests/tests/__init__.py b/system_tests/tests/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/system_tests/tests/triton.py b/system_tests/tests/triton.py new file mode 100644 index 0000000..c1be736 --- /dev/null +++ b/system_tests/tests/triton.py @@ -0,0 +1,258 @@ +import unittest +import itertools + +from utils.channel_access import ChannelAccess +from utils.ioc_launcher import get_default_ioc_dir +from utils.test_modes import TestModes +from utils.testing import get_running_lewis_and_ioc, skip_if_recsim + +DEVICE_PREFIX = "TRITON_01" + +PID_TEST_VALUES = 0, 10**-5, 123.45, 10**5 +TEMPERATURE_TEST_VALUES = 0, 10**-5, 5.4321, 250 +PRESSURE_TEST_VALUES = TEMPERATURE_TEST_VALUES +POWER_TEST_VALUES = TEMPERATURE_TEST_VALUES +HEATER_RANGE_TEST_VALUES = 0.001, 0.316, 1000 +RESISTANCE_TEST_VALUES = 10, 3456 +EXCITATION_TEST_VALUES = PID_TEST_VALUES +TIME_DELAY_TEST_VALUES = RESISTANCE_TEST_VALUES + +VALID_TEMPERATURE_SENSORS = [i for i in range(1, 7)] +VALID_PRESSURE_SENSORS = [1, 2, 3, 5] + + +IOCS = [ + { + "name": DEVICE_PREFIX, + "directory": get_default_ioc_dir("TRITON"), + "emulator": "triton", + "macros": { + "USE_RAMP_FILE": 0, + "RAMP_FILE_NAME": "Default.txt", + "POLL_RATE": "1 second", + "CHANNEL_POLL_RATE": "2 second" + } + }, +] + + +TEST_MODES = [TestModes.RECSIM, TestModes.DEVSIM] + + +class TritonTests(unittest.TestCase): + """ + Tests for the Triton IOC. + """ + def setUp(self): + self._lewis, self._ioc = get_running_lewis_and_ioc("triton", DEVICE_PREFIX) + self.ca = ChannelAccess(device_prefix=DEVICE_PREFIX) + + def test_WHEN_device_is_started_THEN_it_is_not_disabled(self): + self.ca.assert_that_pv_is("DISABLE", "COMMS ENABLED") + + def test_WHEN_P_setpoint_is_set_THEN_readback_updates(self): + for value in PID_TEST_VALUES: + self.ca.assert_setting_setpoint_sets_readback(value, "P") + + def test_WHEN_I_setpoint_is_set_THEN_readback_updates(self): + for value in PID_TEST_VALUES: + self.ca.assert_setting_setpoint_sets_readback(value, "I") + + def test_WHEN_D_setpoint_is_set_THEN_readback_updates(self): + for value in PID_TEST_VALUES: + self.ca.assert_setting_setpoint_sets_readback(value, "D") + + def test_WHEN_temperature_setpoint_is_set_THEN_readback_updates(self): + for value in TEMPERATURE_TEST_VALUES: + self.ca.assert_setting_setpoint_sets_readback(value, set_point_pv="TEMP:SP", readback_pv="TEMP:SP:RBV") + + @skip_if_recsim("This is implemented at the protocol level, so does not work in recsim") + def test_WHEN_temperature_setpoint_is_set_THEN_closed_loop_turned_on_automatically(self): + for value in TEMPERATURE_TEST_VALUES: + self.ca.set_pv_value("CLOSEDLOOP:SP", "Off") + self.ca.assert_that_pv_is("CLOSEDLOOP", "Off") + self.ca.assert_setting_setpoint_sets_readback(value, set_point_pv="TEMP:SP", readback_pv="TEMP:SP:RBV") + self.ca.assert_that_pv_is("CLOSEDLOOP", "On") + + def test_GIVEN_closed_loop_already_on_WHEN_temperature_setpoint_is_set_THEN_closed_loop_setpoint_not_reprocessed(self): + for value in TEMPERATURE_TEST_VALUES: + self.ca.set_pv_value("CLOSEDLOOP:SP", "On") + self.ca.assert_that_pv_is("CLOSEDLOOP", "On") + timestamp_before = self.ca.get_pv_value("CLOSEDLOOP:SP.TSEL") + self.ca.assert_setting_setpoint_sets_readback(value, set_point_pv="TEMP:SP", readback_pv="TEMP:SP:RBV") + self.ca.assert_that_pv_is("CLOSEDLOOP", "On") + self.ca.assert_that_pv_is("CLOSEDLOOP:SP.TSEL", timestamp_before) + self.ca.assert_that_pv_value_is_unchanged("CLOSEDLOOP:SP.TSEL", wait=5) + + def test_WHEN_heater_range_is_set_THEN_readback_updates(self): + for value in HEATER_RANGE_TEST_VALUES: + self.ca.assert_setting_setpoint_sets_readback(value, "HEATER:RANGE") + + @skip_if_recsim("Lewis backdoor not available in recsim") + def test_WHEN_heater_power_is_set_via_backdoor_THEN_pv_has_the_value_just_set(self): + for value in HEATER_RANGE_TEST_VALUES: + self._lewis.backdoor_set_on_device("heater_power", value) + self.ca.assert_that_pv_is("HEATER:POWER", value) + self.ca.assert_that_pv_alarm_is("HEATER:POWER", self.ca.Alarms.NONE) + + @skip_if_recsim("Lewis backdoor not available in recsim") + def test_WHEN_closed_loop_mode_is_set_via_backdoor_THEN_the_closed_loop_pv_updates_with_value_just_set(self): + for value in [False, True, False]: # Need to check both transitions work properly + self._lewis.backdoor_set_on_device("closed_loop", value) + self.ca.assert_that_pv_is("CLOSEDLOOP", "On" if value else "Off") + + @skip_if_recsim("Behaviour too complex for recsim") + def test_WHEN_channels_are_enabled_and_disabled_via_pv_THEN_the_readback_pv_updates_with_value_just_set(self): + for chan in VALID_TEMPERATURE_SENSORS: + for enabled in [False, True, False]: # Need to check both transitions work properly + self.ca.assert_setting_setpoint_sets_readback( + "ON" if enabled else "OFF", "CHANNELS:T{}:STATE".format(chan)) + + @skip_if_recsim("Lewis backdoor not available in recsim") + def test_WHEN_a_short_status_is_set_on_device_THEN_displayed_status_is_identical(self): + # Status message that could be contained in an EPICS string type + short_status = "Device status" + assert 0 < len(short_status) < 40 + + # Status message that device is likely to return - longer than EPICS string type but reasonable for a protocol + medium_status = "This is a device status that contains a bit more information" + assert 40 < len(medium_status) < 256 + + # Short and medium statuses should be displayed in full. + for status in [short_status, medium_status]: + self._lewis.backdoor_set_on_device("status", status) + self.ca.assert_that_pv_is("STATUS", status) + + @skip_if_recsim("Lewis backdoor not available in recsim") + def test_WHEN_long_status_is_set_on_device_THEN_displayed_status_truncated_but_displays_at_least_500_chars(self): + + # Somewhat arbitrary, but decide on a minimum number of characters that should be displayed in a + # status message to the user if the status message is very long. This seems to be a reasonable + # number given the messages expected, but the manual does not provide an exhaustive list. + minimum_characters_in_pv = 500 + + # Very long status message, used to check that very long messages can be handled gracefully + long_status = "This device status is quite long:" + " (here is a load of information)" * 50 + + assert minimum_characters_in_pv < len(long_status) + + # Allow truncation for long status, but it should still display as many characters as possible + self._lewis.backdoor_set_on_device("status", long_status) + self.ca.assert_that_pv_value_causes_func_to_return_true( + "STATUS", lambda val: long_status.startswith(val) and len(val) >= minimum_characters_in_pv) + + @skip_if_recsim("Lewis backdoor not available in recsim") + def test_WHEN_automation_is_set_on_device_THEN_displayed_automation_is_identical(self): + automations = [ + "Warming up to 200K", + "Cooling down to 1K", + ] + + for automation in automations: + self._lewis.backdoor_set_on_device("automation", automation) + self.ca.assert_that_pv_is("AUTOMATION", automation) + + def _set_temp_via_backdoor(self, channel, temp): + self._lewis.backdoor_command(["device", "set_temperature_backdoor", "'{}'".format(channel), "{}".format(temp)]) + + @skip_if_recsim("Lewis backdoor not available in recsim") + def test_WHEN_stil_temp_is_set_via_backdoor_THEN_pv_updates(self): + for temp in TEMPERATURE_TEST_VALUES: + self._set_temp_via_backdoor("STIL", temp) + self.ca.assert_that_pv_is("STIL:TEMP", temp) + + @skip_if_recsim("Lewis backdoor not available in recsim") + def test_WHEN_mc_temp_is_set_via_backdoor_THEN_pv_updates(self): + for temp in TEMPERATURE_TEST_VALUES: + self._set_temp_via_backdoor("MC", temp) + self.ca.assert_that_pv_is("MC:TEMP", temp) + + @skip_if_recsim("Lewis backdoor not available in recsim") + def test_WHEN_sorb_temp_is_set_via_backdoor_THEN_pv_updates(self): + for temp in TEMPERATURE_TEST_VALUES: + self._set_temp_via_backdoor("SORB", temp) + self.ca.assert_that_pv_is("SORB:TEMP", temp) + + @skip_if_recsim("Lewis backdoor not available in recsim") + def test_WHEN_4KHX_temp_is_set_via_backdoor_THEN_pv_updates(self): + for temp in TEMPERATURE_TEST_VALUES: + self._set_temp_via_backdoor("PT2", temp) + self.ca.assert_that_pv_is("4KHX:TEMP", temp) + + @skip_if_recsim("Lewis backdoor not available in recsim") + def test_WHEN_jthx_temp_is_set_via_backdoor_THEN_pv_updates(self): + for temp in TEMPERATURE_TEST_VALUES: + self._set_temp_via_backdoor("PT1", temp) + self.ca.assert_that_pv_is("JTHX:TEMP", temp) + + @skip_if_recsim("Lewis backdoor not available in recsim") + def test_WHEN_pressure_is_set_via_backdoor_THEN_pressure_pv_updates(self): + for sensor, pressure in itertools.product(VALID_PRESSURE_SENSORS, PRESSURE_TEST_VALUES): + self._lewis.backdoor_command(["device", "set_pressure_backdoor", str(sensor), str(pressure)]) + self.ca.assert_that_pv_is("PRESSURE:P{}".format(sensor), pressure) + + def test_WHEN_closed_loop_is_set_via_pv_THEN_readback_updates(self): + for state in [False, True, False]: + self.ca.assert_setting_setpoint_sets_readback("On" if state else "Off", "CLOSEDLOOP") + + @skip_if_recsim("Lewis backdoor not available in recsim") + def test_WHEN_read_mc_id_is_issued_via_arbitrary_command_THEN_response_is_in_format_device_uses(self): + self.ca.set_pv_value("ARBITRARY:SP", "READ:SYS:DR:CHAN:MC") + self.ca.assert_that_pv_value_causes_func_to_return_true("ARBITRARY", + lambda val: val.startswith("STAT:SYS:DR:CHAN:MC:")) + + @skip_if_recsim("Lewis backdoor not available in recsim") + def test_WHEN_channel_temperature_is_set_via_backdoor_THEN_the_pvs_update_with_values_just_written(self): + for chan, value in itertools.product(VALID_TEMPERATURE_SENSORS, TEMPERATURE_TEST_VALUES): + self._lewis.backdoor_command(["device", "set_sensor_property_backdoor", str(chan), "temperature", str(value)]) + self.ca.assert_that_pv_is("CHANNELS:T{}:TEMP".format(chan), value) + + @skip_if_recsim("Lewis backdoor not available in recsim") + def test_WHEN_channel_resistance_is_set_via_backdoor_THEN_the_pvs_update_with_values_just_written(self): + for chan, value in itertools.product(VALID_TEMPERATURE_SENSORS, RESISTANCE_TEST_VALUES): + self._lewis.backdoor_command(["device", "set_sensor_property_backdoor", str(chan), "resistance", str(value)]) + self.ca.assert_that_pv_is("CHANNELS:T{}:RES".format(chan), value) + + @skip_if_recsim("Lewis backdoor not available in recsim") + def test_WHEN_channel_excitation_is_set_via_backdoor_THEN_the_pvs_update_with_values_just_written(self): + for chan, value in itertools.product(VALID_TEMPERATURE_SENSORS, EXCITATION_TEST_VALUES): + self._lewis.backdoor_command(["device", "set_sensor_property_backdoor", str(chan), "excitation", str(value)]) + self.ca.assert_that_pv_is("CHANNELS:T{}:EXCITATION".format(chan), value) + + @skip_if_recsim("Lewis backdoor not available in recsim") + def test_WHEN_channel_pause_is_set_via_backdoor_THEN_the_pvs_update_with_values_just_written(self): + for chan, value in itertools.product(VALID_TEMPERATURE_SENSORS, TIME_DELAY_TEST_VALUES): + self._lewis.backdoor_command(["device", "set_sensor_property_backdoor", str(chan), "pause", str(value)]) + self.ca.assert_that_pv_is("CHANNELS:T{}:PAUSE".format(chan), value) + + @skip_if_recsim("Lewis backdoor not available in recsim") + def test_WHEN_channel_dwell_is_set_via_backdoor_THEN_the_pvs_update_with_values_just_written(self): + for chan, value in itertools.product(VALID_TEMPERATURE_SENSORS, TIME_DELAY_TEST_VALUES): + self._lewis.backdoor_command(["device", "set_sensor_property_backdoor", str(chan), "dwell", str(value)]) + self.ca.assert_that_pv_is("CHANNELS:T{}:DWELL".format(chan), value) + + @skip_if_recsim("Lewis backdoor not available in recsim") + def test_WHEN_heater_resistance_is_changed_THEN_heater_heater_resistance_pv_updates(self): + for heater_resistance in RESISTANCE_TEST_VALUES: + self._lewis.backdoor_set_on_device("heater_resistance", heater_resistance) + self.ca.assert_that_pv_is_number("HEATER:RES", heater_resistance) + + @skip_if_recsim("Lewis backdoor not available in recsim") + def test_WHEN_heater_resistance_and_power_are_changed_THEN_heater_current_is_calculated_correctly(self): + for res, power in itertools.product(RESISTANCE_TEST_VALUES, POWER_TEST_VALUES): + self._lewis.backdoor_set_on_device("heater_resistance", res) + self._lewis.backdoor_set_on_device("heater_power", power) + + self.ca.assert_that_pv_is_number("HEATER:CURR", (power/res)**0.5, tolerance=0.01) # Ohm's law P = RI^2 + + @skip_if_recsim("Lewis backdoor not available in recsim") + def test_WHEN_heater_current_and_range_are_changed_THEN_heater_percent_power_is_calculated_correctly(self): + + for rang, res, power in itertools.product(HEATER_RANGE_TEST_VALUES, RESISTANCE_TEST_VALUES, POWER_TEST_VALUES): + self._lewis.backdoor_set_on_device("heater_resistance", res) + self._lewis.backdoor_set_on_device("heater_power", power) + self._lewis.backdoor_set_on_device("heater_range", rang) + + assert rang != 0, "Heater range of zero will cause a zero division error!" + + self.ca.assert_that_pv_is_number("HEATER:PERCENT", 100*((power/res)**0.5)/rang, tolerance=0.05)