import logging
import time
import lib
"""
Implementation and decoding of lots of commands.
Each command inherits from :py:`BaseCommand`, which takes care of the
basic logic for informing the stick if we have recieved all the data
we expect to recieve.
Many commands are supported by Medtronic but not listed here.
Examples would include setting profiles and rates.
(One theory is that these commands are turned into setters with
correct arguments.)
"""
log = logging.getLogger( ).getChild(__name__)
def CRC8(data):
return lib.CRC8.compute(data)
class BaseCommand(object):
[docs] code = 0x00
descr = "(error)"
retries = 2
timeout = 3
params = [ ]
bytesPerRecord = 0
maxRecords = 0
effectTime = 0
responded = False
def __init__(self, code, descr, *args):
self.code = code
self.descr = descr
self.params = [ ]
def done(self):
[docs] found = len(self.data or [ ])
expect = int(self.maxRecords * self.bytesPerRecord)
expect_size = "found[{}] expected[{}]".format(found, expect)
log.info("%s:download:done?explain=%s" % (self, expect_size))
return found >= expect
def format(self):
[docs] self.data = data
self.getData( )
self.responded = True
def hexdump (self):
[docs] return lib.hexdump(self.data)
class PumpCommand(BaseCommand):
[docs] #serial = '665455'
#serial = '206525'
serial = '208850'
params = [ ]
bytesPerRecord = 64
maxRecords = 1
retries = 2
effectTime = .500
data = bytearray( )
__fields__ = ['maxRecords', 'code', 'descr',
'serial', 'bytesPerRecord', 'retries', 'params']
def __init__(self, **kwds):
for k in self.__fields__:
value = kwds.get(k, getattr(self, k))
setattr(self, k, value)
self.allocateRawData( )
self.data = bytearray( )
self.name = self.log_name( )
def log_name(self, prefix=''):
[docs] return prefix + '{}.data'.format(self.__class__.__name__)
def save(self, prefix=''):
[docs] name = '{}'.format(self.log_name(prefix))
handle = open(name, 'wb')
handle.write(self.data)
handle.close( )
def __str__(self):
if self.responded:
return '{}:size[{}]:data:{}'.format(self.__class__.__name__,
self.size, repr(self.getData( )))
return '{}:data:unknown'.format(self.__class__.__name__)
def __repr__(self):
return '<{0}>'.format( self)
def getData(self):
[docs] return self.data
def allocateRawData(self):
[docs] self.size = self.bytesPerRecord * self.maxRecords
def format(self):
[docs] length = self.bytesPerRecord * self.maxRecords
i = length / 64
j = length % 64
if j > 0:
return i + 1
return i
class ManualCommand(PumpCommand):
[docs] def __init__(self, **kwds):
self.name = kwds.get('name', self.__class__.__name__)
super(type(self), self).__init__(**kwds)
self.kwds = kwds
self.name = kwds.get('name', self.__class__.__name__)
def __str__(self):
if self.responded:
return '{}:{}:size[{}]:'.format(self.name, self.kwds,
self.size)
return '{}:{}:data:unknown'.format(self.name, self.kwds)
def log_name(self, prefix=''):
[docs] return prefix + '{}.data'.format(self.name)
def __repr__(self):
return '<{0}>'.format(self)
def getData(self):
[docs] return self.hexdump( )
class PowerControl(PumpCommand):
[docs] """
>>> PowerControl(serial='665455').format() == PowerControl._test_ok
True
"""
_test_ok = bytearray( [ 0x01, 0x00, 0xA7, 0x01, 0x66, 0x54, 0x55, 0x80,
0x02, 0x55, 0x00, 0x00, 0x00, 0x5D, 0xE6, 0x01,
0x0A, 0xA2 ] )
code = 93
descr = "RF Power On"
params = [ 0x01, 0x0A ]
retries = 0
maxRecords = 0
#timeout = 1
# effectTime = 7
effectTime = 17
def __init__(self, minutes=None, **kwds):
if minutes is not None:
self.minutes = int(minutes)
kwds['params'] = [ 0x01, self.minutes ]
super(PowerControl, self).__init__(**kwds)
class PowerControlOff(PowerControl):
[docs] """
Here's an example where arguments clearly modify behavior.
"""
params = [ 0x00, 0x00 ]
# MMPump???/ CMD_???????? 69 0x45 ('E') ??
class PumpExperiment_OP69 (PumpCommand):
[docs] code = 69
# MMPump???/ CMD_???????? 70 0x46 ('F') ??
class PumpExperiment_OP70 (PumpCommand):
[docs] code = 70
# MMPump???/ CMD_???????? 71 0x47 ('G') ??
class PumpExperiment_OP71 (PumpCommand):
[docs] code = 71
# MMPump???/ CMD_???????? 72 0x48 ('H') ??
class PumpExperiment_OP72 (PumpCommand):
[docs] code = 72
# MMPump???/ CMD_???????? 73 0x49 ('I') ??
class PumpExperiment_OP73 (PumpCommand):
[docs] code = 73
# MMPump???/ SelectBasalProfile 74 0x4a ('J') OK
class SelectBasalProfile (PumpCommand):
[docs] code = 74
class SelectBasalProfileSTD (SelectBasalProfile):
[docs] params = [ 0 ]
class SelectBasalProfileA (SelectBasalProfile):
[docs] params = [ 1 ]
class SelectBasalProfileB (SelectBasalProfile):
[docs] params = [ 2 ]
# MMPump???/ CMD_???????? 75 0x4b ('K') ??
class PumpExperiment_OP75 (PumpCommand):
[docs] code = 75
class TempBasal(PumpCommand):
[docs] """
"""
code = 76
descr = "Set temp basal"
params = [ 0x00, 0x00, 0x00 ]
retries = 0
#maxRecords = 0
#timeout = 1
class SetSuspend(PumpCommand):
[docs] code = 77
descr = "Set Pump Suspend/Resume status"
params = [ ]
retries = 2
maxRecords = 1
class PumpSuspend(SetSuspend):
[docs] descr = "Suspend pump"
params = [ 1 ]
class PumpResume(SetSuspend):
[docs] descr = "Resume pump (cancel suspend)"
params = [ 0 ]
class KeypadPush(PumpCommand):
[docs] code = 91
descr = "Press buttons on the keypad"
params = [ ]
retries = 1
maxRecords = 0
@classmethod
def ACT(klass, **kwds):
[docs] return klass(params=[0x02], **kwds)
@classmethod
def ESC(klass, **kwds):
[docs] return klass(params=[0x01], **kwds)
@classmethod
def DOWN(klass, **kwds):
[docs] return klass(params=[0x04], **kwds)
@classmethod
def UP(klass, **kwds):
[docs] return klass(params=[0x03], **kwds)
@classmethod
def EASY(klass, **kwds):
[docs] return klass(params=[0x00], **kwds)
def PushACT (**kwds):
[docs] return KeypadPush.ACT(**kwds)
def PushESC (**kwds):
[docs] return KeypadPush.ESC(**kwds)
def PushDOWN (**kwds):
[docs] return KeypadPush.DOWN(**kwds)
def PushUP (**kwds):
[docs] return KeypadPush.UP(**kwds)
def PushEASY (**kwds):
[docs] return KeypadPush.EASY(**kwds)
class ReadErrorStatus508 (PumpCommand):
[docs] """
"""
code = 38
descr = "error status"
params = [ ]
class ReadBolusHistory (PumpCommand):
[docs] """
"""
code = 39
descr = "bolus history"
params = [ ]
class ReadDailyTotals (PumpCommand):
[docs] """
"""
code = 40
descr = "..."
params = [ ]
class ReadPrimeBoluses (PumpCommand):
[docs] """
"""
code = 41
descr = "..."
params = [ ]
class ReadAlarms (PumpCommand):
[docs] """
"""
code = 42
descr = "..."
params = [ ]
class ReadProfileSets (PumpCommand):
[docs] """
"""
code = 43
descr = "..."
params = [ ]
class ReadUserEvents (PumpCommand):
[docs] """
"""
code = 44
descr = "..."
params = [ ]
class ReadRemoteControlID (PumpCommand):
[docs] """
"""
code = 46
descr = "..."
params = [ ]
class Read128KMem (PumpCommand):
[docs] """
"""
code = 55
descr = "..."
params = [ ]
class Read256KMem (PumpCommand):
[docs] """
"""
code = 56
descr = "..."
params = [ ]
class Bolus (PumpCommand):
[docs] """
Bolus some insulin.
XXX: Be careful please.
Best trying this not connected to the pump until you trust it.
"""
code = 66
descr = "Bolus"
params = [ ]
class ReadErrorStatus(PumpCommand):
[docs] """
>>> ReadErrorStatus(serial='665455').format() == ReadErrorStatus._test_ok
True
"""
_test_ok = bytearray([ 0x01, 0x00, 0xA7, 0x01, 0x66, 0x54, 0x55, 0x80,
0x00, 0x00, 0x02, 0x01, 0x00, 0x75, 0xD7, 0x00 ])
code = 117
descr = "Read Error Status any current alarms set?"
params = [ ]
retries = 2
maxRecords = 1
class ReadHistoryData(PumpCommand):
[docs] """
>>> ReadHistoryData(serial='208850', params=[ 0x03 ]).format() == ReadHistoryData._test_ok
True
>>> ReadHistoryData(params=[ 0x01 ]).params
[1]
>>> ReadHistoryData(params=[ 0x02 ]).params
[2]
>>> ReadHistoryData(params=[ 0x03 ]).params
[3]
>>> ReadHistoryData(page=0x01).params
[1]
>>> ReadHistoryData(page=0x02).params
[2]
>>> ReadHistoryData(page=0x03).params
[3]
"""
__fields__ = PumpCommand.__fields__ + ['page']
_test_ok = bytearray([ 0x01, 0x00, 0xA7, 0x01, 0x20, 0x88, 0x50, 0x80,
0x01, 0x00, 0x02, 0x02, 0x00, 0x80, 0x9B, 0x03,
0x36, ])
page = None
def __init__(self, page=None, **kwds):
if page is None and kwds.get('params', [ ]):
page = kwds.pop('params')[0] or 0
if page is not None:
self.page = int(page)
kwds['params'] = [ self.page ]
super(ReadHistoryData, self).__init__(**kwds)
def log_name(self, prefix=''):
[docs] return prefix + '{}-page-{}.data'.format(self.__class__.__name__, self.page)
def __str__(self):
base = ''.join([ self.__class__.__name__,
':size[%s]:' % self.size,
'[page][%s]' % self.page ])
return '{}:data[{}]:'.format(base, len(self.data))
def done(self):
[docs] eod = False
found = len(self.data or [ ])
expect = int(self.maxRecords * self.bytesPerRecord)
expect_crc = CRC8(self.data[:-1])
expect_size = "size check found[{}] expected[{}]".format(found, expect)
found_crc = 0
if self.responded and len(self.data) > 5:
found_crc = self.data[-1]
self.eod = eod = (self.data[5] & 0x80) > 0
explain_crc = "CRC ACK check found[{}] expected[{}]".format(found_crc, expect_crc)
is_eod = 'and has eod set? %s' % (eod)
log.info("%s:download:done %s:%s:%s" % (self, expect_size, explain_crc, is_eod))
return found >= expect
def respond(self, raw):
[docs] log.info('{} extending original {} with found {}'.format(str(self), len(self.data), len(raw)))
if len(raw) == self.size:
log.info('{} download respond replace original {} with found {}'.format(str(self), len(self.data), len(raw)))
self.data = raw
elif len(self.data) == self.size:
log.info('{} download respond original {}, XXX IGNORE found {}'.format(str(self), len(self.data), len(raw)))
pass
else:
log.info('{} download respond extend original {} with found {}'.format(str(self), len(self.data), len(raw)))
self.data.extend(raw)
self.responded = True
code = 128
descr = "Read History Data"
params = [ ]
retries = 2
maxRecords = 16
effectTime = .100
data = bytearray( )
def getData(self):
[docs] data = self.data
# log.info("XXX: READ HISTORY DATA!!:\n%s" % lib.hexdump(data))
return self.hexdump( )
return data
class ReadCurPageNumber(PumpCommand):
[docs] """
"""
code = 157
descr = "Read Cur Page Number"
params = [ ]
retries = 2
maxRecords = 1
pages = 'unknown'
def __str__(self):
return ':pages:'.join([self.__class__.__name__, str(self.pages) ])
def respond(self, data):
[docs] self.data = data
self.pages = self.getData( )
self.responded = True
def getData(self):
[docs] data = self.data
log.info("XXX: READ cur page number:\n%s" % lib.hexdump(data))
if len(data) == 1:
return int(data[0])
page = lib.BangLong(data[0:4])
# https://bitbucket.org/bewest/carelink/src/419fbf23495a/ddmsDTWApplet.src/minimed/ddms/deviceportreader/MMX15.java#cl-157
if page <= 0 or page > 36:
page = 36
return page
# MMX22/ CMD_READ_CURRENT_GLUCOSE_HISTORY_PAGE_NUMBER 205 0xcd ('\xcd') OK
class ReadCurGlucosePageNumber(PumpCommand):
[docs] """
"""
code = 205
descr = "Read Cur Glucose Page Number"
params = [ ]
retries = 2
maxRecords = 1
def getData(self):
[docs] data = self.data
log.info("XXX: READ cur page number:\n%s" % lib.hexdump(data))
if len(data) == 1:
return int(data[0])
result = dict(page= lib.BangLong(data[0:4]), glucose=data[5], isig=data[7])
return result
class ReadRTC(PumpCommand):
[docs] """
"""
code = 112
descr = "Read RTC"
params = [ ]
retries = 2
maxRecords = 1
def getData(self):
[docs] data = self.data
d = {
'hour' : int(data[0]),
'minute': int(data[1]),
'second': int(data[2]),
# XXX
'year' : 2000 + (data[4] & 0x0F),
'month' : int(data[5]),
'day' : int(data[6]),
}
return "%(year)s-%(month)s-%(day)sT%(hour)s:%(minute)s:%(second)s" % (d)
class ReadPumpID(PumpCommand):
[docs] """
"""
code = 113
descr = "Read Pump ID"
params = [ ]
retries = 2
maxRecords = 1
def getData(self):
[docs] data = self.data
return str(data[0:6])
class ReadBatteryStatus(PumpCommand):
[docs] """
"""
code = 114
descr = "Read Battery Status"
params = [ ]
retries = 2
maxRecords = 1
def getData(self):
[docs] data = self.data
bd = bytearray(data)
volt = lib.BangInt((bd[1], bd[2]))
indicator = bd[0]
battery = {'status': {0: 'normal', 1: 'low'}[indicator], 'voltage': volt/100.0 }
return battery
class ReadFirmwareVersion(PumpCommand):
[docs] """
"""
code = 116
descr = "Read Firmware Version"
params = [ ]
retries = 2
maxRecords = 1
def getData(self):
[docs] data = self.data
log.debug("READ FIRMWARE HEX:\n%s" % lib.hexdump(data))
return str(data.split( chr(0x0b) )[0]).strip( )
class ReadRemainingInsulin(PumpCommand):
[docs] """
"""
code = 115
descr = "Read Remaining Insulin"
params = [ ]
retries = 2
maxRecords = 1
def getData(self):
[docs] data = self.data
log.info("READ remaining insulin:\n%s" % lib.hexdump(data))
return lib.BangInt(data[0:2])/10.0
class ReadBasalTemp508 (PumpCommand):
[docs] """
"""
code = 64
descr = "Read Temp Basal 508 (old)"
params = [ ]
retries = 2
maxRecords = 1
def getData(self):
[docs] data = self.data
rate = lib.BangInt(data[2:4])/40.0
duration = lib.BangInt(data[4:6])
log.info("READ temporary basal:\n%s" % lib.hexdump(data))
return { 'rate': rate, 'duration': duration }
class ReadTodayTotals508 (PumpCommand):
[docs] """
"""
code = 65
descr = "Read Totals Today"
params = [ ]
retries = 2
maxRecords = 1
def getData(self):
[docs] data = self.data
log.info("READ totals today:\n%s" % lib.hexdump(data))
totals = {
'today': lib.BangInt(data[0:2]) / 10.0,
'yesterday': lib.BangInt(data[2:4]) / 10.0
}
return totals
# MMPump511/ ReadTotalsToday 121 0x79 ('y') OK
class ReadTotalsToday(PumpCommand):
[docs] """
"""
code = 121
descr = "Read Totals Today"
params = [ ]
retries = 2
maxRecords = 1
def getData(self):
[docs] data = self.data
log.info("READ totals today:\n%s" % lib.hexdump(data))
totals = {
'today': lib.BangInt(data[0:2]) / 10.0,
'yesterday': lib.BangInt(data[2:4]) / 10.0
}
return totals
# MMPump511/ ReadProfiles_STD 122 0x7a ('z') OK
class ReadProfiles511_STD (PumpCommand):
[docs] code = 122
# MMPump511/ ReadProfiles_A 123 0x7b ('{') ??
class ReadProfiles511_A (PumpCommand):
[docs] code = 123
# MMPump511/ ReadProfiles_B 124 0x7c ('|') ??
class ReadProfiles511_B (PumpCommand):
[docs] code = 124
# MMPump???/ CMD_????? 125 0x7d ('}') ??
class Model511_ExperimentOP125 (PumpCommand):
[docs] code = 125
# MMPump???/ CMD_????? 126 0x7e ('~') ??
class Model511_ExperimentOP126 (PumpCommand):
[docs] code = 126
# MMPump511/ ReadSettings 127 0x7f DEL
class ReadSettings511 (PumpCommand):
[docs] code = 127
# MMX11/ CMD_ENABLE_DISABLE_DETAIL_TRACE 160 0x9f ('\x9f') ??
class PumpTraceSelect (PumpCommand):
[docs] code = 160
class PumpEnableDetailTrace (PumpTraceSelect):
[docs] params = [ 1 ]
class PumpDisableDetailTrace (PumpTraceSelect):
[docs] params = [ 0 ]
class Experiment_OP161 (PumpCommand):
[docs] code = 161
class Experiment_OP162 (PumpCommand):
[docs] code = 162
# MMPump511/ ReadPumpTrace 163 0xa3 ('\xa3') ??
class ReadPumpTrace (PumpCommand):
[docs] code = 163
maxRecords = 16
# MMPump511/ ReadDetailTrace 164 0xa4 ('\xa4') ??
class ReadDetailTrace (PumpCommand):
[docs] code = 164
maxRecords = 16
# MMPump11??/ CMD_???????????? 165 0xa5 0xa5 ??
class Model511_Experiment_OP165 (PumpCommand):
[docs] code = 165
# MMPump511/ ReadNewTraceAlarm 166 0xa6 ('\xa6') ??
class ReadNewTraceAlarm (PumpCommand):
[docs] code = 166
maxRecords = 16
# MMPump511/ ReadOldTraceAlarm 167 0xa7 ('\xa7') ??
class ReadOldTraceAlarm (PumpCommand):
[docs] maxRecords = 16
code = 167
# MMPump???/ CMD_????? 36 0x24 ('$') ??
class PumpExperimentSelfCheck_OP36 (PumpCommand):
code = 36
# MMX22/ CMD_WRITE_GLUCOSE_HISTORY_TIMESTAMP 40 0x28 ('(') ??
class WriteGlucoseHistoryTimestamp (PumpCommand):
[docs] code = 40
class ReadRadioCtrlACL(PumpCommand):
[docs] """
"""
code = 118
descr = "Read Radio ACL"
params = [ ]
retries = 2
maxRecords = 1
def getData(self):
[docs] data = self.data
ids = [ ]
ids.append( str(data[0:6]) )
ids.append( str(data[6:12]) )
ids.append( str(data[12:18]) )
log.info("READ radio ACL:\n%s" % lib.hexdump(data))
return ids
class Model511_Experiment_OP119 (PumpCommand):
[docs] code = 119
class Model511_Experiment_OP120 (PumpCommand):
[docs] code = 120
class Model511_Experiment_OP121 (PumpCommand):
[docs] code = 121
class Model511_Experiment_OP122 (PumpCommand):
[docs] code = 122
class Model511_Experiment_OP123 (PumpCommand):
[docs] code = 123
class Model511_Experiment_OP124 (PumpCommand):
[docs] code = 124
class Model511_Experiment_OP125 (PumpCommand):
[docs] code = 125
class Model511_Experiment_OP126 (PumpCommand):
[docs] code = 126
class Model511_Experiment_OP127 (PumpCommand):
[docs] code = 127
class Model511_Experiment_OP128 (PumpCommand):
[docs] code = 128
class Model511_Experiment_OP129 (PumpCommand):
[docs] code = 129
class Model511_Experiment_OP130 (PumpCommand):
[docs] code = 130
# MMPump512/ CMD_READ_LANGUAGE 134 0x86 ('\x86') ??
class ReadLanguage (PumpCommand):
[docs] code = 134
# MMPump512/ CMD_READ_BOLUS_WIZARD_SETUP_STATUS 135 0x87 ('\x87') ??
class ReadBolusWizardSetupStatus (PumpCommand):
[docs] code = 135
# MMPump512/ CMD_READ_CARB_UNITS 136 0x88 ('\x88') OK
class ReadCarbUnits (PumpCommand):
[docs] code = 136
# MMPump512/ CMD_READ_BG_UNITS 137 0x89 ('\x89') ??
class ReadBGUnits (PumpCommand):
[docs] code = 137
# MMPump512/ CMD_READ_CARB_RATIOS 138 0x8a ('\x8a') OK
class ReadCarbRatios (PumpCommand):
[docs] code = 138
# MMPump512/ CMD_READ_INSULIN_SENSITIVITIES 139 0x8b ('\x8b') OK
class ReadInsulinSensitivities (PumpCommand):
[docs] code = 139
resp_1 = bytearray(b'\x01\x00-\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00')
def getData (self):
[docs] data = self.data
isFast = data[17] == 0
if isFast:
return 'Fast'
return 'Regular'
# MMPump512/ CMD_READ_BG_TARGETS 140 0x8c ('\x8c') ??
class ReadBGTargets (PumpCommand):
[docs] code = 140
# MMPump512/ CMD_READ_BG_ALARM_CLOCKS 142 0x8e ('\x8e') ??
class ReadBGAlarmCLocks (PumpCommand):
[docs] code = 142
# MMPump512/ CMD_READ_RESERVOIR_WARNING 143 0x8f ('\x8f') ??
class ReadReservoirWarning (PumpCommand):
[docs] code = 143
# MMPump512/ CMD_READ_BG_REMINDER_ENABLE 144 0x90 ('\x90') ??
class ReadBGReminderEnable (PumpCommand):
[docs] code = 144
# MMPump512/ CMD_READ_SETTINGS 145 0x91 ('\x91') ??
class ReadSettings512 (PumpCommand):
[docs] code = 145
# MMPump512/ CMD_READ_STD_PROFILES 146 0x92 ('\x92') ??
class ReadProfile_STD512 (PumpCommand):
[docs] """
>>> import json
>>> schedule = ReadProfile_STD512.decode(ReadProfile_STD512._test_result_1)
>>> len(schedule)
4
>>> print json.dumps(schedule[0])
{"start": "00:00:00", "rate": 0.8}
>>> print json.dumps(schedule[1])
{"start": "06:30:00", "rate": 0.9500000000000001}
>>> print json.dumps(schedule[2])
{"start": "09:30:00", "rate": 1.1}
>>> print json.dumps(schedule[3])
{"start": "14:00:00", "rate": 0.9500000000000001}
"""
_test_result_1 = bytearray([
32, 0, 0,
38, 0, 13,
44, 0, 19,
38, 0, 28,
0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0,
0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0,
0, 0, 0, 0, 0, 0, 0, 0
])
_test_schedule = {'total': 22.50, 'schedule': [
{ 'start': '12:00A', 'rate': 0.80 },
{ 'start': '6:30A', 'rate': 0.95 },
{ 'start': '9:30A', 'rate': 1.10 },
{ 'start': '2:00P', 'rate': 0.95 },
]}
code = 146
@staticmethod
def decode (data):
[docs] i = 0
schedule = [ ]
end = [ 0, 0, 0 ]
none = [ 0, 0, 0x3F ]
for i in xrange(len(data)/3):
off = i*3
r, z, m = data[off : off + 3]
if [r,z,m] in [end, none]:
break
schedule.append(dict(start=str(lib.basal_time(m)), rate=r*.025))
return schedule
def getData (self):
[docs] return self.decode(self.data)
# MMPump512/ CMD_READ_A_PROFILES 147 0x93 ('\x93') OK
class ReadProfile_A512 (ReadProfile_STD512):
[docs] code = 147
# MMPump512/ CMD_READ_B_PROFILES 148 0x94 ('\x94') OK
class ReadProfile_B512 (ReadProfile_STD512):
[docs] code = 148
# MMPump512/ CMD_READ_LOGIC_LINK_IDS 149 0x95 ('\x95') OK
class ReadLogicLinkIDS (PumpCommand):
[docs] code = 149
# MMPump512??/ CMD_???????????????? 150 0x96 ('\x96') ??
class Model512Experiment_OP150 (PumpCommand):
[docs] code = 150
# MMPump512/ CMD_READ_BG_ALARM_ENABLE 151 0x97 ('\x97') ??
class ReadBGAlarmEnable (PumpCommand):
[docs] code = 151
# MMPump512/ CMD_READ_TEMP_BASAL 152 0x98 ('\x98') OK
class ReadBasalTemp(PumpCommand):
[docs] """
MM511 - 120
MM512 and up - opcode 152
# strokes per basalunit = 40 - mm12, 10 in mm11
"""
code = 152
descr = "Read Temp Basal"
params = [ ]
retries = 2
maxRecords = 1
def getData(self):
[docs] data = self.data
rate = lib.BangInt(data[2:4])/40.0
duration = lib.BangInt(data[4:6])
log.info("READ temporary basal:\n%s" % lib.hexdump(data))
return { 'rate': rate, 'duration': duration }
# MMGuardian3/ CMD_READ_SENSOR_SETTINGS 207 0xcf ('\xcf') ??
class GuardianSensorSettings (PumpCommand):
code = 207
# MMGuardian3/ CMD_READ_SENSOR_PREDICTIVE_ALERTS 209 0xd1 ('\xd1') ??
class GuardianSensorSettings (PumpCommand):
[docs] code = 209
# MMGuardian3/ CMD_READ_SENSOR_DEMO_AND_GRAPH_TIMEOUT 210 0xd2 ('\xd2') ??
class GuardianSensorDemoGraphTimeout (PumpCommand):
[docs] code = 210
# MMGuardian3/ CMD_READ_SENSOR_ALARM_SILENCE 211 0xd3 ('\xd3') ??
class GuardianSensorAlarmSilence (PumpCommand):
[docs] code = 211
# MMGuardian3/ CMD_READ_SENSOR_RATE_OF_CHANGE_ALERTS 212 0xd4 ('\xd4') ??
class GuardianSensorRateChangeAlerts (PumpCommand):
[docs] code = 212
class ReadSettings(PumpCommand):
[docs] """
XXX: changed in MM512 to 192
"""
code = 192
descr = "Read Settings"
params = [ ]
retries = 2
maxRecords = 1
byte_map = {
}
def alarm(self, alarm):
[docs] d = { 'volume': alarm, 'mode': 2 }
if alarm == 4:
d = { 'volume': -1, 'mode': 1 }
return d
def temp_basal_type(self, data):
[docs] temp = { 'type': data[0] == 1 and "Percent" or "Units/hour",
'percent': data[1]
}
return temp
def getData(self):
[docs] data = self.data
log.info("READ pump settings:\n%s" % lib.hexdump(data))
if len(data) < 2:
log.info("pump settings: unsupported version, sorry")
return data
auto_off_duration_hrs = data[0]
alarm = self.alarm(data[1])
audio_bolus_enable = data[2] == 1
audio_bolus_size = 0
if audio_bolus_enable:
audio_bolus_size = data[3] / 10.0
variable_bolus_enable = data[4] == 1
#MM23 is different
maxBolus = data[5]/ 10.0
# MM512 and up
maxBasal = lib.BangInt(data[6:8]) / 40
timeformat = data[8]
insulinConcentration = {0: 100, 1: 50}[data[9]]
patterns_enabled = data[10] == 1
selected_pattern = data[11]
rf_enable = data[12] == 1
block_enable = data[13] == 1
temp_basal = self.temp_basal_type(data[14:16])
paradigm_enabled = data[16]
"""
# MM12
insulin_action_type = data[17] == 0 and 'Fast' or 'Regular'
"""
#MM15
insulin_action_type = data[17]
low_reservoir_warn_type = data[18]
low_reservoir_warn_point = data[19]
keypad_lock_status = data[20]
values = locals( )
# safety
values.pop('self')
values.pop('data')
return values
# MMX15/ CMD_READ_SAVED_SETTINGS_DATE 193 0xc1 ('\xc1') ??
class ReadSavedSettingsDate (PumpCommand):
[docs] code = 193
class ReadContrast(PumpCommand):
[docs] """
"""
code = 195
descr = "Read Contrast"
params = [ ]
retries = 2
maxRecords = 1
def getData(self):
[docs] data = self.data
log.info("READ contrast:\n%s" % lib.hexdump(data))
return data
# MMX15/ CMD_READ_BOLUS_REMINDER_ENABLE 197 0xc5 ('\xc5') ??
class ReadBolusReminderEnable (PumpCommand):
[docs] code = 197
# MMX15/ CMD_READ_BOLUS_REMINDERS 198 0xc6 ('\xc6') ??
class ReadBolusReminders (PumpCommand):
[docs] code = 198
# MMX15/ CMD_READ_FACTORY_PARAMETERS 199 0xc7 ('\xc7') ??
class ReadFactoryParameters (PumpCommand):
[docs] code = 199
class ReadPumpState(PumpCommand):
[docs] """
>>> ReadPumpState(serial='665455').format() == ReadPumpState._test_ok
True
"""
_test_ok = bytearray([ 0x01, 0x00, 0xA7, 0x01, 0x66, 0x54, 0x55, 0x80,
0x00, 0x00, 0x02, 0x01, 0x00, 0x83, 0x2E, 0x00 ])
code = 131
descr = "Read Pump State"
params = [ ]
retries = 2
maxRecords = 1
class ReadPumpStatus(PumpCommand):
[docs] """
"""
code = 206
descr = "Read Pump Status"
params = [ ]
retries = 2
maxRecords = 1
def getData(self):
[docs] data = self.data
normal = { 03: 'normal' }
status = { 'status': normal.get(data[0], 'error'),
'bolusing': data[1] == 1,
'suspended': data[2] == 1
}
return status
# MMX22/ CMD_READ_SENSOR_SETTINGS 153 0x99 ('\x99') ??
class ReadSensorSettings (PumpCommand):
[docs] """
"""
descr = "Read sensor settings"
code = 153
params = [ ]
retries = 2
class ReadSensorHistoryData (ReadHistoryData):
[docs] def __init__(self, page=None, **kwds):
params = kwds.pop('params', [ ])
if page is not None:
params = [ lib.LowByte(page >> 24), lib.LowByte(page >> 16),
lib.LowByte(page >> 8), lib.LowByte(page) ]
self.page = page
super(ReadSensorHistoryData, self).__init__(params=params, **kwds)
self.params = params
self.page = page
# MMX22/ CMD_READ_GLUCOSE_HISTORY 154 0x9a ('\x9a') ??
class ReadGlucoseHistory (ReadSensorHistoryData):
[docs] """
>>> ReadGlucoseHistory(page=1).params
[0, 0, 0, 1]
>>> list(ReadGlucoseHistory(page=1).format( ))
[1, 0, 167, 1, 32, 136, 80, 128, 4, 0, 2, 2, 0, 154, 34, 0, 0, 0, 1, 155]
>>> ReadGlucoseHistory(page=2).params
[0, 0, 0, 2]
>>> ReadGlucoseHistory(page=3)
<ReadGlucoseHistory:size[1024]:[page][3]:data[0]:>
>>> list(ReadGlucoseHistory(page=3).format( ))
[1, 0, 167, 1, 32, 136, 80, 128, 4, 0, 2, 2, 0, 154, 34, 0, 0, 0, 3, 54]
>>> ReadGlucoseHistory(page=3).params
[0, 0, 0, 3]
>>> ReadGlucoseHistory(params=[1]).params
[1]
>>> ReadGlucoseHistory(params=[2]).params
[2]
>>> ReadGlucoseHistory(params=[3]).params
[3]
"""
descr = "read glucose history"
code = 154
params = [ ]
# MMX22/ CMD_READ_ISIG_HISTORY 155 0x9b ('\x9b') ??
class ReadISIGHistory (ReadSensorHistoryData):
[docs] """
>>> ReadISIGHistory(page=0).params
[0, 0, 0, 0]
>>> ReadISIGHistory(page=1).params
[0, 0, 0, 1]
>>> ReadISIGHistory(page=2).params
[0, 0, 0, 2]
"""
descr = "read ISIG history"
code = 155
params = [ ]
maxRecords = 32
# MMX22/ CMD_READ_CALIBRATION_FACTOR 156 0x9c ('\x9c') ??
class ReadCalibrationFactor (PumpCommand):
[docs] """
"""
code = 156
# MMX23/ CMD_READ_VCNTR_HISTORY 213 0xd5 ('\xd5') ??
class ReadVCNTRHistory (ReadSensorHistoryData):
[docs] code = 213
# MMX23/ CMD_READ_OTHER_DEVICES_IDS 240 0xf0 ('\xf0') ??
class ReadOtherDevicesIDS (PumpCommand):
[docs] code = 240
class FilterHistory (PumpCommand):
[docs] """
"""
code = None
begin = None
end = None
__fields__ = PumpCommand.__fields__ + ['begin', 'end']
def __init__(self, begin=None, end=None, **kwds):
params = kwds.get('params', [ ])
if len(params) == 0:
params.extend(lib.format_filter_date(begin))
params.extend(lib.format_filter_date(end))
kwds['params'] = params
super(FilterHistory, self).__init__(**kwds)
def getData(self):
[docs] data = self.data
if len(data) < 4:
return bytearray(data)
begin = lib.BangInt(data[0:2])
end = lib.BangInt(data[2:4])
return dict(begin=begin, end=end)
@classmethod
def ISO (klass, begin=None, end=None, **kwds):
[docs] return klass(begin=lib.parse.date(begin), end=lib.parse.date(end), **kwds)
# MMX22??/ CMD_FILTER_BG 168 0xa8 ('\xa8') ??
class FilterGlucoseHistory (FilterHistory):
[docs] """
>>> FilterGlucoseHistory.ISO(begin='2014-04-13', end='2014-04-14').params
[7, 222, 4, 13, 7, 222, 4, 14]
"""
code = 168
# MMX22??/ CMD_FILTER_ISIG 169 0xa9 ('\xa9') ??
class FilterISIGHistory (FilterHistory):
[docs] """
>>> FilterISIGHistory.ISO(begin='2014-04-13', end='2014-04-14').params
[7, 222, 4, 13, 7, 222, 4, 14]
"""
code = 169
class TweakAnotherCommand (ManualCommand):
@classmethod
def get_kwds (klass, Other, args):
kwds = { }
fields = list(set(Other.__fields__) - set(['serial', ]))
for k in fields:
value = getattr(args, k, None)
if value is not None:
kwds[k] = value
return kwds
@classmethod
def config_argparse (klass, parser):
parser.add_argument('--params', type=int, action="append",
help="parameters to format into sent message"
)
parser.add_argument('--descr', type=str,
help="Description of command"
)
parser.add_argument('--name', type=str,
help="Proposed name of command"
)
parser.add_argument('--save', action="store_true", default=False,
help="Save response in a file."
)
parser.add_argument('--effectTime', type=float,
help="time to sleep before responding to message, float in seconds"
)
parser.add_argument('--maxRecords', type=int,
help="number of frames in a packet composing payload response"
)
parser.add_argument('--bytesPerRecord', type=int,
help="bytes per frame"
)
parser.add_argument('--page', type=int,
help="Page to fetch (for ReadHistoryData)"
)
parser.add_argument('--begin', type=lib.parse.date,
help="begin date for FilterHistory"
)
parser.add_argument('--end', type=lib.parse.date,
help="end date for FilterHistory"
)
return parser
class ReadPumpModel(PumpCommand):
[docs] """
>>> ReadPumpModel(serial='665455').format() == ReadPumpModel._test_ok
True
"""
code = 141
descr = "Read Pump Model Number"
params = [ ]
retries = 2
maxRecords = 1
_test_ok = bytearray([ 0x01, 0x00, 0xA7, 0x01, 0x66, 0x54, 0x55, 0x80,
0x00, 0x00, 0x02, 0x01, 0x00, 0x8D, 0x5B, 0x00 ])
def getData(self):
[docs] data = self.data
length = data[0]
msg = data[1:1+length]
self.model = msg
return str(msg)
def do_commands(device):
comm = ReadPumpModel( serial=device.serial )
device.execute(comm)
log.info('comm:%s:data:%s' % (comm, getattr(comm.getData( ), 'data', None)))
log.info('REMOTE PUMP MODEL NUMBER: %s' % comm.getData( ))
log.info("READ RTC")
comm = ReadRTC( serial=device.serial )
device.execute(comm)
log.info('comm:RTC:%s' % (comm.getData( )))
log.info("READ PUMP ID")
comm = ReadPumpID( serial=device.serial )
device.execute(comm)
log.info('comm:READ PUMP ID: ID: %s' % (comm.getData( )))
log.info("Battery Status")
comm = ReadBatteryStatus( serial=device.serial )
device.execute(comm)
log.info('comm:READ Battery Status: %r' % (comm.getData( )))
log.info("Firmware Version")
comm = ReadFirmwareVersion( serial=device.serial )
device.execute(comm)
log.info('comm:READ Firmware Version: %r' % (comm.getData( )))
log.info("remaining insulin")
comm = ReadRemainingInsulin( serial=device.serial )
device.execute(comm)
log.info('comm:READ Remaining Insulin: %r' % (comm.getData( )))
log.info("read totals today")
comm = ReadTotalsToday( serial=device.serial )
device.execute(comm)
log.info('comm:READ totals today: %r' % (comm.getData( )))
log.info("read remote IDS")
comm = ReadRadioCtrlACL( serial=device.serial )
device.execute(comm)
log.info('comm:READ radio ACL: %r' % (comm.getData( )))
log.info("read temporary basal")
comm = ReadBasalTemp( serial=device.serial )
device.execute(comm)
log.info('comm:READ temp basal: %r' % (comm.getData( )))
log.info("read settings")
comm = ReadSettings( serial=device.serial )
device.execute(comm)
log.info('comm:READ settings!: %r' % (comm.getData( )))
log.info("read contrast")
comm = ReadContrast( serial=device.serial )
device.execute(comm)
log.info('comm:READ contrast: %r' % (comm.getData( )))
log.info("read cur page number")
comm = ReadCurPageNumber(serial=device.serial )
device.execute(comm)
log.info('comm:READ page number!!!: %r' % (comm.getData( )))
log.info("read HISTORY DATA")
comm = ReadHistoryData(serial=device.serial, page=0)
device.execute(comm)
log.info('comm:READ history data page!!!:\n%s' % (lib.hexdump(comm.getData( ))))
def get_pages(device):
log.info("read cur page number")
comm = ReadCurPageNumber(serial=device.serial )
device.execute(comm)
pages = comm.getData( )
log.info('attempting to read %s pages of history' % pages)
for x in range(pages + 1):
log.info('comm:READ HISTORY DATA page number: %r' % (x))
comm = ReadHistoryData(serial=device.serial, params=[ x ] )
device.execute(comm)
page = comm.getData( )
log.info("XXX: READ HISTORY DATA!!:\n%s" % lib.hexdump(page))
time.sleep(.100)
__all__ = [
'BaseCommand', 'KeypadPush', 'PowerControl', 'PowerControlOff',
'PumpCommand', 'PumpResume', 'PumpSuspend',
'ReadBasalTemp', 'ReadBatteryStatus', 'ReadContrast',
'ReadCurPageNumber', 'ReadErrorStatus', 'ReadFirmwareVersion',
'ReadGlucoseHistory', 'ReadHistoryData', 'ReadPumpID',
'ReadPumpModel', 'ReadPumpState', 'ReadPumpStatus',
'ReadRTC', 'ReadRadioCtrlACL', 'ReadRemainingInsulin',
'ReadSettings', 'ReadTotalsToday', 'SetSuspend',
'PushEASY', 'PushUP', 'PushDOWN', 'PushACT', 'PushESC',
'TempBasal', 'ManualCommand', 'ReadCurGlucosePageNumber',
'Bolus',
'ReadErrorStatus508',
'ReadBolusHistory',
'ReadDailyTotals',
'ReadPrimeBoluses',
'ReadAlarms',
'ReadProfileSets',
'ReadUserEvents',
'ReadRemoteControlID',
'Read128KMem',
'Read256KMem',
'ReadBasalTemp508',
'ReadTodayTotals508',
'ReadSensorSettings',
'ReadSensorHistoryData',
'ReadISIGHistory',
'FilterHistory',
'FilterGlucoseHistory',
'FilterISIGHistory',
'ReadProfiles511_STD',
'ReadProfiles511_A',
'ReadProfiles511_B',
'Model511_ExperimentOP125',
'Model511_ExperimentOP126',
'ReadSettings511',
'ReadPumpTrace',
'ReadDetailTrace',
'Model511_Experiment_OP165',
'ReadNewTraceAlarm',
'ReadOldTraceAlarm',
'WriteGlucoseHistoryTimestamp',
'ReadLanguage',
'ReadBolusWizardSetupStatus',
'ReadCarbUnits',
'ReadBGUnits',
'ReadCarbRatios',
'ReadInsulinSensitivities',
'ReadBGTargets',
'ReadBGAlarmCLocks',
'ReadReservoirWarning',
'ReadBGReminderEnable',
'ReadSettings512',
'ReadProfile_STD512',
'ReadProfile_A512',
'ReadProfile_B512',
'ReadLogicLinkIDS',
'Model512Experiment_OP150',
'ReadBGAlarmEnable',
'GuardianSensorSettings',
'GuardianSensorSettings',
'GuardianSensorDemoGraphTimeout',
'GuardianSensorAlarmSilence',
'GuardianSensorRateChangeAlerts',
'ReadSavedSettingsDate',
'ReadBolusReminderEnable',
'ReadBolusReminders',
'ReadFactoryParameters',
'ReadCalibrationFactor',
'ReadVCNTRHistory',
'ReadOtherDevicesIDS',
'PumpTraceSelect',
'PumpEnableDetailTrace',
'PumpDisableDetailTrace',
'Experiment_OP161',
'Experiment_OP162',
'Model511_Experiment_OP119',
'Model511_Experiment_OP120',
'Model511_Experiment_OP121',
'Model511_Experiment_OP122',
'Model511_Experiment_OP123',
'Model511_Experiment_OP124',
'Model511_Experiment_OP125',
'Model511_Experiment_OP126',
'Model511_Experiment_OP127',
'Model511_Experiment_OP128',
'Model511_Experiment_OP129',
'Model511_Experiment_OP130',
'SelectBasalProfile',
'SelectBasalProfileSTD',
'SelectBasalProfileA',
'SelectBasalProfileB',
'PumpExperiment_OP69',
'PumpExperiment_OP70',
'PumpExperiment_OP71',
'PumpExperiment_OP72',
'PumpExperiment_OP73',
'PumpExperiment_OP75',
]
if __name__ == '__main__':
import doctest
doctest.testmod( )
import sys
port = None
port = sys.argv[1:] and sys.argv[1] or False
serial_num = sys.argv[2:] and sys.argv[2] or False
if not port or not serial_num:
print "usage:\n%s <port> <serial>, eg /dev/ttyUSB0 208850" % sys.argv[0]
sys.exit(1)
import link
import stick
import session
from pprint import pformat
logging.basicConfig(stream=sys.stdout, level=logging.DEBUG)
log.info("howdy! I'm going to take a look at your pump and grab lots of info.")
stick = stick.Stick(link.Link(port, timeout=.400))
stick.open( )
session = session.Pump(stick, serial_num)
log.info(pformat(stick.interface_stats( )))
log.info('PUMP MODEL: %s' % session.read_model( ))
do_commands(session)
log.info(pformat(stick.interface_stats( )))
#get_pages(session)
#log.info(pformat(stick.interface_stats( )))
log.info("howdy! we downloaded a lot of pump info successfully.")
# stick.open( )