File content as of revision 5:90b73268e270:

#!/usr/bin/python
#
# This is a very simple test driver for demonstrating the StreamDownloader.

import pexpect
import sys
import time
import select
import re
import struct

class StreamDownloaderClient(object):
    """docstring for StreamDownloaderClient"""
    def __init__(self, bluetoothAddr, length):
        super(StreamDownloaderClient, self).__init__()
        self.bluetoothAddr = bluetoothAddr

        self.con = pexpect.spawn('gatttool -b ' + bluetoothAddr + ' --interactive -t random')
        self.con.expect('\[LE\]>', timeout=600)

        self.con.sendline('connect')
        self.con.expect('\[CON\]\[.+\]\[LE\]>')
        print('connected')

        # ensure that we've got the correct primary service available
        self.con.sendline('primary')
        self.con.expect('\[CON\]\[.+\]\[LE\]>')
        self.con.expect('\[CON\]\[.+\]\[LE\]>')

        emptyString = re.compile('^\s*$')
        for line in self.con.before.decode('utf-8').split('\r\n'):
            m = re.match('^attr handle: (0x[\dabcdef]{4}).*adc710c2-acdc-4bf5-8244-3ceaaa0f87f5.*', line)
            if m:
                break
        if not m:
            self.disconnect()

        self.serviceHandle = m.group(1)
        print('discovered service handle as ' + self.serviceHandle)

        self.discoverCharHandles()
        self.sendFileInfoBlock(length)
        for blocknum in range(64):
            self.sendFileDataBlock(blocknum)

        self.disconnect()

    def discoverCharHandles(self):
        self.con.sendline('characteristics ' + self.serviceHandle)
        self.con.expect('\[CON\]\[.+\]\[LE\]>')
        self.con.expect('\[CON\]\[.+\]\[LE\]>')
        for line in self.con.before.decode('utf-8').split('\r\n'):
            uuidRe = re.compile('.* char value handle: (0x[\dabcdef]{4}).*uuid: adc710c2\-(\w{4}).*')
            m = uuidRe.match(line)
            if m:
                if m.group(2) == 'acdf':
                    self.transferFileInfoHandle = m.group(1)
                    print("transferFileInfoHandle: " + self.transferFileInfoHandle)
                elif m.group(2) == 'ace0':
                    self.transferFileBlockHandle = m.group(1)
                    print("transferFileBlockHandle: " + self.transferFileBlockHandle)

    def sendFileInfoBlock(self, length):
        print("setup fileInfoBlock for transferring {} bytes".format(length))

        writeCommand = 'char-write-req ' + self.transferFileInfoHandle + ' '
        for c in struct.pack('<HH', length, 0):
            writeCommand = writeCommand + '{:02x}'.format(c)
        self.con.sendline(writeCommand)
        self.con.expect('\[CON\]\[.+\]\[LE\]> Characteristic value was written successfully', timeout=200)

    def sendFileDataBlock(self, blocknum):
        print("will send command for block {}".format(blocknum))

        writeCommand = 'char-write-req ' + self.transferFileBlockHandle + ' '
        for c in struct.pack('<HBBBBBBBBBBBBBBBB', blocknum, 0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15):
            writeCommand = writeCommand + '{:02x}'.format(c)
        self.con.sendline(writeCommand)
        self.con.expect('\[CON\]\[.+\]\[LE\]> Characteristic value was written successfully', timeout=200)

    def disconnect(self):
        self.con.sendline('disconnect')
        self.con.expect('\[\s+\]\[.+\]\[LE\]>')

def main():
    bluetoothAddr = "CC:59:FD:D8:3B:A9" # update as necessary

    target = StreamDownloaderClient(bluetoothAddr, 1024)

if __name__ == "__main__":
    main()