Source code for device.interfer_motors

from . import base




class InterferometryMotors(base.ArduinoMotorsBase):

    def __init__(self, calibfile, buses=None, limits_enabled=True, **kwargs):
        super().__init__(calibfile, buses, limits_enabled, **kwargs)
        self._relay_on = {"1": False, "2": False}

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

    def _defaultstate(self):
        speed = 450
        accel = 200

        for a in self._ardu.values():
            a.write("s{:d}".format(speed).encode("ascii"))
            a.readline()
            a.write("a{:d}".format(accel).encode("ascii"))
            a.readline()

        self._relay_on = {"1": False, "2": False}
        # self._update_relay_states()  # Unnecessary: both are off after reset

        # self._homed = True  # TODO Just temporary, homing does not work well

    def _disconnect(self):
        self.relay1_off()
        self.relay2_off()
        super()._disconnect()

    # ==== Private commands ====

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

        # Wait for correct response (skip through old responses)
        while True:
            resp = self._read_serial("R")
            if resp == "":
                raise base.TransmissionError("Expected a response (action 'q')")
            elif resp.startswith("Rele 1 :"):
                resp = [resp, self._read_serial("R")]
                break

        for i in range(2):
            self._relay_on[str(i+1)] = resp[i][len("Rele 1 :"):] == "0"

    @base.base_command
    def _toggle_relay(self, which):
        self._ardu["R"].write({"1": b"r", "2": b"t"}[which])  # Toggle relay 1/2 state
        if self._read_serial("R") == "":
            raise base.TransmissionError("Expected a response (action 'r'/'t')")
        self._relay_on[which] = not self._relay_on[which]

    # ==== Commands ====



    def is_relay1_on(self):
        """Return the state of the relay."""
        self._update_relay_states()
        return self._relay_on["1"]




    def is_relay2_on(self):
        """Return the state of the relay."""
        self._update_relay_states()
        return self._relay_on["2"]




    @base.base_command
    def relay1_on(self):
        """Activate relay 1."""
        if not self.is_relay1_on():
            self._toggle_relay("1")




    @base.base_command
    def relay2_on(self):
        """Activate relay 2."""
        if not self.is_relay2_on():
            self._toggle_relay("2")




    @base.base_command
    def relay1_off(self):
        """Deactivate relay 1."""
        if self.is_relay1_on():
            self._toggle_relay("1")




    @base.base_command
    def relay2_off(self):
        """Deactivate relay 2."""
        if self.is_relay2_on():
            self._toggle_relay("2")




    @base.compound_command
    def anetenna_moveto(self, position):
        """Move ANTENNA to position [mm]."""
        self.moveto("R", position)




    @base.compound_command
    def object_moveto(self, position):
        """Move OBJECT (e.g. glass) to position [mm]."""
        self.moveto("L", position)


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



    def update_frontend(self, device_client):
        for key, name in [("L", "Object"), ("R", "Antenna")]:
            self._idleupdate_position(key)

            value = self._position[key]
            device_client.emit("value", {
                "value": value,
                "formatted": "{:.2f} mm".format(value),
                "label": name,
                "min": min(self.limits(key)),
                "max": max(self.limits(key)),
                "id": "position_" + key
            })