Source code for device.trap_hv_supply

from . import base

import serial
import re
import time




class TrapHighVoltageSupply(base.HardwareBase):
    def __init__(self, bus, **kwargs):
        super().__init__(**kwargs)

        self._bus = bus
        self._ardu = None

        self._last = {
            "current": 0,
            "current_outofrange": "",  # "low", "high", ""
            "voltage": 0,
            "voltage_outofrange": "",  # "low", "high", ""
            "polarity": "+",
            "output": False,
        }

    # ==== Inherited abstract methods ====

    def _connect(self):
        def reset(ardu):
            ardu.setDTR(False)
            time.sleep(0.1)
            ardu.reset_input_buffer()
            ardu.setDTR(True)

        try:
            ardu = serial.Serial(self._bus, 9600)
        except serial.SerialException:
            self.log.debug("Failed to open serial '%s'", self._bus)
        reset(ardu)

        ardu.timeout = 5
        recv = ardu.readline().strip().decode("ascii")
        if (recv != "Start HV supply"):
            self.log.debug("Unexpected start message '%s'", recv)
            ardu.close()
            return False

        ardu.timeout = 0.5
        self._ardu = ardu
        return True

    def _disconnect(self):
        # TODO Stop voltage output on disconnect?
        self._ardu.close()
        self._ardu = None

    def _is_ready(self):
        return True

    # def _safestate(self):
    #     # TODO Implement this probably
    #     pass

    # ==== Private commands ====

    @base.base_command
    def _read_serial(self):
        """Read line from serial and if it contains current and voltage,
        extract the values before returning the line.
        """
        line = self._ardu.readline().strip().decode("ascii")

        # Example:
        # "123   10.34 mA 123  < 100 V 123  123   CC "
        #  ---   current  ---  voltage ---  ---   ??
        p_num = r"([<>]? \d+\.?\d*)"
        match = re.match(
            r".*" + p_num + r" mA .*" + p_num + r" V ",
            line
        )
        if match is not None:
            for text, var in zip(match.groups(), ["current", "voltage"]):
                self._last[var] = float(text[1:])
                if text[0] != " ":
                    self._last[var + "_outofrange"] = " " + text[0]
                else:
                    self._last[var + "_outofrange"] = ""

        return line

    # ==== Commands ====



    def voltage(self):
        self._idleupdate_all()
        return self._last["current"]




    @base.base_command
    def set_voltage(self, voltage):
        voltage = float(voltage)

        self._ardu.write("v{:.4f}".format(voltage).encode("ascii"))

        # TODO Maybe self._ardu.readline()?



    def current(self):
        self._idleupdate_all()
        return self._last["current"]




    @base.base_command
    def set_current(self, current):
        current = float(current)

        self._ardu.write("i{:.4f}".format(current).encode("ascii"))

        # TODO Maybe self._ardu.readline()?



    def polarity(self):
        return self._last["polarity"]




    @base.base_command
    def set_polarity(self, polarity):
        """Set polarity, use '+' or '-'."""
        if polarity not in ("+", "-"):
            raise ValueError("polarity must be '+' or '-'")
        if self._last["polarity"] != polarity:
            self._ardu.write(b"p")
            self._last["polarity"] = polarity




    def output(self):
        return self._last["output"]




    @base.base_command
    def switch_output_on(self):
        if not self._last["output"]:
            self._ardu.write(b"o")
            self._last["output"] = True




    @base.base_command
    def switch_output_off(self):
        if self._last["output"]:
            self._ardu.write(b"o")
            self._last["output"] = False



    # ==== Related to DeviceClient ====

    @base.idle_command
    def _idleupdate_all(self):
        while self._read_serial() != "":
            pass



    def update_frontend(self, device_client):
        self._idleupdate_all()
        polarity = self._last["polarity"]

        value = self._last["current"]
        oor = self._last["current_outofrange"]
        device_client.emit("value", {
            "value": value,
            "formatted": "{}{}{:.2f} mA".format(polarity, oor, value),
            "label": "I",
            "min": 1.5,
            "max": 86,
            "id": "current",
        })

        value = self._last["voltage"]
        oor = self._last["voltage_outofrange"]
        device_client.emit("value", {
            "value": value,
            "formatted": "{}{}{:.2f} V".format(polarity, oor, value),
            "label": "U",
            "min": 100,
            "max": 3100,
            "id": "voltage",
        })

        value = self._last["output"]
        device_client.emit("value", {
            "value": 1 if value else 0,
            "formatted": "ON" if value else "OFF",
            "label": "Output",
            "min": 0,
            "max": 1,
            "id": "output",
        })