# -*- coding: utf-8 -*-
#
# SPDX-License-Identifier: MIT
"""Driver for LifeScan OneTouch Verio (2015) and Select Plus devices.
Verio 2015 devices can be recognized by microUSB connectors.
Supported features:
- get readings, including pre-/post-meal notes †;
- use the glucose unit preset on the device by default;
- get and set date and time;
- get serial number and software version;
- memory reset (caution!)
Expected device path: /dev/sdb or similar USB block device.
† Pre-/post-meal notes are only supported on Select Plus devices.
Further information on the device protocol can be found at
https://flameeyes.github.io/glucometer-protocols/lifescan/onetouch-verio-2015
"""
import binascii
import datetime
import logging
import construct
from pyscsi.pyscsi.scsi import SCSI
from pyscsi.pyscsi.scsi_device import SCSIDevice
from glucometerutils import common
from glucometerutils import exceptions
from glucometerutils.support import lifescan
from glucometerutils.support import lifescan_binary_protocol
# This device uses SCSI blocks as registers.
_REGISTER_SIZE = 512
_PACKET = construct.Padded(
_REGISTER_SIZE,
lifescan_binary_protocol.LifeScanPacket(False))
_COMMAND_SUCCESS = construct.Const(b'\x03\x06')
_QUERY_REQUEST = construct.Struct(
construct.Const(b'\x03\xe6\x02'),
'selector' / construct.Enum(
construct.Byte, serial=0x00, model=0x01, software=0x02),
)
_QUERY_RESPONSE = construct.Struct(
construct.Const(b'\x03\x06'),
'value' / construct.CString(encoding='utf-16-le'),
)
_READ_PARAMETER_REQUEST = construct.Struct(
construct.Const(b'\x03'),
'selector' / construct.Enum(
construct.Byte, unit=0x04),
)
_READ_UNIT_RESPONSE = construct.Struct(
_COMMAND_SUCCESS,
'unit' / lifescan_binary_protocol.GLUCOSE_UNIT,
construct.Padding(3),
)
_READ_RTC_REQUEST = construct.Const(b'\x03\x20\x02')
_READ_RTC_RESPONSE = construct.Struct(
_COMMAND_SUCCESS,
'timestamp' / lifescan_binary_protocol.VERIO_TIMESTAMP,
)
_WRITE_RTC_REQUEST = construct.Struct(
construct.Const(b'\x03\x20\x01'),
'timestamp' / lifescan_binary_protocol.VERIO_TIMESTAMP,
)
_MEMORY_ERASE_REQUEST = construct.Const(b'\x03\x1a')
_READ_RECORD_COUNT_REQUEST = construct.Const(b'\x03\x27\x00')
_READ_RECORD_COUNT_RESPONSE = construct.Struct(
_COMMAND_SUCCESS,
'count' / construct.Int16ul,
)
_READ_RECORD_REQUEST = construct.Struct(
construct.Const(b'\x03\x31\x02'),
'record_id' / construct.Int16ul,
construct.Const(b'\x00'),
)
_MEAL_FLAG = {
common.Meal.NONE: 0x00,
common.Meal.BEFORE: 0x01,
common.Meal.AFTER: 0x02,
}
_READ_RECORD_RESPONSE = construct.Struct(
_COMMAND_SUCCESS,
'inverse_counter' / construct.Int16ul,
construct.Padding(1),
'lifetime_counter' / construct.Int16ul,
'timestamp' / lifescan_binary_protocol.VERIO_TIMESTAMP,
'value' / construct.Int16ul,
'meal' / construct.Mapping(
construct.Byte, _MEAL_FLAG),
construct.Padding(4),
)
class Device:
def __init__(self, device):
if not device:
raise exceptions.CommandLineError(
'--device parameter is required, should point to the disk '
'device representing the meter.')
self.device_name_ = device
self.scsi_device_ = SCSIDevice(device, readwrite=True)
self.scsi_ = SCSI(self.scsi_device_)
self.scsi_.blocksize = _REGISTER_SIZE
def connect(self):
inq = self.scsi_.inquiry()
logging.debug('Device connected: %r', inq.result)
vendor = inq.result['t10_vendor_identification'][:32]
if vendor != b'LifeScan':
raise exceptions.ConnectionFailed(
'Device %s is not a LifeScan glucometer.' % self.device_name_)
def disconnect(self): # pylint: disable=no-self-use
return
def _send_request(self, lba, request_format, request_obj, response_format):
"""Send a request to the meter, and read its response.
Args:
lba: (int) the address of the block register to use, known
valid addresses are 3, 4 and 5.
request_format: a construct format identifier of the request to send
request_obj: the object to format with the provided identifier
response_format: a construct format identifier to parse the returned
message with.
Returns:
The Container object parsed from the response received by the meter.
Raises:
lifescan.MalformedCommand if Construct fails to build the request or
parse the response.
"""
try:
request = request_format.build(request_obj)
request_raw = _PACKET.build({'data': {'value': {
'message': request,
}}})
logging.debug(
'Request sent: %s', binascii.hexlify(request_raw))
self.scsi_.write10(lba, 1, request_raw)
response_raw = self.scsi_.read10(lba, 1)
logging.debug(
'Response received: %s', binascii.hexlify(response_raw.datain))
response_pkt = _PACKET.parse(response_raw.datain).data
logging.debug('Response packet: %r', response_pkt)
response = response_format.parse(response_pkt.value.message)
logging.debug('Response parsed: %r', response)
return response
except construct.ConstructError as e:
raise lifescan.MalformedCommand(str(e))
def _query_string(self, selector):
response = self._send_request(
3, _QUERY_REQUEST, {'selector': selector}, _QUERY_RESPONSE)
return response.value
def get_meter_info(self):
return common.MeterInfo(
'OneTouch %s glucometer' % self._query_string('model'),
serial_number=self.get_serial_number(),
version_info=(
'Software version: ' + self.get_version(),),
native_unit=self.get_glucose_unit())
def get_serial_number(self):
return self._query_string('serial')
def get_version(self):
return self._query_string('software')
def get_datetime(self):
response = self._send_request(
3, _READ_RTC_REQUEST, None, _READ_RTC_RESPONSE)
return response.timestamp
def set_datetime(self, date=datetime.datetime.now()):
self._send_request(
3, _WRITE_RTC_REQUEST, {'timestamp': date},
_COMMAND_SUCCESS)
# The device does not return the new datetime, so confirm by calling
# READ RTC again.
return self.get_datetime()
def zero_log(self):
self._send_request(
3, _MEMORY_ERASE_REQUEST, None,
_COMMAND_SUCCESS)
def get_glucose_unit(self):
response = self._send_request(
4, _READ_PARAMETER_REQUEST, {'selector': 'unit'},
_READ_UNIT_RESPONSE)
return response.unit
def _get_reading_count(self):
response = self._send_request(
3, _READ_RECORD_COUNT_REQUEST, None, _READ_RECORD_COUNT_RESPONSE)
return response.count
def _get_reading(self, record_id):
response = self._send_request(
3, _READ_RECORD_REQUEST, {'record_id': record_id},
_READ_RECORD_RESPONSE)
return common.GlucoseReading(
response.timestamp, float(response.value), meal=response.meal)
def get_readings(self):
record_count = self._get_reading_count()
for record_id in range(record_count):
yield self._get_reading(record_id)
