# -*- coding: utf-8 -*-
#
# SPDX-License-Identifier: MIT
"""Common routines to implement the FreeStyle common protocol.
Protocol documentation available at
https://protocols.glucometers.tech/abbott/shared-hid-protocol
"""
import csv
import datetime
import logging
import re
from abc import ABC
from typing import AnyStr, Callable, Iterator, List, Optional, Text, Tuple
import construct
from glucometerutils import exceptions
from glucometerutils.support import hiddevice, driver_base
_INIT_COMMAND = 0x01
_INIT_RESPONSE = 0x71
_KEEPALIVE_RESPONSE = 0x22
_UNKNOWN_MESSAGE_RESPONSE = 0x30
_ENCRYPTION_SETUP_COMMAND = 0x14
_ENCRYPTION_SETUP_RESPONSE = 0x33
_ALWAYS_UNENCRYPTED_MESSAGES = (
_INIT_COMMAND, 0x04, 0x05, 0x06, 0x0c, 0x0d,
_ENCRYPTION_SETUP_COMMAND, 0x15,
_ENCRYPTION_SETUP_RESPONSE, 0x34, 0x35,
_INIT_RESPONSE,
_KEEPALIVE_RESPONSE,
)
def _create_matcher(message_type, content):
# type: (int, Optional[bytes]) -> Callable[[Tuple[int, bytes]], bool]
def _matcher(message):
return (
message[0] == message_type and
(content is None or content == message[1]))
return _matcher
_is_init_reply = _create_matcher(_INIT_RESPONSE, b'\x01')
_is_keepalive_response = _create_matcher(_KEEPALIVE_RESPONSE, b'\x05')
_is_uknown_message_error = _create_matcher(_UNKNOWN_MESSAGE_RESPONSE, b'\x85')
_is_encryption_missing_error = _create_matcher(
_ENCRYPTION_SETUP_RESPONSE, b'\x15')
_is_encryption_setup_error = _create_matcher(
_ENCRYPTION_SETUP_RESPONSE, b'\x14')
_FREESTYLE_MESSAGE = construct.Struct(
'hid_report' / construct.Const(0, construct.Byte),
'message_type' / construct.Byte,
'command' / construct.Padded(
63, # command can only be up to 62 bytes, but one is used for length.
construct.Prefixed(construct.Byte, construct.GreedyBytes)),
)
_FREESTYLE_ENCRYPTED_MESSAGE = construct.Struct(
'hid_report' / construct.Const(0, construct.Byte),
'message_type' / construct.Byte,
'command' / construct.Padded(
63, # command can only be up to 62 bytes, but one is used for length.
construct.GreedyBytes),
)
_TEXT_COMPLETION_RE = re.compile(b'CMD (?:OK|Fail!)')
_TEXT_REPLY_FORMAT = re.compile(
b'^(?P<message>.*)CKSM:(?P<checksum>[0-9A-F]{8})\r\n'
b'CMD (?P<status>OK|Fail!)\r\n$', re.DOTALL)
_MULTIRECORDS_FORMAT = re.compile(
'^(?P<message>.+\r\n)(?P<count>[0-9]+),(?P<checksum>[0-9A-F]{8})\r\n$',
re.DOTALL)
def _verify_checksum(message, expected_checksum_hex):
# type: (AnyStr, AnyStr) -> None
"""Calculate the simple checksum of the message and compare with expected.
Args:
message: (str) message to calculate the checksum of.
expected_checksum_hex: hexadecimal string representing the checksum
expected to match the message.
Raises:
InvalidChecksum: if the message checksum calculated does not match the one
received.
"""
expected_checksum = int(expected_checksum_hex, 16)
if isinstance(message, bytes):
all_bytes = (c for c in message)
else:
all_bytes = (ord(c) for c in message)
calculated_checksum = sum(all_bytes)
if expected_checksum != calculated_checksum:
raise exceptions.InvalidChecksum(expected_checksum, calculated_checksum)
def convert_ketone_unit(raw_value):
"""Convert raw ketone value as read in the device to its value in mmol/L.
As per https://protocols.glucometers.tech/abbott/freestyle-libre this is
actually not using any mg/dL→mmol/L conversion, but rather the same as the
meter uses for blood glucose.
"""
return raw_value / 18.0
class FreeStyleHidDevice(hiddevice.HidDevice, driver_base.GlucometerDriver, ABC):
"""Base class implementing the FreeStyle HID common protocol.
This class implements opening, initializing the connection and sending
commands to the device, reading the response and confirming the checksums.
Commands sent to the devices over this protocol have a "message type"
prefixed to the command itself. Text command are usually sent with message
type 0x60, and the replied received with the same. Some devices may diverge
though.
"""
TEXT_CMD = 0x60
TEXT_REPLY_CMD = 0x60
USB_VENDOR_ID = 0x1a61 # type: int # Abbott Diabetes Care
USB_PRODUCT_ID = None # type: int
def connect(self):
"""Open connection to the device, starting the knocking sequence."""
self._send_command(_INIT_COMMAND, b'')
response = self._read_response()
if not _is_init_reply(response):
raise exceptions.ConnectionFailed(
'Connection error: unexpected message %02x:%s' % (
response[0], response[1].hex()))
def disconnect(self):
"""Disconnect the device, nothing to be done."""
pass
def _send_command(self, message_type, command, encrypted=False):
# type: (int, bytes, bool) -> None
"""Send a raw command to the device.
Args:
message_type: (int) The first byte sent with the report to the device.
command: (bytes) The command to send out the device.
"""
if encrypted:
assert message_type not in _ALWAYS_UNENCRYPTED_MESSAGES
meta_construct = _FREESTYLE_ENCRYPTED_MESSAGE
else:
meta_construct = _FREESTYLE_MESSAGE
usb_packet = meta_construct.build(
{'message_type': message_type, 'command': command})
logging.debug('Sending packet: %r', usb_packet)
self._write(usb_packet)
def _read_response(self, encrypted=False):
# type: (bool) -> Tuple[int, bytes]
"""Read the response from the device and extracts it."""
usb_packet = self._read()
logging.debug('Read packet: %r', usb_packet)
assert usb_packet
message_type = usb_packet[0]
if not encrypted or message_type in _ALWAYS_UNENCRYPTED_MESSAGES:
message_length = usb_packet[1]
message_content = usb_packet[2:2+message_length]
else:
message_content = usb_packet[1:]
# hidapi module returns a list of bytes rather than a bytes object.
message = (message_type, bytes(message_content))
# There appears to be a stray number of 22 01 xx messages being returned
# by some devices after commands are sent. These do not appear to have
# meaning, so ignore them and proceed to the next. These are always sent
# unencrypted, so we need to inspect them before we decide what the
# message content is.
if _is_keepalive_response(message):
return self._read_response(encrypted=encrypted)
if _is_uknown_message_error(message):
raise exceptions.CommandError('Invalid command')
if _is_encryption_missing_error(message):
raise exceptions.CommandError(
'Device encryption not initialized.')
if _is_encryption_setup_error(message):
raise exceptions.CommandError(
'Device encryption initialization failed.')
return message
def _send_text_command(self, command):
# type: (bytes) -> Text
"""Send a command to the device that expects a text reply."""
self._send_command(self.TEXT_CMD, command)
# Reply can stretch multiple buffers
full_content = b''
while True:
message_type, content = self._read_response()
logging.debug(
'Received message: type %02x content %s',
message_type, content.hex())
if message_type != self.TEXT_REPLY_CMD:
raise exceptions.InvalidResponse(
'Message type %02x does not match expectations: %r' %
(message_type, content))
full_content += content
if _TEXT_COMPLETION_RE.search(full_content):
break
match = _TEXT_REPLY_FORMAT.search(full_content)
if not match:
raise exceptions.InvalidResponse(full_content)
message = match.group('message')
_verify_checksum(message, match.group('checksum'))
if match.group('status') != b'OK':
raise exceptions.InvalidResponse(message or "Command failed")
# If there is anything in the response that is not ASCII-safe, this is
# probably in the patient name. The Windows utility does not seem to
# validate those, so just replace anything non-ASCII with the correct
# unknown codepoint.
return message.decode('ascii', 'replace')
# Some of the commands are also shared across devices that use this HID
# protocol, but not many. Only provide here those that do seep to change
# between them.
def _get_version(self):
# type: () -> Text
"""Return the software version of the device."""
return self._send_text_command(b'$swver?').rstrip('\r\n')
def get_serial_number(self):
# type: () -> Text
"""Returns the serial number of the device."""
return self._send_text_command(b'$serlnum?').rstrip('\r\n')
def get_patient_name(self):
# type: () -> Optional[Text]
patient_name = self._send_text_command(b'$ptname?').rstrip('\r\n')
if not patient_name:
return None
return patient_name
def set_patient_name(self, name):
# type: (Text) -> None
try:
encoded_name = name.encode('ascii')
except UnicodeDecodeError:
raise ValueError('Only ASCII-safe names are tested working')
result = self._send_text_command(b'$ptname,' + encoded_name)
def get_datetime(self):
# type: () -> datetime.datetime
"""Gets the date and time as reported by the device.
This is one of the few commands that appear common to many of the
FreeStyle devices that use the HID framing protocol.
"""
date = self._send_text_command(b'$date?').rstrip('\r\n')
time = self._send_text_command(b'$time?').rstrip('\r\n')
# Year is returned as an offset to 2000.
month, day, year = (int(x) for x in date.split(','))
hour, minute = (int(x) for x in time.split(','))
# At least Precision Neo devices can have an invalid date (bad RTC?),
# and report 255 for each field, which is not valid for
# datetime.datetime().
try:
return datetime.datetime(year + 2000, month, day, hour, minute)
except ValueError:
raise exceptions.InvalidDateTime()
def _set_device_datetime(self, date):
# type: (datetime.datetime) -> datetime.datetime
# The format used by the FreeStyle devices is not composable based on
# standard strftime() (namely it includes no leading zeros), so we need
# to build it manually.
date_cmd = '$date,{month},{day},{year}'.format(
month=date.month, day=date.day, year=(date.year-2000))
time_cmd = '$time,{hour},{minute}'.format(
hour=date.hour, minute=date.minute)
self._send_text_command(bytes(date_cmd, "ascii"))
self._send_text_command(bytes(time_cmd, "ascii"))
return self.get_datetime()
def _get_multirecord(self, command):
# type: (bytes) -> Iterator[List[Text]]
"""Queries for, and returns, "multirecords" results.
Multirecords are used for querying events, readings, history and similar
other data out of a FreeStyle device. These are comma-separated values,
variable-length.
The validation includes the general HID framing parsing, as well as
validation of the record count, and of the embedded records checksum.
Args:
command: (bytes) the text command to send to the device for the query.
Returns:
(csv.reader): a CSV reader object that returns a record for each line
in the record file.
"""
message = self._send_text_command(command)
logging.debug('Received multirecord message:\n%s', message)
if message == "Log Empty\r\n":
return iter(())
match = _MULTIRECORDS_FORMAT.search(message)
if not match:
raise exceptions.InvalidResponse(message)
records_str = match.group('message')
_verify_checksum(records_str, match.group('checksum'))
logging.debug('Received multi-record string: %s', records_str)
return csv.reader(records_str.split('\r\n'))
