Source code for device.trap_cathode

import serial
from . import base

import re
import time


# TODO
#   - 1. upload modified arduino code
#   - 2. test and adjust calibration file ("calibrations/calib_trap.cfg")



class TrapCathodeHeater(base.ArduinoMotorsBase):
    """Arduino based motor controller (trap experiment), which controls the
    autotransformator output. Also there are two power relays.

    Communication:

    - serial interface to Arduino
    """

    def __init__(self, calibfile, bus, limits_enabled=True, **kwargs):
        super().__init__(calibfile, [bus], limits_enabled, **kwargs)
        self._last = {
            "switched_on": False
        }

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

    def _defaultstate(self):
        self._position = {key: 0 for key in self._ardu}
        self._last = {
            "current": 0.0,
            "trafo_on": False,
            "softstart_on": False,
        }

    def _safestate(self):
        """Cut cathode power, stop motor and request new homing."""
        self.switch_output_off()
        super()._safestate()
        self._homed = False

    # ==== Private commands ====

    # Overriding parent function - There are no sensors of position so we need
    # to manage the self._position property manually.
    @base.compound_command
    def _move(self, motor_id, distance, check_limits=True):
        super()._move(motor_id, distance, check_limits)
        self._position[motor_id] += distance

    # ==== Commands ====



    @base.compound_command
    def homing(self):
        super().homing()
        self._position["zhaveni"] = 0.0




    def current(self):
        "Return the current in cathode"
        self._idleupdate_position("zhaveni")
        self._idleupdate_other()
        return self._last["current"]




    @base.compound_command
    def moveto(self, position):
        """Move motor to position [mm]."""
        if not self.is_output_on():
            raise base.CommandError("Switch-on output first")
        super().moveto("zhaveni", position)




    @base.compound_command
    def set_current(self, current):
        """Set output transformator current [A] by moving the motor."""
        if not self.is_output_on():
            raise base.CommandError("Switch-on output first")
        current = float(current)
        if not (0 <= current <= 30):
            raise base.CommandError("Current out of range (0-30A)")

        for _ in range(3):
            self.moveto(
                self.position("zhaveni")
                + (current-self.current()) / (45 / 3.5)  # 45A is cca 3.5mm
            )
            time.sleep(0.2)
            if abs(self.current() - current) < 1:
                break




    def is_output_on(self):
        """Return the state of relay that powers transformator."""
        return self._last["trafo_on"]




    @base.base_command
    def switch_output_on(self):
        if not self.is_homed():
            raise base.CommandError("Homing required")
        if not self._last["trafo_on"]:
            self._ardu["zhaveni"].write(b"t")
            self._read_serial("zhaveni")  # "Trafo 1\r\n"
            time.sleep(1.5)
            self._read_serial("zhaveni")  # "Soft Start 1\r\n"

            self._idleupdate_other()




    @base.base_command
    def switch_output_off(self):
        if self._last["trafo_on"]:
            self._ardu["zhaveni"].write(b"t")
            self._read_serial("zhaveni")  # "Trafo 0\r\n"
            time.sleep(1.5)
            self._read_serial("zhaveni")  # "Soft Start 0\r\n"

            self._idleupdate_other()
            self._homed = False


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

    @base.idle_command
    def _idleupdate_other(self):
        self._ardu["zhaveni"].write(b"q")

        # Ignore any left-over lines
        line = ""
        while "Proud" not in line:
            line = self._read_serial("zhaveni")

        # Then parse various states
        #
        # "Proud 1.23 A "
        # "Trafo 1"
        # "Soft Start 0"
        parsed = re.search(r"Proud ([0-9]+\.[0-9]+) A", line)
        self._last["current"] = float(parsed.group(1))
        self._last["trafo_on"] = self._read_serial("zhaveni")[-1] == "1"
        self._last["softstart_on"] = self._read_serial("zhaveni")[-1] == "1"



    def update_frontend(self, device_client):
        self._idleupdate_position("zhaveni")  # This serves no purpose (no position sensors)
        self._idleupdate_other()

        if self._last["current"] > 30:
            self.log.warning("Tripped 30A current limit, stopping")
            self.to_safestate()

        value = self.position("zhaveni")
        device_client.emit("value", {
            "value": value,
            "formatted": "{:.2f} mm".format(value),
            "label": "Motor",
            "min": min(self.limits("zhaveni")),
            "max": max(self.limits("zhaveni")),
            "id": "position_" + "zhaveni"
        })

        value = self._last["current"]
        device_client.emit("value", {
            "value": value,
            "formatted": "{:.2f} A".format(value),
            "label": "Current",
            "min": 0,
            "max": 100,  # TODO Adjust as needed
            "id": "current",
        })

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