import lib
import logging
import time
"""
stick - implement a naive open source driver for Medtronic's
Carelink USB stick.
Please ask Medtronic for additional information on how to use the usb
stick.
Consumes a :ref:`link`, which allows us to debug everything on the
wire.
"""
log = logging.getLogger( ).getChild(__name__)
from errors import StickError, AckError, BadDeviceCommError
class BadCRC(StickError): pass
[docs] return lib.CRC8.compute(data)
class StickCommand(object):
[docs] """Basic stick command
Each command is used to talk to the usb stick.
The usb stick interprets the opcode, and then performs the function
associated with the opcode.
Altogether, the suite of opcodes that the stick responds to allows
you to debug and track all packets you are sending/receiving plus
allows you to send recieve commands to the pump, by formatting your
message into payloads with opcodes, and then letting the stick work
on what you've given it. It's kind of like a modem with this funky
binary interface.
"""
code = [ 0x00 ]
label = 'example stick command'
delay = .001
size = 64
def __str__(self):
code = ' '.join([ '%#04x' % op for op in self.code ])
return '{0}:{1}'.format(self.__class__.__name__, code)
def __repr__(self):
return '<{0:s}:size({1})>'.format(self, self.size)
def format(self):
[docs] self.data = data
def respond(self, raw):
[docs] if len(raw) == 0:
log.error("ACK is zero bytes!")
# return False
raise AckError("ACK is 0 bytes:\n%s" % lib.hexdump(raw))
commStatus = raw[0]
# usable response
assert commStatus == 1, ('commStatus: %02x expected 0x1' % commStatus)
status = raw[1]
# status == 102 'f' NAK, look up NAK
if status == 85: # 'U'
return raw[:3], raw[3:]
assert False, ("NAK!!\n%s" % lib.hexdump(raw[:3]))
class ProductInfo(StickCommand):
[docs] """Get product info from the usb device.
Useful for identifying
what kind of usb stick you've got; there are a few different kinds.
Eg, European vs US regulatory domains require different frequencies for compliance.
"""
code = [ 4 ]
SW_VER = 16
label = 'usb.productInfo'
rf_table = { 001: '868.35Mhz' ,
000: '916.5Mhz' ,
255: '916.5Mhz' }
iface_key = { 3: 'USB',
1: 'Paradigm RF' }
@classmethod
def decodeInterfaces( klass, L ):
[docs] n, tail = L[ 0 ], L[ 1: ]
interfaces = [ ]
for x in xrange( n ):
i = x*2
k, v = tail[i], tail[i+1]
interfaces.append( ( k, klass.iface_key.get( v, 'UNKNOWN' ) ) )
return interfaces
@classmethod
def decode( klass, data ):
[docs] return {
'rf.freq' : klass.rf_table.get( data[ 5 ], 'UNKNOWN' )
, 'serial' : str( data[ 0:3 ]).encode( 'hex' )
, 'product.version' : '{0}.{1}'.format( *data[ 3:5 ] )
, 'description' : str( data[ 06:16 ] )
, 'software.version' : '{0}.{1}'.format( *data[ 16:18 ] )
, 'interfaces' : klass.decodeInterfaces( data[ 18: ] )
}
_test_ok = bytearray( [
] )
def parse(self, data):
[docs] """
#>>>
"""
return self.decode(data)
class InterfaceStats(StickCommand):
[docs] """Abstract stats decoder.
"""
code = [ 5 ]
INTERFACE_IDX = 19
label = 'usb.interfaceStats'
@classmethod
def decode( klass, data):
[docs] """
Decode interface stats. The stick exposes 6 counters to monitor errors,
crcs, naks, timeouts, rx, and tx packets. Very useful for debugging.
"""
return {
'errors.crc' : data[ 0 ]
, 'errors.sequence' : data[ 1 ]
, 'errors.naks' : data[ 2 ]
, 'errors.timeouts' : data[ 3 ]
, 'packets.received': lib.BangLong( data[ 4: 8 ] )
, 'packets.transmit': lib.BangLong( data[ 8:12 ] )
}
def parse(self, data):
[docs] """
#>>>
"""
return self.decode(data)
class UsbStats(InterfaceStats):
[docs] """Count of packets and stats on the usb side of the stick."""
code = [ 5, 1 ]
class RadioStats(InterfaceStats):
[docs] """Count of packets and stats on the radio side of the stick."""
code = [ 5, 0 ]
class SignalStrength(StickCommand):
[docs] """This seems to be required to initialize communications with the
usb stick. Also, you should wait until a minimum threshold is
reached.
"""
code = [ 6, 0 ]
def parse(self, data):
[docs] """
#>>>
"""
# result[0] is signal strength
self.value = int(data[0])
log.info('%r:readSignalStrength:%s' % (self, int(data[0])))
return int(data[0])
class LinkStatus(StickCommand):
[docs] """Basic ACK type of command.
Used to poll the modem's radio buffer. When the radio buffer is
full, we can download a packet from the buffer. Otherwise, we need
to be mindful of the state the radio is in. This opcode tells you
the current state of the radio/stick.
"""
code = [ 0x03 ]
reasons = ['OK']
def __str__(self):
extra = ''
size = getattr(self, 'size', None) or '??'
extra += "size=%s" % size
if getattr(self, 'error', False):
extra += '{0}:error:{1}:reason:{2}'.format(self.__class__.__name__, self.error, str(self.reasons))
base = super(type(self), self).__str__( )
return '{0}:status:{1}'.format(base, extra)
def record_error(self, result):
[docs] self.error = True
self.ack = result[0] # 0 indicates success
# believe success = result[1] # 'U' or 'f'
self.status = result[2]
lb, hb = result[3], result[4]
self.size = lib.BangInt((lb, hb))
if self.ack == 0 and (self.status & 0x1) > 0:
self.error = False
self.set_reason(self.ack)
def set_reason(self, status):
[docs] reasons = [ ]
if (status & 0x2) > 0:
reasons.append('STATUS: receive in progress!')
if (status & 0x4) > 0:
reasons.append('STATUS: transmit in progress!')
if (status & 0x8) > 0:
reasons.append('STATUS: interface error!')
if (status & 0x10) > 0:
reasons.append('STATUS: receive overflow!')
if (status & 0x20) > 0:
reasons.append('STATUS: transmit overflow!')
if (status & 0x1) > 0:
reasons.append('STATUS: OK')
msg = '\n'.join(map(str, [ self, '|'.join(reasons) ]))
log.info(msg)
self.reasons = reasons
def parse(self, result):
[docs] """
#>>>
"""
self.record_error(result)
if self.ack != 0:
log.error("readStatus: non-zero status: %02x" % self.ack)
# should this trigger a retry, and if so where?
# should the other usb commands also trigger these retries? and at the
# same points?
raise AckError("readStatus: non-zero status: %02x" % self.ack)
if self.error is not True:
return self.size
return 0
class ReadRadio(StickCommand):
[docs] """
Read buffer from the radio.
Downloads a packet from the radio buffer.
"""
code = [ 0x0C, 0x00 ]
dl_size = 0
size = 64
def __init__(self, size):
self.size = size
self.dl_size = size
packet = [12, 0, lib.HighByte(size), lib.LowByte(size)]
if size < 64 and size != 15:
log.error('size (%s) is less than 64 and not 15, which may cause an error.' % size)
self.size = 64
self.code = packet + [ CRC8(packet) ]
def __str__(self):
return '{0}:size:{1}'.format(self.__class__.__name__, self.dl_size)
def __repr__(self):
return '<{0:s}>'.format(self)
def format(self):
[docs] if len(raw) == 0:
log.error("ReadRadio ACK is zero bytes!")
# return False
raise AckError("ACK is 0 bytes: %s" % lib.hexdump(raw))
log.info('readData validating remote raw[ack]: %02x' % raw[0])
log.info('readData; foreign raw should be at least 14 bytes? %s %s' % (len(raw), len(raw) > 14))
log.info('readData; raw[retries] %s' % int(raw[3]))
dl_status = int(raw[0])
if dl_status != 0x02: # this differs from the others?
raise BadDeviceCommError("bad dl raw! %r" % raw)
assert (int(raw[0]) == 2), repr(raw)
return raw[:1], raw
def parse(self, raw):
[docs] """
Detect BadCRC here. Also, look for eod set.
"""
"""
log.info('readData validating remote raw[ack]: %02x' % raw[0])
log.info('readData; foreign raw should be at least 14 bytes? %s %s' % (len(raw), len(raw) > 14))
log.info('readData; raw[retries] %s' % int(raw[3]))
dl_status = int(raw[0])
if dl_status != 0x02: # this differs from the others?
raise BadDeviceCommError("bad dl raw! %r" % raw)
assert (int(raw[0]) == 2), repr(raw)
# raw[1] != 0 # interface number !=0
# raw[2] == 5 # timeout occurred
# raw[2] == 2 # NAK
# raw[2] # should be within 0..4
log.info("readData ACK")
lb, hb = raw[5] & 0x7F, raw[6]
self.eod = (raw[5] & 0x80) > 0
"""
lb, hb = raw[5] & 0x7F, raw[6]
self.eod = (raw[5] & 0x80) > 0
resLength = lib.BangInt((lb, hb))
# we don't really care about the length
#assert resLength < 64, ("cmd low byte count:\n%s" % lib.hexdump(raw))
data = raw[13:13+resLength]
self.packet = data
log.info('%s:eod:found eod (%s)' % (self, self.eod))
log.info('found packet len(%s), link expects(%s)' % (len(self.packet), resLength))
assert len(data) == resLength
head = raw[13:]
crc = raw[-1]
# crc check
if crc == 0 and len(data) > 1:
log.warn('bad zero CRC?')
expected_crc = CRC8(data)
if crc != expected_crc:
msg = ':'.join( [ 'ReadRadio:BAD ACK:found raw[crc]: %#04x' % (crc),
'expected_crc(data): %#04x' % (expected_crc),
'raw:\n%s\n' % (lib.hexdump(raw)),
'head:\n%s\n' % (lib.hexdump(head)),
'data:\n%s\n' % (lib.hexdump(data)) ] )
log.info(msg)
log.info("XXX:IGNORE:BadCRC:returning empty message, sleep .100, avoid errors.")
time.sleep(.100)
return bytearray( )
raise BadCRC(msg)
assert crc == expected_crc
return data
class TransmitPacket(StickCommand):
[docs] """Format a packet to send on the radio.
This commands formats a packet from usb, and shoves it into the
radio buffer.
The radio buffer is broadcast "over the air" so that any device
sensitive to the packets you sent will respond accordingly
(probably sending data back).
For this reason, the serial number of the device you'd like to talk
to is formatted into the packet.
"""
code = [ 1, 0, 167, 1 ]
head = [ 1, 0, 167, 1 ]
# wraps pump commands
def __init__(self, command):
self.command = command
self.params = command.params
self.code = command.code
self.retries = command.retries
self.serial = command.serial
# self.delay = command.effectTime
def __str__(self):
if getattr(self, 'command', False):
return '{0}:{1:s}'.format(self.__class__.__name__, self.command)
code = ' '.join([ '%#04x' % op for op in self.head ])
return '{0}:{1}'.format(self.__class__.__name__, code)
def __repr__(self):
return '<{0:s}>'.format(self)
def calcRecordsRequired(self):
[docs] return self.command.calcRecordsRequired( )
def format(self):
[docs] code = self.command.code
params = self.params
if code != 93 or params[0] != 0:
ack, body = super(type(self), self).respond(raw)
return ack, body
return (bytearray(raw), bytearray(raw))
def parse(self, results):
[docs] return results
#self.checkAck(results)
class Stick(object):
[docs] """
The carelink usb stick acts like a buffer.
It has a variety of commands providing synchronous IO, eg, you may
generally perform a read immediately after writing to it, and expect a
response.
The commands operate on a local buffer used to facilitate exchanging
messages over RF with the pump. RF communication with the pump
happens asynchronously, requiring us to go through 3 separate
phases for each message we'd like to exchange with the pumps:
* transmit - send commmand
* poll_size - loop
* download - loop
Each command is usually only 3 bytes.
The protocol offers some facility for detecting and recovering
from inconsistencies in the underlying transport of data, however,
we are somwhat ignorant of them. The tricky bits are exactly how
to recover from, eg CRC, errors that can occur.
The "shape" and timing of these loops seem to mostly get the job
done.
The Stick object provides a bunch of useful methods, that given a link,
will represent the state of one active usb stick.
"""
link = None
def __init__(self, link):
self.link = link
self.command = None
self._download_i = False
def __str__(self):
s = [ self.__class__.__name__,
'transmit[{}]' .format(str(getattr(self, 'transmit', None))),
'reader[{}]' .format(str(getattr(self, 'reader', None))),
'download_i[{}]'.format(str(getattr(self, '_download_i', None))),
'status[{}]' .format(repr(getattr(self, 'last_status', None))),
'poll_size[{}]' .format(str(getattr(self, '_poll_size', None))),
'poll_i[{}]' .format(str(getattr(self, '_poll_i', None))),
'command[{}]' .format(repr(getattr(self, 'command', None))),
]
return ' '.join(s)
def __repr__(self):
return '<{0}>'.format(str(self))
def process(self):
[docs] """
Working with the usb stick typically follows a pretty routine process:
1. send our opcode, get a response
2. use some custom logic, per opcode to respond to the stick's reponse
3. parse the response from that, return result
This has to be done for each opcode.
"""
msg = ':'.join(['PROCESS', 'START'
] + map(str, [ self.timer.millis( ), self.command]))
log.info(msg)
log.info('link %s processing %s)' % ( self, self.command ))
"""
self.link.write(self.command.format( ))
log.debug('sleeping %s' % self.command.delay)
time.sleep(self.command.delay)
size = max(64, self.command.size)
raw = bytearray(self.link.read(size))
"""
raw = self.send_force_read( )
if not raw or len(raw) == 0:
log.info('process zero length READ, try once more sleep .010')
time.sleep(.010)
raw = bytearray(self.link.read(self.command.size))
ack, response = self.command.respond(raw)
info = self.command.parse(response)
log.info('finished processing {0}, {1}'.format(self.command, repr(info)))
msg = ':'.join(['PROCESS', 'END'
] + map(str, [ self.timer.millis( ), self.command]))
log.info(msg)
return info
def query(self, Command):
[docs] """
query - simplify the process of working with the stick, pass your command, get the result
"""
self.command = command = Command( )
return self.process( )
def product_info(self):
[docs] """
Get the product info from the connected stick.
"""
return self.query(ProductInfo)
def interface_stats(self):
[docs] """
debug both sets of interface stats.
"""
return {'usb': self.usb_stats( ), 'radio': self.radio_stats( ) }
def usb_stats(self):
[docs] """
just get usb stats.
"""
return self.query(UsbStats)
def radio_stats(self):
[docs] """
just get radio stats.
"""
return self.query(RadioStats)
def signal_strength(self):
[docs] """
just get signal strength from connected stick
"""
return self.query(SignalStrength)
def poll_size(self):
[docs] """
query how many bytes are waiting in the radio buffer, ready to be downloaded
There seem to be a few sweet spots, where you want to download the data.
"""
size = 0
start = time.time()
i = 0
log.debug('%r:STARTING POLL PHASE:attempt:%s' % (self, i))
#while size == 0 and size < 64 and time.time() - start < 1:
while size == 0 and time.time() - start < 1:
self._poll_i = i
self._poll_size = size
log.debug('%r:poll:attempt:%s' % (self, i))
size = self.read_status( )
self._poll_size = size
if size == 0:
log.debug('poll zero, sleeping in POLL, .100')
time.sleep(.100)
i += 1
log.info('%s:STOP POLL after %s attempts:size:%s' % (self, i, size))
self._poll_size = size
self._poll_i = False
return size
def read_status(self):
[docs] """
Get current link status.
"""
# log.debug('read_status')
result = self.query(LinkStatus)
self.last_status = self.command
return result
def old_download_packet(self, size):
[docs] """
Naive version of downloading a packet.
Didn't quite work right.
"""
log.info("download_packet:%s" % (size))
self.command = ReadRadio(size)
packet = self.process( )
return packet
def send_force_read(self, retries=5, timeout=1):
[docs] """
Pretty simple, try really hard to ensure that we've sent our bytes, and we
get a response.
This is probably overkill, but seems to get the job done.
"""
#
# so the behavior of a read_radio should probably be similar to
# poll_size??
reader = self.command
read_size = 64
size = reader.size
start = time.time( )
raw = bytearray( )
for attempt in xrange(retries):
log.info(' '.join([
'send_force_read: attempt {0}/{1}'.format(attempt, retries),
'send command,',
'read until we get something within some timeout']))
log.info('link %s sending %s)' % ( self, reader ))
self.link.write(reader.format( ))
log.debug('sleeping %s' % reader.delay)
time.sleep(reader.delay)
raw = bytearray(self.link.read(size))
if len(raw) == 0:
log.info('zero length READ, try once more sleep .250')
time.sleep(.250)
raw = bytearray(self.link.read(self.command.size))
if len(raw) != 0:
log.info(' '.join(['quit send_force_read,',
'found len:', str(len(raw)),
'expected', str(size),
'after', str(attempt), 'attempts']))
return raw
log.critical(' '.join([ "FAILED TO DOWNLOAD ANYTHING,",
"after %s " % (attempt),
"expected:%s" % (size) ]))
assert not raw
def download_packet(self, size):
[docs] """
This is the tricky bit, where we stroke the radio and hope it gives us a
buffer full of data.
"""
log.info("%s:download_packet:%s" % (self, size))
# XXX: this is the tricky bit
original_size = size
self.command = reader = ReadRadio(size)
self.reader = reader
msg = ':'.join(['PROCESS', 'START'
] + map(str, [ self.timer.millis( ), self.command]))
log.info(msg)
raw = self.send_force_read( )
# return
# packet = self.process( )
# return packet
# copy pasted from process
"""
log.info('link %s processing %s)' % ( self, self.command ))
# self.link.process(command)
self.link.write(self.command.format( ))
log.debug('sleeping %s' % self.command.delay)
time.sleep(self.command.delay)
size = max(64, self.command.size)
raw = bytearray(self.link.read(size))
"""
# if len(raw) == 0:
if not raw:
log.info('zero length READ, try once more sleep .500')
time.sleep(.500)
raw = bytearray(self.link.read(self.command.size))
try:
ack, response = self.command.respond(raw)
info = self.command.parse(response)
msg = ':'.join(['PROCESS', 'END'
] + map(str, [ self.timer.millis( ), self.command]))
log.info(msg)
return info
except BadDeviceCommError, e:
log.critical("download_packet:%s:ERROR:%s:ACK!?" % (self, e))
log.info("we failed to pass %s ACK!?" % (self.command))
log.info('expected size was: %s' % original_size)
status = LinkStatus( )
if original_size < 64:
#size = self.read_status( )
#size = self.poll_size( )
log.info('XXX:JUST a bit more READ new size: %s, sleep .100' % original_size)
self.link.write(status.format( ))
time.sleep(.100)
raw = bytearray(self.link.read(64))
ack, response = reader.respond(raw)
info = reader.parse(response)
return info
ack, body = status.respond(raw)
size = status.parse(body)
log.info('attempt another read')
info = None
raw = bytearray(self.link.read(size))
if len(raw) == 0:
log.info('NESTED zero length READ, try once more sleep .100')
time.sleep(.100)
raw = bytearray(self.link.read(self.command.size))
ack, body = status.respond(raw)
info = self.command.parse(body)
log.info('finished processing {0}, {1}'.format(self.command, repr(info)))
return info
def download(self, size=None):
[docs] """
Theory is to download anything and everything available off the radio
buffer, and to wait if necessary.
"""
eod = False
results = bytearray( )
ailing = 0
i = 0
log_head = 'download(attempts[{}])'
expecting = 'download(attempts[{}],expect[{}])'
stats = '{}:download(attempts[{}],expect[{}],results[{}]:data[{}])'
expect_eod = False
log.info('download:start:%s' % i)
data = bytearray( )
while not eod:
i += 1
self._download_i = i
data = bytearray( )
if size is None:
log.info("%s:begin first poll first sleep .250" % (stats.format(self, i, 0,
len(results), len(data))))
time.sleep(.250)
size = self.poll_size( )
log.info("%s:end first poll" % (stats.format(self, i, size,
len(results), len(data))))
if size == 0:
if i % 3 == 0:
time.sleep(1.5)
time.sleep(1.5)
size = self.poll_size( )
"""
if size == 0:
# if size == 0 and i > 1:
log.info("%s:zero poll size, sleep .500 try again" % ( \
stats.format(self, i, size, len(results), len(data))))
size = self.poll_size( )
time.sleep(.500)
"""
if size == 0 and i > 1:
log.warn("%s:BAD AILING" % (stats.format(self, i, size,
len(results), len(data))))
ailing = ailing + 1
if ailing > 5:
break
continue
# break
elif ailing > 0:
ailing = ailing - 1
log.info("%s:proceed to download packet" % (stats.format(self, i, size,
len(results), len(data))))
#time.sleep(.100)
data = self.download_packet(size)
expect_eod = False
if data:
results.extend(data)
expect_eod = self.command.eod
log.info("%s:adding segment" % (stats.format(self, i, size,
len(results), len(data))))
else:
log.info("%s:no data, try again sleep .400" % (stats.format(self, i, size,
len(results), len(data))))
time.sleep(.400)
# eod = expect_eod and size < 15
eod = expect_eod
# or size < 15
if not eod:
log.info("%s:no eod, sleep .200 try again" % (stats.format(self, i, size,
len(results), len(data))))
time.sleep(.200)
size = self.poll_size( )
log.info("%s:DONE" % (stats.format(self, i, size,
len(results), len(data))))
self._download_i = False
# self.reader = None
return results
def clear_buffer(self):
[docs] """
An alternative download solution. This can be helpful in
scenarios where a prior run seems crashed your process, but the
radio is still transmitting and receiving data. Running this
method collects data from the radio until it's done receiving,
more or less, at which point you should be free to try again.
"""
bad = bytearray( )
raw = bytearray( )
for attempt in xrange( 3 ):
segments = [ ]
segs_vs_raw = 'segments[{0}],total_segments[{1}]:raw[{2}]'
seg_stats = ( len(segments), sum(map(len, segments)), len(raw) )
log_detail = segs_vs_raw.format(*seg_stats)
log_head = "XXX:clear_buffer[attempt][%s]" % (attempt)
log.debug('INTERFACE STATS:\n%s' % lib.pformat(self.interface_stats( )))
log.info(":".join([ log_head, log_detail, "BEGIN ", "first poll" ]))
size = self.poll_size( )
end_poll = ':'.join( [ log_head, log_detail,
"END first poll %s" % (size),
"SHOULD DOWNLOAD ", str(size != 0) ] )
log.info(end_poll)
if size == 0:
break
seg_stats = ( len(segments), sum(map(len, segments)), len(raw) )
log_detail = segs_vs_raw.format(*seg_stats)
log.info("%s:download the size? %s:%s" % (log_head, size, log_detail))
while size > 14:
seg_stats = ( len(segments), sum(map(len, segments)), len(raw) )
log_detail = segs_vs_raw.format(*seg_stats)
log_head = "XXX:clear_buffer[attempt][%s]" % (attempt)
log.info( ':'.join([ "%s size:%s" % (log_head, size),
log_detail,
"clear_buffer BUFFER self.download( )" ]))
try:
segment = self.download( )
raw.extend(segment)
segments.append(segment)
seg_stats = ( len(segments), sum(map(len, segments)), len(raw) )
log_detail = segs_vs_raw.format(*seg_stats)
log.info(":".join([ "%s:tx:found" % (log_head),
log_detail,
'len(raw)', str(len(raw)),
'expected', str(size),
'len(segment)', str(len(segment)) ]))
except BadCRC, e:
seg_stats = ( len(segments), sum(map(len, segments)), len(raw) )
log_detail = segs_vs_raw.format(*seg_stats)
log.critical('%s:IGNORING:%s:%s' % (log_head, log_detail, e))
seg_stats = ( len(segments), sum(map(len, segments)), len(raw) )
log_detail = segs_vs_raw.format(*seg_stats)
log.info(':'.join([ "%s downloaded %s segment" % (log_head, len(raw)),
log_detail,
"RAW:\n%s" % lib.hexdump(raw) ]))
size = self.poll_size( )
log.debug("INTERFACE STATS:\n%s" % lib.pformat(self.interface_stats( )))
if raw:
return raw
if size == 0:
log.info("\n".join([ "%s:END:no data:INTERFACE STATS" % (log_head),
lib.pformat(self.interface_stats( )) ]))
def transmit_packet(self, command):
[docs] """
Address a pump with a request.
"""
packet = TransmitPacket(command)
self.command = packet
self.transmit = packet
log.info('transmit_packet:write:%r' % (self.command))
result = self.process( )
return result
def open(self):
[docs] """
Open and get signal strength so everything is ready to go.
"""
self.link.baudrate = 9600
self.timer = lib.Timer( )
for attempt in xrange( 3 ):
try:
msg = ':'.join(['PROCESS', 'OPEN', str(self.timer.millis( ))] )
log.info(msg)
log.info('%s' % self.product_info( ))
log.info('%s' % self.product_info( ))
log.info('get signal strength of %s' % self)
signal = 0
while signal < 50:
signal = self.signal_strength( )
log.info('we seem to have found a nice signal strength of: %s' % signal)
return True
except AckError, e:
log.info('failed:(%s):\n%s' % (attempt, e))
@staticmethod
def decode_hex (msg, Candidate):
[docs] candidate = Candidate( )
raw = lib.hexbytes(msg)
ack, resp = candidate.respond(raw)
result = candidate.parse(resp)
return result
if __name__ == '__main__':
import doctest
doctest.testmod( )
import sys
port = None
port = sys.argv[1:] and sys.argv[1] or False
if not port:
print "usage:\n%s /dev/ttyUSB0" % sys.argv[0]
sys.exit(1)
import link
from pprint import pformat
logging.basicConfig(stream=sys.stdout, level=logging.DEBUG)
log.info("howdy! I'm going to take a look at your carelink usb stick.")
stick = Stick(link.Link(port))
stick.open( )
log.info('test fetching product info %s' % stick)
log.info(pformat(stick.product_info( )))
log.info('get signal strength of %s' % stick)
signal = 0
while signal < 50:
signal = stick.signal_strength( )
log.info('we seem to have found a nice signal strength of: %s' % signal)
log.info("""
at this point, we could issue remote commands to a medical
device, let's inspect the interfaces""".strip( ))
#log.info(pformat(stick.usb_stats( )))
#log.info(pformat(stick.radio_stats( )))
log.info(pformat(stick.interface_stats( )))
"""
size = stick.poll_size( )
log.info("can we poll the size? %s" % (size))
if size > 14:
log.info("DOWNLOADING %s TO CLEAR BUFFER" % size)
log.info('\n'.join(["can we download ?", lib.hexdump(stick.download( ))]))
"""
log.info("CLEAR BUFFERS")
extra = stick.clear_buffer( )
if extra:
log.info(lib.hexdump(extra))
else:
log.info("NO PENDING BUFFER")
log.info("DONE CLEARING BUFFERS")
log.info("INTERFACE STATS:\n%s" % pformat(stick.interface_stats( )))
log.info("howdy! all done looking at the stick")
#####
# EOF