pipette.py

Go to the documentation of this file.

#!/usr/bin/env python

# Software License Agreement (BSD License)
#
# Copyright (c) 2010, Willow Garage, Inc.
# All rights reserved.
#
# Redistribution and use in source and binary forms, with or without
# modification, are permitted provided that the following conditions
# are met:
#
#  * Redistributions of source code must retain the above copyright
#    notice, this list of conditions and the following disclaimer.
#  * Redistributions in binary form must reproduce the above
#    copyright notice, this list of conditions and the following
#    disclaimer in the documentation and/or other materials provided
#    with the distribution.
#  * Neither the name of Willow Garage, Inc. nor the names of its
#    contributors may be used to endorse or promote products derived
#    from this software without specific prior written permission.
#
# THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
# "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
# LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS
# FOR A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE
# COPYRIGHT OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT,
# INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING,
# BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES;
# LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER
# CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT
# LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN
# ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
# POSSIBILITY OF SUCH DAMAGE.

##\author Kevin Watts, Derek King

"""
Provides interface to Viaflo Single Channel Vision Pipette with Bluetooth
connection.

Interface allows opening, closing, polling and pipette operations.
"""

import bluetooth
import re
from collections import defaultdict

class NotConnectedException(Exception): pass
class InvalidChecksumError(Exception): pass
class IncompleteResponseError(Exception): pass
class InvalidMessageTypeError(Exception): pass
class InvalidEscapeSequenceError(Exception) : pass
class SeqNumMismatchError(Exception) : pass
class SeqNumMismatchError(Exception) : pass
class ResponseError(Exception) : pass

def _short_num_to_bytes(num):
    """
    Convert short number into two byte list representation
    @param number int : Number to convert
    @return [int,int] : Two bytes
    """
    return [(num >> 8) & 0xFF,  num & 0xFF]

def _bytes_to_num(data):
    """
    Convert list with two or four bytes into integer
    @param data, list of bytes (integers), with either 2 or 4 elements
    @return int : Number
    """
    total = 0
    for d in data:
        total = (total << 8) + d
    return total

def _convert_bytes_to_str(data):
    """
    Convert list of bytes into string
    """
    chrs = [ chr(val) for val in data ]
    return ''.join(chrs)

def _convert_str_to_bytes(data):
    """
    Unpack string, convert to list of bytes
    """
    return [ ord(c) for c in data ]


# special ASCII values
# start-of-text 
_STX = 2
# end-of-test 
_ETX = 3
# escape
_ESC = 27
# line-feed
_LF = 10

# Mapping from action commands to string descriptions of command
_action_dict = defaultdict(lambda : "???", {1:'Position Inquiry', 2:'Homing', 3:'Purge', 4:'Aspirate', 5:'Aspirate with jog', 6:'Dispense', 7:'Dispense to 0', 8:'Mix', 9:'Prompt'})


class ViafloResponse(object): 
    def __init__(self, data):
        self.length = _bytes_to_num((data[0], data[1]))
        if self.length != len(data): # Not sure 
            print data
            #print "Incomplete response. Expected data of length %d, received %d" % (self.length, len(data))
            raise IncompleteResponseError("Incomplete response. Expected data of length %d, received %d" % (self.length, len(data)))

        self.checksum = data[2]

        data_checksum = sum(data) & 0xFF
        if data_checksum != 0:
            #print "Checksum is invalid. Computed sum %d, Data checksum %d" % ( data_checksum, self.checksum)
            #print data
            raise InvalidChecksumError("Checksum is invalid. Computed sum %d, Msg checksum %d" % (data_checksum, self.checksum))

        self.seqnum = _bytes_to_num((data[3], data[4]))
        self.resend = data[5]
        self.msg_type = _bytes_to_num((data[6], data[7]))
        self.error_code = data[8]

        if self.error_code != 0:
            raise ResponseError("Got response error code %d" % self.error_code)
        
        self.body = data[9:]

    def _body_substr(self, msg_offset, length):
        """
        Semi-knock off of perl substring function.  Makes porting perl code to python easier.
        """
        # Perl code offsets from begining of response, however we separate out the body.  
        # So the msg_offset is off by 9 bytes.
        if msg_offset < 9:
            raise RuntimeError("Invalid offset %d" % msg_offset)
        return self.body[msg_offset-9:msg_offset-9+length]

    def _body_at(self, msg_offset):
        """
        Semi-knock off of perl substr function.  Returns value at index
        """
        # Perl code offsets from begining of response, however we separate out the body.  
        # So the msg_offset is off by 9 bytes.
        if msg_offset < 9:
            raise RuntimeError("Invalid offset %d" % msg_offset)
        return self.body[msg_offset-9]


class ViafloInfoResponse(ViafloResponse):
    def __init__(self, data):
        ViafloResponse.__init__(self, data)

        if self.msg_type != 1:
            raise InvalidMessageTypeError("Info message should be of type 1. Got type %d" % self.msg_type)

        self.owner = _convert_bytes_to_str(self._body_substr(10,12))
        self.serial = _convert_bytes_to_str(self._body_substr(22,10))

        self.fwmajor = _bytes_to_num(self._body_substr(32,2))
        self.fwminor = _bytes_to_num(self._body_substr(34,2))
        self.model = _convert_bytes_to_str(self._body_substr(36,5))
        self.type = _bytes_to_num(self._body_substr(41,1))
        self.voyager_flag = _bytes_to_num(self.body[33:34])
        self.aspirate_dispense_table_page = _bytes_to_num(self.body[34:36])
        self.aspirate_factory_factor = _bytes_to_num(self.body[36:38])
        self.dispense_factory_factor = _bytes_to_num(self.body[38:40])
        self.aspirate_dispense_factor = _bytes_to_num(self.body[40:42])
        self.dispense_dispense_factor = _bytes_to_num(self.body[42:44])
        self.pipette_min_stops = _bytes_to_num(self._body_substr(53,2))
        self.pipette_max_stops = _bytes_to_num(self._body_substr(55,2))
        self.first_flash_page = _bytes_to_num(self.body[48:50])
        self.calibration_timer_on = _bytes_to_num(self.body[50:52])
        self.calibration_timer_time_months = _bytes_to_num(self.body[52:54])
        self.time_high = _bytes_to_num(self.body[54:56])
        self.time_low = _bytes_to_num(self.body[56:58])
        self.motor_steps_millions = _bytes_to_num(self._body_substr(67,4))
        self.motor_steps = _bytes_to_num(self._body_substr(71,4))
        self.display_decimal = _bytes_to_num(self._body_substr(75,2))

        # Other values for FW > 1.7
        self.at_least_1_7 = self.fwmajor > 1 or (self.fwmajor == 1 and self.fwminor >= 7)
        
        if (self.at_least_1_7):
            self.firmware_engineering_revision = _bytes_to_num(self._body_substr(77,2))
            self.firmware_type = _bytes_to_num(self._body_substr(79,2))
            
            # More values...


    def __str__(self):
        type_dict = defaultdict(lambda : "???", {1:'Single Channel', 2:'Multi Channel', 3:'Step'})

        out = [ "ViafloInfoResponse" ]
        out.append("\tOwner: '%s'" % self.owner )
        out.append("\tSerial: '%s'" % self.serial )
        out.append("\tFW Major: %d" % self.fwmajor )
        out.append("\tFW Minor: %d" % self.fwminor )
        out.append("\tModel: '%s'" % self.model)
        out.append("\tType: %d '%s'" % (self.type, type_dict[self.type]) )
        out.append("\tVoyager Flag: %d" % self.voyager_flag )
        out.append("\tMotor Steps Millions: %d" % self.motor_steps_millions )
        out.append("\tMotor Steps: %d" % self.motor_steps)
        out.append("\tMin Stops: %d" % self.pipette_min_stops)
        out.append("\tMax Stops: %d" % self.pipette_max_stops)
        out.append("\tCalibration Timer On: %d" % self.calibration_timer_on)
        out.append("\tCalibration Timer Time Months: %d" % self.calibration_timer_time_months)
        out.append("\tDisplay Decimal: %d" % self.display_decimal)
        if (self.at_least_1_7):
            out.append("\tEngineering FW Rev: %d" % self.firmware_engineering_revision)
            out.append("\tFW Type: %d" % self.firmware_type)
        
        out.append("\tError Code (msg): %d" % self.error_code)
        out.append("\tBody Length: %d" % (len(self.body)))
        #out.append("\tRemainder: '%s'" % (_convert_bytes_to_str(self.body[68:])))

        return '\n'.join(out)
        
class ViafloPipettorActionResponse(ViafloResponse):
    def __init__(self, data):
        ViafloResponse.__init__(self, data)
        
        if self.msg_type != 4:
            raise InvalidMessageTypeError("Action response should be of type 4. Got type %d" % self.msg_type)

        self.action = _bytes_to_num(self._body_substr(10,1))
        self.pipette_stop = _bytes_to_num(self._body_substr(11,2))        

    def __str__(self):
        out = [ "ViafloPipettorActionResponse "]
        out.append("\tAction: %d '%s'" % (self.action, _action_dict[self.action]) )
        out.append("\tPipette Stop: %d" % self.pipette_stop)
        out.append("\tError Code: %d" % self.error_code)
        out.append("\tBody Length: %d" % (len(self.body)))
        return '\n'.join(out)

class ViafloVersionResponse(ViafloResponse):
    def __init__(self, data):
        ViafloResponse.__init__(self, data)

        def dot_join(data):
            return '.'.join([str(n) for n in data])
        
        self.tracking_number = dot_join(self._body_substr(10,5))
        self.overall_version = dot_join([self._body_at(19), self._body_at(20), self._body_at(27)])
        self.firmware_version = dot_join([_bytes_to_num(self._body_substr(15,2)),_bytes_to_num(self._body_substr(17,2)),self._body_at(28)])
        self.interpreter_version = dot_join(self._body_substr(21,2))
        self.hardware_control_version = dot_join(self._body_substr(23,2))
        self.motor_control_version = dot_join(self._body_substr(25,2))

    def __str__(self):
        out = [ "ViafloVersionResponse "]
        out.append("\tTracking Number: '%s'" % (self.tracking_number) )
        out.append("\tOverall Version: '%s'" % (self.overall_version) )
        out.append("\tFirmware Version: '%s'" % (self.firmware_version) )
        out.append("\tInterpreter Version: '%s'" % (self.interpreter_version) )
        out.append("\tHW Control Version: '%s'" % (self.hardware_control_version) )
        out.append("\tMotor Control Version: '%s'" % (self.motor_control_version) )
        return '\n'.join(out)

        

class ViafloCommand(object):
    def __init__(self, msg_type, body):
        self.checksum = 0
        self.length = len(body) + 8 # 8 is header
        self.body = body

        self.resend = 0
        self.type = msg_type

    def pack(self, seqnum):
        """
        Packs message for transport
        """
        data = [ 0 for i in range(8)]
        len_bytes = _short_num_to_bytes(self.length)
        data[0] = len_bytes[0]
        data[1] = len_bytes[1]
        data[2] = 0 #checksum placeholder    
        seq_bts = _short_num_to_bytes(seqnum)
        data[3] = seq_bts[0]
        data[4] = seq_bts[1]
        data[5] = 0 #resend?
        typ_bts = _short_num_to_bytes(self.type)
        data[6] = typ_bts[0]
        data[7] = typ_bts[1]
        data += self.body # Add body 
        
        # Compute checksum (negative checksum)
        data[2] = (-sum(data)) & 0xff

        #insert escape sequences in msg
        escaped_data = []
        for d in data:
            if (d==_STX) or (d==_ETX) or (d==_ESC):
                escaped_data.append(_ESC)
            escaped_data.append(d)            

        # todo: need debug print
        #print 'Outgoing data:', data
            
        # Document says to put STX at beginning of message, and ETX and end of message
        # Perl code puts 3 STX and 1 LF at end of message.
        cmd = [_STX] + escaped_data + [_ETX,_ETX,_ETX,_LF] 

        return  _convert_bytes_to_str(cmd)

class ViafloInfoCommand(ViafloCommand):
    def __init__(self):
        ViafloCommand.__init__(self, 1, [])

class ViafloVersionCommand(ViafloCommand):
    def __init__(self):        
        ViafloCommand.__init__(self, 10, [])        

class ViafloPurgeCommand(ViafloCommand):
    def __init__(self, speed, number_of_bells=0):
        body = [3,speed]
        if number_of_bells!=0:
            body.append(number_of_bells)
        ViafloCommand.__init__(self, 4, body)
        
class ViafloAspirateCommand(ViafloCommand):
    def __init__(self, speed, stops, number_of_bells=0):
        body = [4, speed] + _short_num_to_bytes(stops)
        if number_of_bells!=0:
            body.append(number_of_bells)
        ViafloCommand.__init__(self, 4, body)

class ViafloDispenseCommand(ViafloCommand):
    def __init__(self, speed, stops, number_of_bells=0):
        body = [6, speed] + _short_num_to_bytes(stops)
        if number_of_bells!=0:
            body.append(number_of_bells)
        ViafloCommand.__init__(self, 4, body)

class ViafloPositionQueryCommand(ViafloCommand):
    def __init__(self):
        ViafloCommand.__init__(self, 4, [1])


def unpack_msg(raw_data, expected_seq_num):
    """
    Unpacks data and returns response message
    @return ViafloResponse 
    """
    # Extra check, look for invalid escape sequences in message
    # An invalid escape sequence would be a ESC followed by anything 
    # other than a ESC, STX, or ETX

    if re.search("\x1B[^\x02\x03\x1B]", raw_data):
        print "Invalid escape seqeunce found"
        #raise InvalidEscapeSequenceError("Invalid escape seqeunce found")

    # Replace escape sequences
    raw_data = re.sub("\x1B([\x02\x03\x1B])","\\1",raw_data)

    # Convert string data, into array of integers
    data = _convert_str_to_bytes(raw_data)

    if len(data) < 11:
        raise IncompleteResponseError("Incomplete response. Data was only %d long." % len(data))
    
    if data[0] != _STX: # Check for start-of-text
        print data
        raise IncompleteResponseError("Data began with %d, expected STX (%d)." % (data[0], _STX))
    if data[-1] != _ETX: # Check for end-of-text
        print data
        raise IncompleteResponseError("Data ended with %d, not ETX (%d)." % (data[-1], _ETX))

    data = data[1:-1] # Strip off STX, ETX

    # Perl code checks that response sequence number matches command sequence number
    seqnum = _bytes_to_num((data[3],data[4]))
    if seqnum != expected_seq_num:
        print data
        raise SeqNumMismatchError("Message # mismatch, got %d, expected %d" % (seqnum, expected_seq_num))        

    msg_type = _bytes_to_num((data[6], data[7]))
                             
    if msg_type == 1:
        return ViafloInfoResponse(data)
    elif msg_type == 4:
        return ViafloPipettorActionResponse(data)
    elif msg_type == 10:
        return ViafloVersionResponse(data)
    else:
        print 'Data length', len(data)
        print 'MSG data:', data
        raise InvalidMessageTypeError("Got message type of %d. Unable to process" % msg_type)


class ViafloPipette(object):
    def __init__(self):
        self._socket = None
        self._seq_num = 0
        self._bd_addr = None
        self._port = None

    def open(self, bd_addr, port):
        """
        Open specified device.
        """
        self._bd_addr = bd_addr
        self._port = port
        self._seq_num = 0
        
        self._sock = bluetooth.BluetoothSocket( bluetooth.RFCOMM )
        #self._sock.setttimeout(10.0) # 10s timeout for now.

        self._sock.connect( (self._bd_addr, self._port ))

        data = self._sock.recv(1024)
        
    def _recv_data(self):
        data = ""
        while (len(data) == 0) or (data[-1] != chr(_ETX)):
            data += self._sock.recv(1024)

        return data

    def send_msg(self, message, receive = True):
        """
        @param message bytes : Message must be converted into hex format
        @param receive bool : Whether to wait and receive data
        @return ViafloResponse : Response message
        """
        if not self._sock:
            raise NotConnectedException("Pipette isn't connected")

        self._seq_num += 1
        self._sock.send( message.pack(self._seq_num) )

        if not receive:
            return
        
        data = self._recv_data()

        return unpack_msg(data, self._seq_num)
        
    def close(self):
        self._sock.close()
        self._sock = None