diff options
Diffstat (limited to '')
| -rw-r--r-- | glucometerutils/drivers/freestyle_optium.py | 249 | ||||
| -rw-r--r-- | glucometerutils/exceptions.py | 7 |
2 files changed, 256 insertions, 0 deletions
diff --git a/glucometerutils/drivers/freestyle_optium.py b/glucometerutils/drivers/freestyle_optium.py new file mode 100644 index 0000000..0be4995 --- /dev/null +++ b/glucometerutils/drivers/freestyle_optium.py @@ -0,0 +1,249 @@ +# -*- coding: utf-8 -*- +"""Driver for FreeStyle Optium devices""" + +__author__ = 'Diego Elio Pettenò' +__email__ = 'flameeyes@flameeyes.eu' +__copyright__ = 'Copyright © 2016, Diego Elio Pettenò' +__license__ = 'MIT' + +import datetime +import re +import sys + +import serial + +from glucometerutils import common +from glucometerutils import exceptions + +_CLOCK_RE = re.compile( + r'^Clock:\t(?P<month>[A-Z][a-z]{2}) (?P<day>[0-9]{2}) (?P<year>[0-9]{4})\t' + r'(?P<time>[0-9]{2}:[0-9]{2}:[0-9]{2})$') + +# The reading can be HI (padded to three-characters by a space) if the value was +# over what the meter was supposed to read. Unlike the "Clock:" line, the months +# of June and July are written in full, everything else is truncated to three +# characters, so accept a space or 'e'/'y' at the end of the month name. Also, +# the time does *not* include seconds. Note that the G is likely going to +# indicate a blood-glucose reading rather than a blood-β-ketones. Unfortunately +# I don't have β-ketone strips to test. +_READING_RE = re.compile( + r'^(?P<reading>HI |[0-9]{3}) (?P<month>[A-Z][a-z]{2})[ ey] (?P<day>[0-9]{2}) ' + r'(?P<year>[0-9]{4}) (?P<time>[0-9]{2}:[0-9]{2}) G 0x00') + +# There are two date format used by the device. One uses three-letters month +# names, and that's easy enough. The other uses three-letters month names, +# except for (at least) July. So ignore the fourth character. +# explicit mapping. Note that the mapping *requires* a trailing whitespace. +_MONTH_MATCHES = { + 'Jan': 1, + 'Feb': 2, + 'Mar': 3, + 'Apr': 4, + 'May': 5, + 'Jun': 6, + 'Jul': 7, + 'Aug': 8, + 'Sep': 9, + 'Oct': 10, + 'Nov': 11, + 'Dec': 12 +} + + +def _parse_clock(datestr): + """Convert the date/time string used by the the device into a datetime. + + Args: + datestr: a string as returned by the device during information handling. + """ + match = _CLOCK_RE.match(datestr) + if not match: + raise exceptions.InvalidResponse(datestr) + + # int() parses numbers in decimal, so we don't have to worry about '08' + day = int(match.group('day')) + month = _MONTH_MATCHES[match.group('month')] + year = int(match.group('year')) + + hour, minute, second = (int(part) for part in match.group('time').split(':')) + + return datetime.datetime(year, month, day, hour, minute, second) + + +class Device(object): + def __init__(self, device): + self.serial_ = serial.Serial( + port=device, baudrate=19200, bytesize=serial.EIGHTBITS, + parity=serial.PARITY_NONE, stopbits=serial.STOPBITS_ONE, + timeout=1, xonxoff=True, rtscts=False, dsrdtr=False, writeTimeout=None) + + def _send_command(self, command): + cmd_bytes = bytes('$%s\r\n' % command, 'ascii') + self.serial_.write(cmd_bytes) + self.serial_.flush() + + response = self.serial_.readlines() + # We always want to decode the output, and remove stray \r\n. Any failure in + # decoding means the output is invalid anyway. + decoded_response = [line.decode('ascii').rstrip('\r\n') + for line in response] + return decoded_response + + def connect(self): + self._send_command('xmem') # ignore output this time + self._fetch_device_information() + + def disconnect(self): + return + + def _fetch_device_information(self): + data = self._send_command('colq') + + for line in data: + parsed_line = line.split('\t') + + if parsed_line[0] == 'S/N:': + self.device_serialno_ = parsed_line[1] + elif parsed_line[0] == 'Ver:': + self.device_version_ = parsed_line[1] + if parsed_line[2] == 'MMOL': + self.device_glucose_unit_ = common.UNIT_MMOLL + else: # I only have a mmol/l device, so I can't be sure. + self.device_glucose_unit_ = common.UNIT_MGDL + # There are more entries: Clock, Market, ROM and Usage, but we don't care + # for those here. + elif parsed_line[0] == 'CMD OK': + return + + # I have not figured out why this happens, but sometimes it's echoing back + # the commands and not replying to them. + raise exceptions.ConnectionFailed() + + def get_information_string(self): + """Returns a single string with all the identification information. + + Returns: + A string including the serial number, software version, date and time and + default unit. + """ + return ('Freestyle Optium glucometer\n' + 'Serial number: %s\n' + 'Software version: %s\n' + 'Time: %s\n' + 'Default unit: %s' % ( + self.get_serial_number(), + self.get_version(), + self.get_datetime(), + self.get_glucose_unit())) + + def get_version(self): + """Returns an identifier of the firmware version of the glucometer. + + Returns: + The software version returned by the glucometer, such as "0.22" + """ + return self.device_version_ + + def get_serial_number(self): + """Retrieve the serial number of the device. + + Returns: + A string representing the serial number of the device. + """ + return self.device_serialno_ + + def get_glucose_unit(self): + """Returns a constant representing the unit displayed by the meter. + + Returns: + common.UNIT_MGDL: if the glucometer displays in mg/dL + common.UNIT_MMOLL: if the glucometer displays in mmol/L + """ + return self.device_glucose_unit_ + + def get_datetime(self): + """Returns the current date and time for the glucometer. + + Returns: + A datetime object built according to the returned response. + """ + data = self._send_command('colq') + + for line in data: + if not line.startswith('Clock:'): + continue + + return _parse_clock(line) + + raise exceptions.InvalidResponse('\n'.join(data)) + + def set_datetime(self, date=datetime.datetime.now()): + """Sets the date and time of the glucometer. + + Args: + date: The value to set the date/time of the glucometer to. If none is + given, the current date and time of the computer is used. + + Returns: + A datetime object built according to the returned response. + """ + data = self._send_command(date.strftime('tim,%m,%d,%y,%H,%M')) + + parsed_data = ''.join(data) + if parsed_data != 'CMD OK': + raise exceptions.InvalidResponse(parsed_data) + + return self.get_datetime() + + def zero_log(self): + """Zeros out the data log of the device. + + This function will clear the memory of the device deleting all the readings + in an irrecoverable way. + """ + raise NotImplementedError + + def get_readings(self): + """Iterates over the reading values stored in the glucometer. + + Args: + unit: The glucose unit to use for the output. + + Yields: + A tuple (date, value) of the readings in the glucometer. The value is a + floating point in the unit specified; if no unit is specified, the default + unit in the glucometer will be used. + + Raises: + exceptions.InvalidResponse: if the response does not match what expected. + """ + data = self._send_command('xmem') + + # The first line is empty, the second is the serial number, the third the + # version, the fourth the current time, and the fifth the record count.. The + # last line has a checksum and the end. + count = int(data[4]) + if count != (len(data) - 6): + raise exceptions.InvalidResponse('\n'.join(data)) + + for line in data[5:-1]: + match = _READING_RE.match(line) + if not match: + raise exceptions.InvalidResponse(line) + + if match.group('reading') == 'HI ': + value = float("inf") + else: + value = float(match.group('reading')) + + day = int(match.group('day')) + month = _MONTH_MATCHES[match.group('month')] + year = int(match.group('year')) + + hour, minute = (int(part) for part in match.group('time').split(':')) + + timestamp = datetime.datetime(year, month, day, hour, minute) + + # The reading, if present, is always in mg/dL even if the glucometer is + # set to mmol/L. + yield common.Reading(timestamp, value) diff --git a/glucometerutils/exceptions.py b/glucometerutils/exceptions.py index 9802004..3e73048 100644 --- a/glucometerutils/exceptions.py +++ b/glucometerutils/exceptions.py @@ -13,6 +13,13 @@ class Error(Exception): return self.message +class ConnectionFailed(Error): + """It was not possible to connect to the meter.""" + + def __init__(self): + self.message = 'Unable to connect to the meter.' + + class InvalidResponse(Error): """The response received from the meter was not understood""" |