Custom Instruments

				
					""" Opentrons MagDeck Driver """
from enum import Enum
from typing import List, Optional
from serial import Serial  # type: ignore
from serial.tools.list_ports import comports  # type: ignore
from labsdk.state.parameters import load_parameters_env
 
 
class Commands(str, Enum):
    Home = 'G28.2'
    Probe = 'G38.2'
    GetPlateHeight = 'M836'
    GetInfo = 'M115'
    GetPosition = 'M114.2'
    Move = 'G0'
 
 
DEFAULT_PLATE_HEIGHT = 30.0
MAX_TRAVEL_DISTANCE = 45.0
 
COMMAND_TERMINATOR = '\r\n\r\n'
ACK = 'ok\r\nok\r\n'
MAGDECK_PORT_NAME = 'Opentrons MagDeck'
 
 
class OpentronsMagnetDriver():
    """ Driver for Opentrons Magnet """
 
    def __init__(self, port_name: Optional[str] = None):
        self.port_name = port_name
        self._port = None
 
    def connect(self) -> None:
        if self.port_name is None:
            ports = list(map(lambda port: port.device, filter(lambda port: port.product == MAGDECK_PORT_NAME, comports())))
            if len(ports) == 0:
                raise ValueError('Could not find an Opentrons MagDeck')
            self.port_name = ports[0]
 
        if self._port is None:
            try:
                self._port = Serial(self.port_name, baudrate=115200, timeout=10)
                self.get_device_info()  # throws if no response is received
            except:
                self.disconnect()
                raise
 
    def disconnect(self) -> None:
        if self._port is not None:
            self._port.close()
            self._port = None
 
    @property
    def port(self) -> Serial:
        if self._port is None:
            raise ValueError('Not connected to a Opentrons MagDeck')
        return self._port
 
    def _send_command(self, elements: List[str]) -> str:
        self.port.reset_input_buffer()
        if elements[-1] is not COMMAND_TERMINATOR:
            elements.append(COMMAND_TERMINATOR)
        self.port.write(' '.join(elements).encode('utf8'))
 
        # Read until we reach the ack, or we hit the timeout
        response_bytes = self.port.read_until(ACK.encode('utf8'))
        if len(response_bytes) == 0:
            raise ValueError('MagDeck did not return valid device info')
 
        # verify that an ACK was in the response and then return the response 'body'
        response = response_bytes.decode('utf8')
        return response.split(ACK)[0].strip() if ACK in response else ''
 
    def move(self, mag_height: float = DEFAULT_PLATE_HEIGHT) -> None:
        """ Move to the specified height """
        if mag_height > MAX_TRAVEL_DISTANCE or mag_height < 0:
            raise ValueError(f'Height of {mag_height} is not supported')
        self._send_command([Commands.Move, f'Z{mag_height}'])
 
    def home(self) -> None:
        """ move the magnet to the home position """
        self._send_command([Commands.Home])
 
    def get_device_info(self) -> str:
        """ return the device info """
        response = self._send_command([Commands.GetInfo])
        if response == '':
            raise ValueError('Did not receive device info details from Opentrons MagDeck')
        return response
 
 
def main() -> None:
    params = load_parameters_env()
    height = params['height'].data
    driver = OpentronsMagnetDriver()
    driver.connect()
    try:
        driver.move(height)
    finally:
        driver.disconnect()
 
 
if __name__ == "__main__":
    main()