Merge branch 'temp_branch' into add_isis_devices_2

This merges history of https://github.com/ISISComputingGroup/EPICS-DeviceEmulator into lewis

Author: rerpha

.git-blame-ignore-revs

@@ -0,0 +1,3 @@
+# ignore ruff format and --fix
+7d074d129ef30f64dee361531fe3d6a020cdff98
+3c5addd30d6e6f6ae34b02333d34360d773ed3bf

lewis/devices/Lksh218/__init__.py

@@ -0,0 +1,3 @@
+from .device import SimulatedLakeshore218
+
+__all__ = ["SimulatedLakeshore218"]

lewis/devices/Lksh218/device.py

@@ -0,0 +1,86 @@
+from collections import OrderedDict
+
+from lewis.devices import StateMachineDevice
+
+from .states import DefaultState
+
+NUMBER_OF_TEMP_CHANNELS = 8
+NUMBER_OF_SENSOR_CHANNELS = 8
+
+
+class SimulatedLakeshore218(StateMachineDevice):
+    """Simulated Lakeshore 218
+    """
+
+    def _initialize_data(self):
+        """Sets the initial state of the device.
+        """
+        self._temps = [1.0] * NUMBER_OF_TEMP_CHANNELS
+        self._sensors = [0.5] * NUMBER_OF_SENSOR_CHANNELS
+        self.temp_all = ""
+        self.sensor_all = ""
+        self.connected = True
+
+    @staticmethod
+    def _get_state_handlers():
+        """Returns: States and their names.
+        """
+        return {DefaultState.NAME: DefaultState()}
+
+    @staticmethod
+    def _get_initial_state():
+        """Returns: The name of the initial state.
+        """
+        return DefaultState.NAME
+
+    @staticmethod
+    def _get_transition_handlers():
+        """Returns: The state transitions.
+        """
+        return OrderedDict()
+
+    def get_temp(self, number):
+        """Gets the temperature of a specific temperature sensor.
+
+        Args:
+            number: Integer between 1 and 8.
+
+        Returns:
+            float: Temperature value at position (number - 1) in temps.
+        """
+        return self._temps[number - 1]
+
+    def set_temp(self, number, temperature):
+        """Sets the (number - 1) temp pv to temperature.
+
+        Args:
+            number: Integer between 1 and 8.
+            temperature: Temperature reading to set.
+
+        Returns:
+            None
+        """
+        self._temps[number - 1] = temperature
+
+    def get_sensor(self, number):
+        """Gets the sensor reading of a specific sensor.
+
+        Args:
+            number: Integer between 1 and 8.
+
+        Returns:
+            float: Value of sensor at position (number - 1) in sensors.
+        """
+        return self._sensors[number - 1]
+
+    def set_sensor(self, number, value):
+        """Sets the (number - 1) sensor pv to value.
+
+        Args:
+            number: Integer between 1 and 8.
+            value: Sensor reading to set.
+
+        Returns:
+            None
+        """
+        self._sensors[number - 1] = value

lewis/devices/Lksh218/interfaces/__init__.py

@@ -0,0 +1,3 @@
+from .stream_interface import Lakeshore218StreamInterface
+
+__all__ = ["Lakeshore218StreamInterface"]

lewis/devices/Lksh218/interfaces/stream_interface.py

@@ -0,0 +1,84 @@
+from lewis.adapters.stream import StreamInterface
+from lewis.utils.command_builder import CmdBuilder
+
+
+def if_connected(f):
+    """Decorator that executes f if the device is connected and returns None otherwise.
+
+    Args:
+        f: function to be executed if the device is connected.
+
+    Returns:
+        The value of f(*args) if the device is connected and None otherwise.
+    """
+
+    def wrapper(*args):
+        connected = getattr(args[0], "_device").connected
+        if connected:
+            result = f(*args)
+        else:
+            result = None
+        return result
+
+    return wrapper
+
+
+class Lakeshore218StreamInterface(StreamInterface):
+    """Stream interface for the serial port
+    """
+
+    in_terminator = "\r\n"
+    out_terminator = "\r\n"
+
+    commands = {
+        CmdBuilder("get_temp").escape("KRDG? ").arg("[1-8]").build(),
+        CmdBuilder("get_sensor").escape("SRDG? ").arg("[1-8]").build(),
+        CmdBuilder("get_temp_all").escape("KRDG? 0").build(),
+        CmdBuilder("get_sensor_all").escape("SRDG? 0").build(),
+    }
+
+    @if_connected
+    def get_temp(self, number):
+        """Returns the temperature of a TEMP pv.
+
+        Args:
+            number: integer between 1 and 8
+
+        Returns:
+            float: temperature
+        """
+        number = int(number)
+        temperature = self._device.get_temp(number)
+        return temperature
+
+    @if_connected
+    def get_sensor(self, number):
+        """Returns the temperature of a SENSOR pv.
+
+        Args:
+            number: integer between 1 and 8
+
+        Returns:
+            float: sensor_reading
+        """
+        number = int(number)
+        sensor_reading = self._device.get_sensor(number)
+        return sensor_reading
+
+    @if_connected
+    def get_temp_all(self):
+        """Returns a string from TEMPALL pv.
+
+        Returns:
+            string: value of TEMPALL pv.
+        """
+        return self._device.temp_all
+
+    @if_connected
+    def get_sensor_all(self):
+        """Returns a string from SENSORALL pv.
+
+        Returns:
+            string: value of SENSORALL pv.
+        """
+        return self._device.sensor_all

lewis/devices/Lksh218/states.py

@@ -0,0 +1,8 @@
+from lewis.core.statemachine import State
+
+
+class DefaultState(State):
+    """Device is in default state.
+    """
+
+    NAME = "Default"

lewis/devices/ag33220a/__init__.py

@@ -0,0 +1,5 @@
+from ..lewis_versions import LEWIS_LATEST
+from .device import SimulatedAG33220A
+
+framework_version = LEWIS_LATEST
+__all__ = ["SimulatedAG33220A"]

lewis/devices/ag33220a/device.py

@@ -0,0 +1,169 @@
+from lewis.devices import Device
+
+
+class SimulatedAG33220A(Device):
+    """Simulated AG33220A
+    """
+
+    connected = True
+
+    # Constants
+    AMP_MIN = 0.01
+    AMP_MAX = 10
+    OFF_MAX = 4.995
+    VOLT_MAX = 5
+    VOLT_MIN = -5
+    VOLT_LOW_MAX = 4.99
+    VOLT_HIGH_MIN = -4.99
+    VOLT_PRECISION = 0.01
+    FREQ_MINS = {
+        "SIN": 10**-6,
+        "SQU": 10**-6,
+        "RAMP": 10**-6,
+        "PULS": 5 * 10**-4,
+        "NOIS": 10**-6,
+        "USER": 10**-6,
+    }
+    FREQ_MAXS = {
+        "SIN": 2 * 10**7,
+        "SQU": 2 * 10**7,
+        "RAMP": 2 * 10**5,
+        "PULS": 5 * 10**6,
+        "NOIS": 2 * 10**7,
+        "USER": 6 * 10**6,
+    }
+
+    # Device variables
+    idn = "Agilent Technologies,33220A-MY44033103,2.02-2.02-22-2"
+    amplitude = 0.1
+    frequency = 1000
+    offset = 0
+    units = "VPP"
+    function = "SIN"
+    output = "ON"
+    voltage_high = 0.05
+    voltage_low = -0.05
+    range_auto = "OFF"
+
+    def limit(self, value, minimum, maximum):
+        """Limits an input number between two given numbers or sets the value to the maximum or minimum.
+
+        :param value: the value to be limited
+        :param minimum: the smallest that the value can be
+        :param maximum: the largest that the value can be
+
+        :return: the value after it has been limited
+        """
+        if type(value) is str:
+            try:
+                value = float(value)
+            except ValueError:
+                return {"MIN": minimum, "MAX": maximum}[value]
+
+        return max(min(value, maximum), minimum)
+
+    def set_new_amplitude(self, new_amplitude):
+        """Changing the amplitude to the new amplitude whilst also changing the offset if voltage high or low is
+        outside the boundary. The volt high and low are then updated.
+
+        :param new_amplitude: the amplitude to set the devices amplitude to
+        """
+        new_amplitude = self.limit(new_amplitude, self.AMP_MIN, self.AMP_MAX)
+
+        peak_amp = 0.5 * new_amplitude
+        if self.offset + peak_amp > self.VOLT_MAX:
+            self.offset = self.VOLT_MAX - peak_amp
+        elif self.offset - peak_amp < self.VOLT_MIN:
+            self.offset = self.VOLT_MIN + peak_amp
+
+        self.amplitude = new_amplitude
+
+        self._update_volt_high_and_low(self.amplitude, self.offset)
+
+    def set_new_frequency(self, new_frequency):
+        """Sets the frequency within limits between upper and lower bound (depends on the function).
+
+        :param new_frequency: the frequency to set to
+        """
+        self.frequency = self.limit(
+            new_frequency, self.FREQ_MINS[self.function], self.FREQ_MAXS[self.function]
+        )
+
+    def set_new_voltage_high(self, new_voltage_high):
+        """Sets a new voltage high which then changes the voltage low to keep it lower.
+        The voltage offset and amplitude are then updated.
+
+        :param new_voltage_high: the value of voltage high to set to
+        """
+        new_voltage_high = self.limit(new_voltage_high, self.VOLT_HIGH_MIN, self.VOLT_MAX)
+        if new_voltage_high <= self.voltage_low:
+            self.voltage_low = self.limit(
+                new_voltage_high - self.VOLT_PRECISION, self.VOLT_MIN, new_voltage_high
+            )
+        self._update_volt_and_offs(self.voltage_low, new_voltage_high)
+
+    def set_new_voltage_low(self, new_voltage_low):
+        """Sets a new voltage high which then changes the voltage low to keep it higher.
+        The voltage offset and amplitude are then updated.
+
+        :param new_voltage_low: the value of voltage low which is to be set
+        """
+        new_voltage_low = self.limit(new_voltage_low, self.VOLT_MIN, self.VOLT_LOW_MAX)
+        if new_voltage_low >= self.voltage_high:
+            self.voltage_high = self.limit(
+                new_voltage_low + self.VOLT_PRECISION, new_voltage_low, self.VOLT_MAX
+            )
+        self._update_volt_and_offs(new_voltage_low, self.voltage_high)
+
+    def _update_volt_and_offs(self, new_low, new_high):
+        """Updates the value of amplitude and offset if there is a change in voltage low or voltage high.
+
+        :param new_low: the value of voltage low
+        :param new_high: the value of voltage high
+        """
+        self.voltage_high = new_high
+        self.voltage_low = new_low
+        self.amplitude = self.voltage_high - self.voltage_low
+        self.offset = (self.voltage_high + self.voltage_low) / 2
+
+    def set_offs_and_update_voltage(self, new_offset):
+        """Sets the value of offset and updates the amplitude, voltage low and voltage high for a new value of the offset.
+
+        :param new_offset: the new offset to be set
+        """
+        new_offset = self.limit(new_offset, -self.OFF_MAX, self.OFF_MAX)
+        if new_offset + self.voltage_high > self.VOLT_MAX:
+            self.amplitude = 2 * (self.VOLT_MAX - new_offset)
+            self.voltage_high = self.VOLT_MAX
+            self.voltage_low = self.VOLT_MAX - self.amplitude
+        elif new_offset + self.voltage_low < self.VOLT_MIN:
+            self.amplitude = 2 * (self.VOLT_MIN - new_offset)
+            self.voltage_low = self.VOLT_MIN
+            self.voltage_high = self.VOLT_MIN + self.amplitude
+        else:
+            self._update_volt_high_and_low(self.amplitude, new_offset)
+        self.offset = new_offset
+
+    def _update_volt_high_and_low(self, new_volt, new_offs):
+        """Updates the value of voltage high and low for a given value of amplitude and offset.
+
+        :param new_volt: the value of the amplitude
+        :param new_offs: the value of the offset
+        """
+        self.offset = new_offs
+        self.amplitude = new_volt
+        self.voltage_high = new_offs + new_volt / 2
+        self.voltage_low = new_offs - new_volt / 2
+
+    def get_output(self):
+        return ["OFF", "ON"].index(self.output)
+
+    def get_range_auto(self):
+        possible_ranges = ["OFF", "ON", "ONCE"]
+        return possible_ranges.index(self.range_auto)
+
+    def set_function(self, new_function):
+        self.function = new_function
+        self.frequency = self.limit(
+            self.frequency, self.FREQ_MINS[new_function], self.FREQ_MAXS[new_function]
+        )

lewis/devices/ag33220a/interfaces/__init__.py

@@ -0,0 +1,3 @@
+from .stream_interface import AG33220AStreamInterface
+
+__all__ = ["AG33220AStreamInterface"]