magstripe.py

import serial
import sys
from threading import RLock
import time

##
# @mainpage SEN-08634 Magnetic Card Reader/Writer
# This script can be directly edited to accomplish interaction with the
# SEN-08634 Card Reader/Writer or it may be imported as a module.
# Sample Usage:
# <pre>
#    mag = MagStripe()
#    if mag.connect():
#        print("Successfully connected to the Card Reader/Writer")
#        mag.write_tracks('HELLO WORLD','012345689','987654321')
#        mag.peek()
#    else:
#        print("Unable to connect to the Card Reader/Writer")
# </pre>
#
# Sample Output: <pre>
# Successfully connected to the Card Reader/Writer
# (2)[Header Flag]
#  (25)[Track 1 Start]
# HELLO WORLD (3f)[Track End]
#  (3b)[Track  2 Start]
# 012345689 (3f)[Track End]
#  (3)[Trailer Flag]
# </pre>
# <a href="http://www.sparkfun.com/products/8634">SparkFun Product Page</a>


##
# This value controls how much data is read from the device at a time (in bytes)
MAX_READ_LENGTH = 255

##
# This class exposes the SEN-08634 Magnetic Card Reader/Writer as an object to
# accomplish reading and writing of data.
#
# Sample Usage:
# <pre>
#    mag = MagStripe()
#    if mag.connect():
#        print("Successfully connected to the Card Reader/Writer")
#        mag.write_tracks('HELLOWORLD','012345689','987654321')
#        mag.peek()
#    else:
#        print("Unable to connect to the Card Reader/Writer")
# </pre>
class MagStripe:
    ## Default serial port to use when no port is specified
    PORT="/dev/ttyUSB0"
    ## Default baud rate to use when no baud rate is specified
    BAUD=9600
    ## Timeout value to use when reading from a serial port
    TIMEOUT=2

    ## Synchonrizes access to class members
    lock = RLock()
    ## Represents the current state of the device
    connected = False
    ## Holds a reference to the underlying serial port
    serial = None
    ## These are protocol commands and markup
    commands = {'Warm Reset':'\x7F' ,
         'Ping':'\x1B' ,
         'About':'\x1B' ,
         'Red LED on':'\x58' ,
         'Red LED off':'\x59' ,
         'Green LED on':'\x48' ,
         'Green LED off':'\x49' ,
         'Yellow LED on':'\x29' ,
         'Yellow LED off':'\x28' ,
         'Buzzer on':'\x4A' ,
         'Arm to read':'\x44' ,
         'Read T1, Standard T1 Format':'\x54' ,
         'Read T2, Standard T2 Format':'\x4B' ,
         'Read T3, Standard T3 Format':'\x5F' ,
         'Arm to Write with RAW':'\x60' ,
         'Write T1,Standard T1 Format':'\x64' ,
         'Write T2, Standard T2 Format':'\x65' ,
         'Write T3, Standard T3 Format':'\x66' ,
         'Set Write Density for Specified Track':'\x67' ,
         'Set 210 BPI for T1&T3':'\x68' ,
         'Set 75 BPI for T1&T3':'\x69' ,
         'Set 210 BPI for T2':'\x6A' ,
         'Set 75 BPI for T2':'\x6B' ,
         'Set Coercivity High Only':'\x75' ,
         'Set Coercivity Low Only':'\x76' ,
         'Arm to Erase':'\x70' ,
         'Erase T1':'\x77' ,
         'Erase T2':'\x78' ,
         'Erase T3':'\x7D' ,
         'Set Communication Baud Rate':'\x7E' ,
         'Get firmware version':'\x39' ,
         'Ack':'\x06',
         'Nack':'\x15',
         'Start Heading':'\x01', #these are straight from the ascii table
         'Start Text':'\x02',
         'Stop Text':'\x03'
         }

    ##
    # Constructs an instance that's ready to start reading, but cannot be used
    # for writing until connect is called.
    # @param port number or serial device (/dev/ttyUSB0, etc)
    def __init__(self,port=PORT):
        self.serial = serial.Serial(port)
        self.serial.timeout = self.TIMEOUT
        self.serial.baudrate = self.BAUD

    ##
    # Used internally to read data from the serial device
    # @param timeout Number of times to attempt the read
    def _read(self,timeout = 5):
        data = None
        timeouts = 0
        self.lock.acquire()
        while data == None and timeouts < timeout:
            timeout = timeout + 1
            data = self.serial.read(MAX_READ_LENGTH)
        self.lock.release()
        return data

    ##
    # Used internally to write data to the serial device
    # @param value value to be written
    def _write(self,value):
        self.lock.acquire()
        self.serial.write(value)
        self.lock.release()

    ##
    # Tells the writer to get ready to write the specified tracks and writes
    # them as soon as a card is swiped.
    # This function performs no validation on the values that are passed in for the
    # various tracks, but the writer will fail to write if the data is invalid.
    # @param track1 The string to write to track 1
    # @param track2 The digits to write to track 2
    # @param track3 The digits to write to track 3
    # @note Sometimes it makes angry beeps even if it did write to the card.
    def write_tracks( self, track1, track2, track3 ):
        self.lock.acquire()
        self.write_command('Start Text')
        self._write( "%s%s\x3F;%s\x3f+%s\x3f\x03" % ('%',track1, track2, track3) )
        self.write_command('Stop Text')
        self.write_command('Start Heading')

    ##
    # Tests to see if the device is currently in the connected state with the
    # system.
    # @returns true if connected, false otherwise
    def is_connected(self):
        self.lock.acquire()
        conn = self.connected
        self.lock.release()
        return conn

    ##
    # Sets the internal state to connected
    def _connected(self,value=True):
        self.lock.acquire()
        self.connected = value
        self.lock.release()

    ##
    # Connects to the mag stripe reader and initializes it for further
    # communications.  If there is already a connection, this will disconnect
    # and reconnect.  Check the return value, if the device is ready to be used
    # this function returns true.
    # @returns true on successful connect, false on failure
    def connect(self):
        if self.is_connected():
            self.disconnect()
            self.lock.acquire()
            self.serial.open()
            self.lock.release()

        self.lock.acquire()
        #throw away anything that might be waiting to be read
        self.peek() #it seems to get into a funny state if we don't do this
        self.write_command('Warm Reset') #does not respond to this command, so dont read yet...
        self.peek() #it seems to get into a funny state if we don't do this
        self.write_command('Ping')
        read = self._read();
        for c in read:
            if c == self.commands['Ack']:
                self._connected()
        self.lock.release()
        return self.is_connected()

    ##
    # This will flush the underlying buffers and close the serial port.
    def disconnect(self):
        self.lock.acquire()
        self._connected(False)
        self.serial.flush()
        self.serial.close()
        self.lock.release()

    ##
    # Write the specified command to the Card Reader/Writer.
    # This is a convenience method, call it with 'Reset', 'About', 'Write',
    # etc.  For commands that take arguments, provide an optional value.
    # ex: <pre>
    # magstripe.write_command('Ping')
    # magstripe.peek()
    # </pre>
    def write_command(self, command, value = None):
        if command in self.commands:
            self.lock.acquire()
            self._write(self.commands[command])
            if value != None:
                self._write("%s"%value)
            self.lock.release()
        else:
            raise ValueError("Specified command [%s] is not in the command set.  Valid commands: %s" % (command,self.commands))

    ##
    # Peeks at any data that may have been recently read
    # and displays it in a semi-sane fashion.
    # Sample output:
    # <pre>
    # (2)[Header Flag]
    #  (25)[Track 1 Start]
    # HELLO WORLD (3f)[Track End]
    #  (3b)[Track  2 Start]
    # 012345689 (3f)[Track End]
    #  (3)[Trailer Flag]
    # </pre>
    def peek(self):
        response = mag._read()
        if not response:
            print("[no data waiting]")
        else:
            tokens = { '\x02': 'Header Flag', '\x03': 'Trailer Flag', '\x25': 'Track 1 Start', '\x3b': 'Track  2 Start', '\x2b':'Track 3 Start', '\x3F': 'Track End' }
            for t,v in tokens.items():
                response = response.replace(t,' (%x)[%s]\n ' % (ord(t),v))
            print response

if __name__ == "__main__":
    mag = MagStripe('/dev/ttyUSB0')#'/dev/null')
    if mag.connect():
        print("Successfully connected to the Card Reader/Writer")
        mag.write_tracks('HELLO WORLD','012345689','987654321')
        mag.peek();

        mag.disconnect()
    else:
        print("Unable to connect to the Card Reader/Writer")

# These things don't seem to do anything, though they are all supposed to be valid.
# (all attempted (of course) after a successful connect:
#        print("red led on command...")
#        mag.write_command( 'Red LED on' )
#        mag.peek()
#        print("yellow led on command...")
#        mag.write_command( 'Yellow LED on' )
#        mag.peek()
#        print("about command...")
#        mag.write_command( 'About' )
#        mag.peek()
#        print("Firmware version command...")
#        mag.write_command('Get firmware version')
#        mag.peek()
#        print("buzzer command...")
#        mag.write_command('Buzzer on')
#        mag.peek()