From 8e6461c86569c1a90672b1d8dd921fc05b5cbd66 Mon Sep 17 00:00:00 2001 From: Lorenz Gillner Date: Sun, 1 Feb 2026 22:32:30 +0100 Subject: [PATCH 1/8] Implement basic communication for Contour Care devices --- glucometerutils/drivers/contourcare.py | 87 +++++++ glucometerutils/support/contourcare.py | 319 +++++++++++++++++++++++++ 2 files changed, 406 insertions(+) create mode 100644 glucometerutils/drivers/contourcare.py create mode 100644 glucometerutils/support/contourcare.py diff --git a/glucometerutils/drivers/contourcare.py b/glucometerutils/drivers/contourcare.py new file mode 100644 index 0000000..0658cd9 --- /dev/null +++ b/glucometerutils/drivers/contourcare.py @@ -0,0 +1,87 @@ +# -*- coding: utf-8 -*- +# +# SPDX-FileCopyrightText: © 2019 The glucometerutils Authors +# SPDX-License-Identifier: MIT +"""Driver for Contour Care devices. + +Supported features: + - get readings (blood glucose), including comments; + - get date and time; + - get serial number and software version; + - get device info (e.g. unit) + +Expected device path: /dev/hidraw4 or similar HID device. Optional when using +HIDAPI. +""" + +import datetime +from collections.abc import Generator +from typing import NoReturn, Optional + +from glucometerutils import common +from glucometerutils.support import contourcare + + +def _extract_timestamp(parsed_record: dict[str, str]): + """Extract the timestamp from a parsed record. + + This leverages the fact that all the reading records have the same base structure. + """ + datetime_str = parsed_record["datetime"] + + return datetime.datetime( + int(datetime_str[0:4]), # year + int(datetime_str[4:6]), # month + int(datetime_str[6:8]), # day + int(datetime_str[8:10]), # hour + int(datetime_str[10:12]), # minute + int(datetime_str[12:14]), # second + 0, + ) + + +class Device(contourcare.ContourCareHidDevice): + """Glucometer driver for Contour Care devices.""" + + USB_VENDOR_ID: int = 0x1A79 # Bayer Health Care LLC Contour + USB_PRODUCT_ID: int = 0x7950 + + def __init__(self, device: Optional[str]) -> None: + super().__init__((self.USB_VENDOR_ID, self.USB_PRODUCT_ID), device) + + def get_meter_info(self) -> common.MeterInfo: + self._get_info_record() + return common.MeterInfo( + "Contour Care", + serial_number=self.get_serial_number(), + version_info=("Meter versions: " + self._get_version(),), + native_unit=self.get_glucose_unit(), + ) + + def get_glucose_unit(self) -> common.Unit: + if self._get_glucose_unit() == "0": + return common.Unit.MG_DL + else: + return common.Unit.MMOL_L + + def get_readings(self) -> Generator[common.AnyReading, None, None]: + """ + Get reading dump from download data mode(all readings stored) + This meter supports only blood samples + """ + for parsed_record in self._get_multirecord(): + yield common.GlucoseReading( + _extract_timestamp(parsed_record), + int(parsed_record["value"]), + comment=parsed_record["markers"], + measure_method=common.MeasurementMethod.BLOOD_SAMPLE, + ) + + def get_serial_number(self) -> str: + return self._get_serial_number + + def _set_device_datetime(self, date: datetime.datetime) -> NoReturn: + raise NotImplementedError + + def zero_log(self) -> NoReturn: + raise NotImplementedError diff --git a/glucometerutils/support/contourcare.py b/glucometerutils/support/contourcare.py new file mode 100644 index 0000000..13f268c --- /dev/null +++ b/glucometerutils/support/contourcare.py @@ -0,0 +1,319 @@ +# -*- coding: utf-8 -*- +# +# SPDX-FileCopyrightText: © 2026 The glucometerutils Authors +# SPDX-License-Identifier: MIT +"""Common routines to implement communication with Contour Care devices. + +The communication is fairly similar to ContourUSB devices, but slightly different in a few places. +Since there is no official protocol document available + +* glucodump code segments are developed by Anders Hammarquist +* code for the contourusb driver was developed by Arvanitis Christos +* additional edits by Lorenz Gillner +""" + +import datetime +import enum +import re +from collections.abc import Generator +from typing import Optional + +from glucometerutils import driver +from glucometerutils.support import hiddevice + +_RECORD_FORMAT_RE = re.compile( + r"\x02(?P(?P[0-7])(?P[^\x0d]*)\x0d(?P[\x03\x17]))" # haven't seen 0x03 yet + r"(?P[0-9A-F][0-9A-F])\x0d\x0a" +) + +_HEADER_RECORD_RE = re.compile( + r"^(?P[A-Z])\|" + r"(?P.)(?P.)(?P.)\|\|" + r"(?P\w{6})\|" # what is this supposed to be? + r"(?P\w+)" + r"\^(?P\d{2}\.\d{2})" + r"\\(?P\d{2}\.\d{2})" + r"\\(?P\d{2}\.\d{2})" + r"\^(?P\w+)\|" + r"A=(?P\d)\^" + r"C=(?P\d+)\^" + r"R=(?P\d+)\^" + r"S=(?P\d+)\^" + r"U=(?P\d+)\^" + r"V=(?P\d{2})(?P\d{3})\^" + r"X=(?P\d{3})(?P\d{3})" + r"(?P\d{3})(?P\d{3})\^" + r"a=(?P\d+)\^" + r"J=(?P\d+)\|" + r"(?P\d*)\|\|\|\|\|" + r"P\|\d+\|" + r"(?P\d+)\|$" +) + +_RESULT_RECORD_RE = re.compile( + r"^(?P[a-zA-Z])\|(?P\d+)\|\w*\^\w*\^\w*\^" + r"(?P\w+)\|(?P\d+)\|(?P\w+\/\w+)\^" + r"(?P[BPD])\|\|(?P[>\d+)" +) + + +class FrameError(Exception): + pass + + +@enum.unique +class Mode(enum.Enum): + """Operation modes.""" + + ESTABLISH = enum.auto() + DATA = enum.auto() + PRECOMMAND = enum.auto() + COMMAND = enum.auto() + + +class ContourCareHidDevice(driver.GlucometerDevice): + """Base class implementing the Contour Care device.""" + + blocksize = 64 + + state: Optional[Mode] = None + + currecno: Optional[int] = None + + def __init__(self, usb_ids: tuple[int, int], device_path: Optional[str]) -> None: + super().__init__(device_path) + self._hid_session = hiddevice.HidSession(usb_ids, device_path) + + def read(self, r_size=blocksize): + result = [] + + while True: + data = self._hid_session.read() + dstr = data + data_end_idx = data[3] + 4 + result.append(dstr[4:data_end_idx]) + if data[3] != self.blocksize - 4: + break + + return b"".join(result) + + def write(self, data): + data = b"\x00\x00\x00" + chr(len(data)).encode() + data.encode() + pad_length = self.blocksize - len(data) + data += pad_length * b"\x00" + + self._hid_session.write(data) + + def parse_header_record(self, text): + header = _HEADER_RECORD_RE.search(text) + + self.field_del = header.group("field_del") + self.escape_del = header.group("escape_del") + self.component_del = header.group("component_del") + + self.product_code = header.group("product_code") + self.dig_ver = header.group("dig_ver") + self.anlg_ver = header.group("anlg_ver") + self.agp_ver = header.group("agp_ver") + + self.serial_num = header.group("serial_num") + self.res_marking = header.group("res_marking") + self.config_bits = header.group("config_bits") + self.ref_method = header.group("ref_method") + self.internal = header.group("internal") + + # U limit + self.unit = header.group("unit") + self.lo_bound = header.group("lo_bound") + self.hi_bound = header.group("hi_bound") + + # X field + self.post_food_low = header.group("post_food_low") + self.pre_food_low = header.group("pre_food_low") + self.post_food_high = header.group("post_food_high") + self.pre_food_high = header.group("pre_food_high") + + self.total = header.group("total_recs") + + # Datetime string in YYYYMMDDHHMMSS format + self.datetime = header.group("datetime") + + def checksum(self, text): + """ + Implemented by Anders Hammarquist for glucodump project + More info: https://bitbucket.org/iko/glucodump/src/default/ + """ + checksum = hex(sum(ord(c) for c in text) % 256).upper().split("X")[1] + return ("00" + checksum)[-2:] + + def checkframe(self, frame) -> Optional[str]: + """ + Implemented by Anders Hammarquist for glucodump project + More info: https://bitbucket.org/iko/glucodump/src/default/ + """ + match = _RECORD_FORMAT_RE.match(frame) + if not match: + raise FrameError("Couldn't parse frame", frame) + + recno = int(match.group("recno")) + if self.currecno is None: + self.currecno = recno + + if recno + 1 == self.currecno: + return None + + if recno != self.currecno: + raise FrameError( + f"Bad recno, got {recno!r} expected {self.currecno!r}", frame + ) + + calculated_checksum = self.checksum(match.group("check")) + received_checksum = match.group("checksum") + if calculated_checksum != received_checksum: + raise FrameError( + f"Checksum error: received {received_checksum} expected {calculated_checksum}", + frame, + ) + + self.currecno = (self.currecno + 1) % 8 + return match.group("text") + + def connect(self): + """Connecting the device, nothing to be done. + All process is hadled by hiddevice + """ + pass + + def _get_info_record(self): + self.currecno = None + self.state = Mode.ESTABLISH + try: + while True: + self.write("\x06") # this one is different + res = self.read() + if res[0] == 0x04 and res[-1] == 0x05: + # we are connected and just got a header + header_record = res.decode() + stx = header_record.find("\x02") + if stx != -1: + result = _RECORD_FORMAT_RE.match(header_record[stx:-1]).group( + "text" + ) + self.parse_header_record(result) + break + else: + pass + + except FrameError as e: + print("Frame error") + raise e + + except Exception as e: + print("Uknown error occured") + raise e + + def disconnect(self): + """Disconnect the device, nothing to be done.""" + pass + + # 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) -> str: + """Return the software version of the device.""" + return self.dig_ver + " - " + self.anlg_ver + " - " + self.agp_ver + + def _get_serial_number(self) -> str: + """Returns the serial number of the device.""" + return self.serial_num + + def _get_glucose_unit(self) -> str: + """Return 0 for mg/dL, 1 for mmol/L""" + return self.unit + + def get_datetime(self) -> datetime.datetime: + datetime_str = self.datetime + return datetime.datetime( + int(datetime_str[0:4]), # year + int(datetime_str[4:6]), # month + int(datetime_str[6:8]), # day + int(datetime_str[8:10]), # hour + int(datetime_str[10:12]), # minute + 0, + ) + + def sync(self) -> Generator[str, None, None]: + """ + Sync with meter and yield received data frames + FSM implemented by Anders Hammarquist's for glucodump + More info: https://bitbucket.org/iko/glucodump/src/default/ + """ + self.state = Mode.ESTABLISH + try: + tometer = "\x04" + result = None + foo = 0 + while True: + self.write(tometer) + if result is not None and self.state == Mode.DATA: + yield result + result = None + data_bytes = self.read() + data = data_bytes.decode() + + if self.state == Mode.ESTABLISH: + if data_bytes[-1] == 15: + # got a , send + tometer = chr(foo) + foo += 1 + foo %= 256 + continue + if data_bytes[-1] == 5: + # got an , send + tometer = "\x06" + self.currecno = None + continue + if self.state == Mode.DATA: + if data_bytes[-1] == 4: + # got an , done + self.state = Mode.PRECOMMAND + break + stx = data.find("\x02") + if stx != -1: + # got , parse frame + try: + result = self.checkframe(data[stx:]) + tometer = "\x06" + self.state = Mode.DATA + except FrameError: + tometer = "\x15" # Couldn't parse, + else: + # Got something we don't understand, it + tometer = "\x15" + except Exception as e: + raise e + + def parse_result_record(self, text: str) -> dict[str, str]: + result = _RESULT_RECORD_RE.search(text) + assert result is not None + rec_text = result.groupdict() + return rec_text + + def _get_multirecord(self) -> list[dict[str, str]]: + """Queries for, and returns, "multirecords" results. + + Returns: + (csv.reader): a CSV reader object that returns a record for each line + in the record file. + """ + records_arr = [] + for rec in self.sync(): + if rec[0] == "R": + # parse using result record regular expression + rec_text = self.parse_result_record(rec) + # get dictionary to use in main driver module without import re + + records_arr.append(rec_text) + # return csv.reader(records_arr) + return records_arr # array of groupdicts From fda09dc462118bfd5516de8485746f18cbd6ac00 Mon Sep 17 00:00:00 2001 From: Lorenz Gillner Date: Mon, 8 Jun 2026 22:02:54 +0200 Subject: [PATCH 2/8] Refactor communication --- .gitignore | 1 + glucometerutils/drivers/contourcare.py | 34 +++--- glucometerutils/support/contourcare.py | 153 +++++++++++++++---------- 3 files changed, 110 insertions(+), 78 deletions(-) diff --git a/.gitignore b/.gitignore index 0fe925e..e3e9439 100644 --- a/.gitignore +++ b/.gitignore @@ -14,3 +14,4 @@ __pycache__/ build .idea/ *venv/ +*.swp diff --git a/glucometerutils/drivers/contourcare.py b/glucometerutils/drivers/contourcare.py index 0658cd9..47ff7f1 100644 --- a/glucometerutils/drivers/contourcare.py +++ b/glucometerutils/drivers/contourcare.py @@ -30,10 +30,10 @@ def _extract_timestamp(parsed_record: dict[str, str]): datetime_str = parsed_record["datetime"] return datetime.datetime( - int(datetime_str[0:4]), # year - int(datetime_str[4:6]), # month - int(datetime_str[6:8]), # day - int(datetime_str[8:10]), # hour + int(datetime_str[0:4]), # year + int(datetime_str[4:6]), # month + int(datetime_str[6:8]), # day + int(datetime_str[8:10]), # hour int(datetime_str[10:12]), # minute int(datetime_str[12:14]), # second 0, @@ -42,28 +42,19 @@ def _extract_timestamp(parsed_record: dict[str, str]): class Device(contourcare.ContourCareHidDevice): """Glucometer driver for Contour Care devices.""" - - USB_VENDOR_ID: int = 0x1A79 # Bayer Health Care LLC Contour - USB_PRODUCT_ID: int = 0x7950 - + def __init__(self, device: Optional[str]) -> None: - super().__init__((self.USB_VENDOR_ID, self.USB_PRODUCT_ID), device) + super().__init__(device) def get_meter_info(self) -> common.MeterInfo: self._get_info_record() return common.MeterInfo( "Contour Care", serial_number=self.get_serial_number(), - version_info=("Meter versions: " + self._get_version(),), + version_info=("Meter versions: " + self.get_version(),), native_unit=self.get_glucose_unit(), ) - def get_glucose_unit(self) -> common.Unit: - if self._get_glucose_unit() == "0": - return common.Unit.MG_DL - else: - return common.Unit.MMOL_L - def get_readings(self) -> Generator[common.AnyReading, None, None]: """ Get reading dump from download data mode(all readings stored) @@ -78,7 +69,16 @@ def get_readings(self) -> Generator[common.AnyReading, None, None]: ) def get_serial_number(self) -> str: - return self._get_serial_number + return self._get_serial_number() + + def get_version(self): + return self._get_version() + + def get_glucose_unit(self) -> common.Unit: + if self._get_glucose_unit() == "0": + return common.Unit.MG_DL + else: + return common.Unit.MMOL_L def _set_device_datetime(self, date: datetime.datetime) -> NoReturn: raise NotImplementedError diff --git a/glucometerutils/support/contourcare.py b/glucometerutils/support/contourcare.py index 13f268c..53f30f1 100644 --- a/glucometerutils/support/contourcare.py +++ b/glucometerutils/support/contourcare.py @@ -21,20 +21,22 @@ from glucometerutils import driver from glucometerutils.support import hiddevice +# TODO this +# CR = 0x0d 0x0a (\r\n) +# Pipe = 0x7c (124) _RECORD_FORMAT_RE = re.compile( r"\x02(?P(?P[0-7])(?P[^\x0d]*)\x0d(?P[\x03\x17]))" # haven't seen 0x03 yet r"(?P[0-9A-F][0-9A-F])\x0d\x0a" ) _HEADER_RECORD_RE = re.compile( - r"^(?P[A-Z])\|" - r"(?P.)(?P.)(?P.)\|\|" - r"(?P\w{6})\|" # what is this supposed to be? - r"(?P\w+)" - r"\^(?P\d{2}\.\d{2})" - r"\\(?P\d{2}\.\d{2})" - r"\\(?P\d{2}\.\d{2})" - r"\^(?P\w+)\|" + r"^H\|\\\^\&\|\|" # repeat, component, escape, field + r"(?P\w{6})\|" + r"(?P\w+)\^" + r"(?P\d{2}\.\d{2})\\" + r"(?P\d{2}\.\d{2})\\" + r"(?P\d{2}\.\d{2})\^" + r"(?P\w+)\|" r"A=(?P\d)\^" r"C=(?P\d+)\^" r"R=(?P\d+)\^" @@ -46,15 +48,18 @@ r"a=(?P\d+)\^" r"J=(?P\d+)\|" r"(?P\d*)\|\|\|\|\|" - r"P\|\d+\|" + r"[DPT]\|(?P\d+)\|" r"(?P\d+)\|$" ) +_PATIENT_RECORD_RE = re.compile(r"P\|\d+") + _RESULT_RECORD_RE = re.compile( - r"^(?P[a-zA-Z])\|(?P\d+)\|\w*\^\w*\^\w*\^" - r"(?P\w+)\|(?P\d+)\|(?P\w+\/\w+)\^" - r"(?P[BPD])\|\|(?P[>\d+)" + r"^R\|(?P\d+)\|" + r"\^\^\^Glucose\|" + r"(?P\d+\.\d+)\|(?P\w+)\^P\|\|" + r"(?P.+)\|\|" # ??? + r"(?P\d+)$" ) @@ -62,6 +67,19 @@ class FrameError(Exception): pass +@enum.unique +class Term(enum.IntEnum): + """ASTM E1394 vocabulary.""" + + PAD = 0x00 + WAK = 0x58 + ENQ = 0x05 + ACK = 0x06 + STX = 0x02 + EOT = 0x04 + NAK = 0x15 + + @enum.unique class Mode(enum.Enum): """Operation modes.""" @@ -75,43 +93,43 @@ class Mode(enum.Enum): class ContourCareHidDevice(driver.GlucometerDevice): """Base class implementing the Contour Care device.""" - blocksize = 64 - + blocksize: int = 64 state: Optional[Mode] = None - currecno: Optional[int] = None - def __init__(self, usb_ids: tuple[int, int], device_path: Optional[str]) -> None: + USB_VENDOR_ID: int = 0x1A79 # Bayer Health Care LLC + USB_PRODUCT_ID: int = 0x7950 # Contour Care + + def __init__(self, device_path: Optional[str]) -> None: super().__init__(device_path) - self._hid_session = hiddevice.HidSession(usb_ids, device_path) + _id = (self.USB_VENDOR_ID, self.USB_PRODUCT_ID) + self._hid_session = hiddevice.HidSession(_id, device_path) - def read(self, r_size=blocksize): - result = [] + def read(self, r_size=blocksize) -> bytes: + result = bytes() while True: data = self._hid_session.read() dstr = data data_end_idx = data[3] + 4 - result.append(dstr[4:data_end_idx]) + result += dstr[4:data_end_idx] if data[3] != self.blocksize - 4: break - return b"".join(result) + return result - def write(self, data): - data = b"\x00\x00\x00" + chr(len(data)).encode() + data.encode() + def write(self, message: Term) -> None: + pad = bytes([Term.PAD]) + data = 4 * pad + data += chr(1).encode() + data += bytes([message]) pad_length = self.blocksize - len(data) - data += pad_length * b"\x00" - + data += pad_length * pad self._hid_session.write(data) - def parse_header_record(self, text): + def parse_header_record(self, text: str) -> None: header = _HEADER_RECORD_RE.search(text) - self.field_del = header.group("field_del") - self.escape_del = header.group("escape_del") - self.component_del = header.group("component_del") - self.product_code = header.group("product_code") self.dig_ver = header.group("dig_ver") self.anlg_ver = header.group("anlg_ver") @@ -180,9 +198,11 @@ def checkframe(self, frame) -> Optional[str]: return match.group("text") def connect(self): - """Connecting the device, nothing to be done. - All process is hadled by hiddevice - """ + """Connect to the device; handled by `hiddevice`.""" + pass + + def disconnect(self): + """Disconnect from the device; handled by `hiddevice`.""" pass def _get_info_record(self): @@ -190,16 +210,19 @@ def _get_info_record(self): self.state = Mode.ESTABLISH try: while True: - self.write("\x06") # this one is different + # Contour Care answers to pretty much anything with a header ... + self.write(Term.ENQ) res = self.read() - if res[0] == 0x04 and res[-1] == 0x05: - # we are connected and just got a header - header_record = res.decode() - stx = header_record.find("\x02") + + if res[0] == Term.EOT and res[-1] == Term.ENQ: + self.write(Term.ACK) + _ = self.read() + + # We are connected and just got a header + stx = res.find(Term.STX) if stx != -1: - result = _RECORD_FORMAT_RE.match(header_record[stx:-1]).group( - "text" - ) + header_record = res[stx:-1].decode() + result = _RECORD_FORMAT_RE.match(header_record).group("text") self.parse_header_record(result) break else: @@ -210,13 +233,9 @@ def _get_info_record(self): raise e except Exception as e: - print("Uknown error occured") + print("Unknown error occured") raise e - def disconnect(self): - """Disconnect the device, nothing to be done.""" - pass - # 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. @@ -245,19 +264,25 @@ def get_datetime(self) -> datetime.datetime: def sync(self) -> Generator[str, None, None]: """ - Sync with meter and yield received data frames - FSM implemented by Anders Hammarquist's for glucodump - More info: https://bitbucket.org/iko/glucodump/src/default/ + Sync with meter and yield received data frames. """ self.state = Mode.ESTABLISH try: - tometer = "\x04" + # Send "wake up call" + self.write(Term.WAK) + + tometer = Term.ACK result = None foo = 0 + + # Repeat until all records have been sent while True: self.write(tometer) + + # If we are in transmission mode, yield data if result is not None and self.state == Mode.DATA: yield result + result = None data_bytes = self.read() data = data_bytes.decode() @@ -265,32 +290,38 @@ def sync(self) -> Generator[str, None, None]: if self.state == Mode.ESTABLISH: if data_bytes[-1] == 15: # got a , send - tometer = chr(foo) + tometer = Term.EOT foo += 1 foo %= 256 continue - if data_bytes[-1] == 5: + + if data_bytes[-1] == Term.ENQ: # got an , send - tometer = "\x06" + tometer = Term.ACK self.currecno = None - continue + # continue + if self.state == Mode.DATA: - if data_bytes[-1] == 4: + if data_bytes[-1] == Term.EOT: # got an , done self.state = Mode.PRECOMMAND break - stx = data.find("\x02") + + # Search for start of frame + stx = data.find(Term.STX) + if stx != -1: - # got , parse frame + # Got , parse frame try: + dec = b"".join(data) result = self.checkframe(data[stx:]) - tometer = "\x06" + tometer = Term.ACK self.state = Mode.DATA except FrameError: - tometer = "\x15" # Couldn't parse, + tometer = Term.NAK # Couldn't parse, send else: # Got something we don't understand, it - tometer = "\x15" + tometer = Term.NAK except Exception as e: raise e From d662a0a6953ad0db03ed6a8692b86f2beefa32d9 Mon Sep 17 00:00:00 2001 From: Lorenz Gillner Date: Sat, 13 Jun 2026 18:08:36 +0200 Subject: [PATCH 3/8] Add Contour Care dump functionality --- .gitignore | 1 + glucometerutils/common.py | 5 +- glucometerutils/drivers/contourcare.py | 36 ++-- glucometerutils/support/contourcare.py | 225 +++++++++++++------------ 4 files changed, 135 insertions(+), 132 deletions(-) diff --git a/.gitignore b/.gitignore index e3e9439..0a7fcee 100644 --- a/.gitignore +++ b/.gitignore @@ -8,6 +8,7 @@ .cache .mypy_cache/ .vscode/ +.zed/ /MANIFEST /dist/ __pycache__/ diff --git a/glucometerutils/common.py b/glucometerutils/common.py index 24bc0d1..efb8a64 100644 --- a/glucometerutils/common.py +++ b/glucometerutils/common.py @@ -21,8 +21,9 @@ class Unit(enum.Enum): # Constants for meal information class Meal(enum.Enum): NONE = "" - BEFORE = "Before Meal" - AFTER = "After Meal" + BEFORE = "before meal" + AFTER = "after meal" + FASTING = "fasting" # Constants for measure method diff --git a/glucometerutils/drivers/contourcare.py b/glucometerutils/drivers/contourcare.py index 47ff7f1..5efe909 100644 --- a/glucometerutils/drivers/contourcare.py +++ b/glucometerutils/drivers/contourcare.py @@ -21,28 +21,17 @@ from glucometerutils import common from glucometerutils.support import contourcare - -def _extract_timestamp(parsed_record: dict[str, str]): - """Extract the timestamp from a parsed record. - - This leverages the fact that all the reading records have the same base structure. - """ - datetime_str = parsed_record["datetime"] - - return datetime.datetime( - int(datetime_str[0:4]), # year - int(datetime_str[4:6]), # month - int(datetime_str[6:8]), # day - int(datetime_str[8:10]), # hour - int(datetime_str[10:12]), # minute - int(datetime_str[12:14]), # second - 0, - ) +_MEAL_CODES = { + "T": common.Meal.NONE, + "B": common.Meal.BEFORE, + "A": common.Meal.AFTER, + "F": common.Meal.FASTING, +} class Device(contourcare.ContourCareHidDevice): """Glucometer driver for Contour Care devices.""" - + def __init__(self, device: Optional[str]) -> None: super().__init__(device) @@ -61,16 +50,19 @@ def get_readings(self) -> Generator[common.AnyReading, None, None]: This meter supports only blood samples """ for parsed_record in self._get_multirecord(): + timestamp = self.parse_timestamp(parsed_record["datetime"]) + value = float(parsed_record["value"]) + meal = _MEAL_CODES[parsed_record["meal"][0]] yield common.GlucoseReading( - _extract_timestamp(parsed_record), - int(parsed_record["value"]), - comment=parsed_record["markers"], + timestamp, + value, + meal, measure_method=common.MeasurementMethod.BLOOD_SAMPLE, ) def get_serial_number(self) -> str: return self._get_serial_number() - + def get_version(self): return self._get_version() diff --git a/glucometerutils/support/contourcare.py b/glucometerutils/support/contourcare.py index 53f30f1..b9a4481 100644 --- a/glucometerutils/support/contourcare.py +++ b/glucometerutils/support/contourcare.py @@ -21,11 +21,8 @@ from glucometerutils import driver from glucometerutils.support import hiddevice -# TODO this -# CR = 0x0d 0x0a (\r\n) -# Pipe = 0x7c (124) _RECORD_FORMAT_RE = re.compile( - r"\x02(?P(?P[0-7])(?P[^\x0d]*)\x0d(?P[\x03\x17]))" # haven't seen 0x03 yet + r"\x02(?P(?P[0-7])(?P[^\x0d]*)\x0d(?P[\x17\x03]))" r"(?P[0-9A-F][0-9A-F])\x0d\x0a" ) @@ -57,8 +54,9 @@ _RESULT_RECORD_RE = re.compile( r"^R\|(?P\d+)\|" r"\^\^\^Glucose\|" - r"(?P\d+\.\d+)\|(?P\w+)\^P\|\|" - r"(?P.+)\|\|" # ??? + r"(?P\d+\.\d+)\|" + r"(?P\w+\/\w+)\^(?P[BPD])\|\|" + r"(?P(\w+\/)?\w+)\|\|" r"(?P\d+)$" ) @@ -69,7 +67,7 @@ class FrameError(Exception): @enum.unique class Term(enum.IntEnum): - """ASTM E1394 vocabulary.""" + """ASTM E1394-97 vocabulary.""" PAD = 0x00 WAK = 0x58 @@ -93,26 +91,37 @@ class Mode(enum.Enum): class ContourCareHidDevice(driver.GlucometerDevice): """Base class implementing the Contour Care device.""" + USB_VENDOR_ID: int = 0x1A79 # Bayer Health Care LLC + USB_PRODUCT_ID: int = 0x7950 # Contour Care + blocksize: int = 64 state: Optional[Mode] = None currecno: Optional[int] = None - USB_VENDOR_ID: int = 0x1A79 # Bayer Health Care LLC - USB_PRODUCT_ID: int = 0x7950 # Contour Care - def __init__(self, device_path: Optional[str]) -> None: super().__init__(device_path) - _id = (self.USB_VENDOR_ID, self.USB_PRODUCT_ID) - self._hid_session = hiddevice.HidSession(_id, device_path) + hidid = (self.USB_VENDOR_ID, self.USB_PRODUCT_ID) + timeout = 200 + self._hid_session = hiddevice.HidSession(hidid, device_path, timeout) + + def connect(self): + """Connect to the device; handled by `hiddevice`.""" + pass + + def disconnect(self): + """Disconnect from the device; handled by `hiddevice`.""" + pass def read(self, r_size=blocksize) -> bytes: + """Read data via `hiddevice`.""" result = bytes() while True: data = self._hid_session.read() - dstr = data data_end_idx = data[3] + 4 - result += dstr[4:data_end_idx] + result += data[4:data_end_idx] + + # Data is smaller than block size; must be the last block if data[3] != self.blocksize - 4: break @@ -127,36 +136,6 @@ def write(self, message: Term) -> None: data += pad_length * pad self._hid_session.write(data) - def parse_header_record(self, text: str) -> None: - header = _HEADER_RECORD_RE.search(text) - - self.product_code = header.group("product_code") - self.dig_ver = header.group("dig_ver") - self.anlg_ver = header.group("anlg_ver") - self.agp_ver = header.group("agp_ver") - - self.serial_num = header.group("serial_num") - self.res_marking = header.group("res_marking") - self.config_bits = header.group("config_bits") - self.ref_method = header.group("ref_method") - self.internal = header.group("internal") - - # U limit - self.unit = header.group("unit") - self.lo_bound = header.group("lo_bound") - self.hi_bound = header.group("hi_bound") - - # X field - self.post_food_low = header.group("post_food_low") - self.pre_food_low = header.group("pre_food_low") - self.post_food_high = header.group("post_food_high") - self.pre_food_high = header.group("pre_food_high") - - self.total = header.group("total_recs") - - # Datetime string in YYYYMMDDHHMMSS format - self.datetime = header.group("datetime") - def checksum(self, text): """ Implemented by Anders Hammarquist for glucodump project @@ -165,13 +144,14 @@ def checksum(self, text): checksum = hex(sum(ord(c) for c in text) % 256).upper().split("X")[1] return ("00" + checksum)[-2:] - def checkframe(self, frame) -> Optional[str]: + def checkframe(self, frame: str) -> Optional[str]: """ Implemented by Anders Hammarquist for glucodump project More info: https://bitbucket.org/iko/glucodump/src/default/ """ match = _RECORD_FORMAT_RE.match(frame) - if not match: + + if match is None: raise FrameError("Couldn't parse frame", frame) recno = int(match.group("recno")) @@ -197,33 +177,74 @@ def checkframe(self, frame) -> Optional[str]: self.currecno = (self.currecno + 1) % 8 return match.group("text") - def connect(self): - """Connect to the device; handled by `hiddevice`.""" - pass + def parse_header_record(self, text: str) -> None: + """Parse a header record and set device properties.""" + header = _HEADER_RECORD_RE.search(text) + assert header is not None - def disconnect(self): - """Disconnect from the device; handled by `hiddevice`.""" - pass + self.product_code = header.group("product_code") + self.dig_ver = header.group("dig_ver") + self.anlg_ver = header.group("anlg_ver") + self.agp_ver = header.group("agp_ver") + + self.serial_num = header.group("serial_num") + self.res_marking = header.group("res_marking") + self.config_bits = header.group("config_bits") + self.ref_method = header.group("ref_method") + self.internal = header.group("internal") + + # U limit + self.unit = header.group("unit") + self.lo_bound = header.group("lo_bound") + self.hi_bound = header.group("hi_bound") + + # X field + self.post_food_low = header.group("post_food_low") + self.pre_food_low = header.group("pre_food_low") + self.post_food_high = header.group("post_food_high") + self.pre_food_high = header.group("pre_food_high") + + self.total = header.group("total_recs") + + # Datetime string in YYYYMMDDHHMMSS format + self.datetime = header.group("datetime") + + def parse_result_record(self, text: str) -> dict[str, str]: + """Parse a result record and return it as a dictionary.""" + result = _RESULT_RECORD_RE.search(text) + assert result is not None + return result.groupdict() + + def parse_timestamp(self, datetime_str: str) -> datetime.datetime: + """Extract the timestamp from a parsed record.""" + return datetime.datetime( + int(datetime_str[0:4]), # year + int(datetime_str[4:6]), # month + int(datetime_str[6:8]), # day + int(datetime_str[8:10]), # hour + int(datetime_str[10:12]), # minute + int(datetime_str[12:14]), # second + 0, + ) - def _get_info_record(self): + def _get_info_record(self) -> None: self.currecno = None self.state = Mode.ESTABLISH + try: while True: - # Contour Care answers to pretty much anything with a header ... - self.write(Term.ENQ) + # Send EOT to suppress further output; device will answer with a header anyway + self.write(Term.EOT) res = self.read() if res[0] == Term.EOT and res[-1] == Term.ENQ: - self.write(Term.ACK) - _ = self.read() - # We are connected and just got a header stx = res.find(Term.STX) if stx != -1: header_record = res[stx:-1].decode() result = _RECORD_FORMAT_RE.match(header_record).group("text") self.parse_header_record(result) + break else: pass @@ -236,9 +257,6 @@ def _get_info_record(self): print("Unknown error occured") raise e - # 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) -> str: """Return the software version of the device.""" return self.dig_ver + " - " + self.anlg_ver + " - " + self.agp_ver @@ -252,20 +270,10 @@ def _get_glucose_unit(self) -> str: return self.unit def get_datetime(self) -> datetime.datetime: - datetime_str = self.datetime - return datetime.datetime( - int(datetime_str[0:4]), # year - int(datetime_str[4:6]), # month - int(datetime_str[6:8]), # day - int(datetime_str[8:10]), # hour - int(datetime_str[10:12]), # minute - 0, - ) + return self.parse_timestamp(self.datetime) def sync(self) -> Generator[str, None, None]: - """ - Sync with meter and yield received data frames. - """ + """Sync with meter and yield received data frames.""" self.state = Mode.ESTABLISH try: # Send "wake up call" @@ -284,27 +292,29 @@ def sync(self) -> Generator[str, None, None]: yield result result = None - data_bytes = self.read() - data = data_bytes.decode() + data = self.read() if self.state == Mode.ESTABLISH: - if data_bytes[-1] == 15: - # got a , send - tometer = Term.EOT - foo += 1 - foo %= 256 - continue - - if data_bytes[-1] == Term.ENQ: - # got an , send - tometer = Term.ACK - self.currecno = None - # continue + match data[-1]: + case Term.NAK: + # Got a , send + tometer = Term.EOT + foo += 1 + foo %= 256 + continue + + case Term.ENQ: + # Got an , send + tometer = Term.ACK + self.currecno = None + # continue if self.state == Mode.DATA: - if data_bytes[-1] == Term.EOT: - # got an , done + if data[-1] == Term.EOT: + # Got an , done self.state = Mode.PRECOMMAND + tometer = Term.EOT + self.read() break # Search for start of frame @@ -313,8 +323,8 @@ def sync(self) -> Generator[str, None, None]: if stx != -1: # Got , parse frame try: - dec = b"".join(data) - result = self.checkframe(data[stx:]) + frame = bytes.decode(data[stx:]) + result = self.checkframe(frame) tometer = Term.ACK self.state = Mode.DATA except FrameError: @@ -322,29 +332,28 @@ def sync(self) -> Generator[str, None, None]: else: # Got something we don't understand, it tometer = Term.NAK + except Exception as e: raise e - def parse_result_record(self, text: str) -> dict[str, str]: - result = _RESULT_RECORD_RE.search(text) - assert result is not None - rec_text = result.groupdict() - return rec_text - def _get_multirecord(self) -> list[dict[str, str]]: """Queries for, and returns, "multirecords" results. Returns: - (csv.reader): a CSV reader object that returns a record for each line - in the record file. + A list of dictionaries, each representing a record from the record file. """ - records_arr = [] + records = [] + finished = False + for rec in self.sync(): - if rec[0] == "R": - # parse using result record regular expression - rec_text = self.parse_result_record(rec) - # get dictionary to use in main driver module without import re - - records_arr.append(rec_text) - # return csv.reader(records_arr) - return records_arr # array of groupdicts + match rec[0]: + case "R": + record = self.parse_result_record(rec) + records.append(record) + case "L": + # TODO handle L records properly + break + case _: + continue + + return records # array of groupdicts From 133448e134babaffe65e255f58c8d634596440aa Mon Sep 17 00:00:00 2001 From: Lorenz Gillner Date: Sat, 13 Jun 2026 19:01:14 +0200 Subject: [PATCH 4/8] Convert readings to mg/dL by default --- glucometerutils/drivers/contourcare.py | 7 ++++++- glucometerutils/support/contourcare.py | 4 +--- 2 files changed, 7 insertions(+), 4 deletions(-) diff --git a/glucometerutils/drivers/contourcare.py b/glucometerutils/drivers/contourcare.py index 5efe909..d805ce7 100644 --- a/glucometerutils/drivers/contourcare.py +++ b/glucometerutils/drivers/contourcare.py @@ -51,7 +51,12 @@ def get_readings(self) -> Generator[common.AnyReading, None, None]: """ for parsed_record in self._get_multirecord(): timestamp = self.parse_timestamp(parsed_record["datetime"]) - value = float(parsed_record["value"]) + # Apparently the GlucoseReadings expect mg/dL values, so convert if necessary + value = common.convert_glucose_unit( + float(parsed_record["value"]), + self.get_glucose_unit(), + common.Unit.MG_DL, + ) meal = _MEAL_CODES[parsed_record["meal"][0]] yield common.GlucoseReading( timestamp, diff --git a/glucometerutils/support/contourcare.py b/glucometerutils/support/contourcare.py index b9a4481..4afbace 100644 --- a/glucometerutils/support/contourcare.py +++ b/glucometerutils/support/contourcare.py @@ -275,13 +275,13 @@ def get_datetime(self) -> datetime.datetime: def sync(self) -> Generator[str, None, None]: """Sync with meter and yield received data frames.""" self.state = Mode.ESTABLISH + try: # Send "wake up call" self.write(Term.WAK) tometer = Term.ACK result = None - foo = 0 # Repeat until all records have been sent while True: @@ -299,8 +299,6 @@ def sync(self) -> Generator[str, None, None]: case Term.NAK: # Got a , send tometer = Term.EOT - foo += 1 - foo %= 256 continue case Term.ENQ: From 891646f7a7a38fbc7bbf7d18f8c5dc576d2a57e1 Mon Sep 17 00:00:00 2001 From: Lorenz Gillner Date: Sat, 13 Jun 2026 22:09:40 +0200 Subject: [PATCH 5/8] Fix continued transmission after termination --- glucometerutils/support/contourcare.py | 23 ++++++++++++----------- 1 file changed, 12 insertions(+), 11 deletions(-) diff --git a/glucometerutils/support/contourcare.py b/glucometerutils/support/contourcare.py index 4afbace..fd0d700 100644 --- a/glucometerutils/support/contourcare.py +++ b/glucometerutils/support/contourcare.py @@ -27,7 +27,7 @@ ) _HEADER_RECORD_RE = re.compile( - r"^H\|\\\^\&\|\|" # repeat, component, escape, field + r"H\|\\\^\&\|\|" # repeat, component, escape, field r"(?P\w{6})\|" r"(?P\w+)\^" r"(?P\d{2}\.\d{2})\\" @@ -46,20 +46,22 @@ r"J=(?P\d+)\|" r"(?P\d*)\|\|\|\|\|" r"[DPT]\|(?P\d+)\|" - r"(?P\d+)\|$" + r"(?P\d+)\|" ) -_PATIENT_RECORD_RE = re.compile(r"P\|\d+") - _RESULT_RECORD_RE = re.compile( - r"^R\|(?P\d+)\|" + r"R\|(?P\d+)\|" r"\^\^\^Glucose\|" r"(?P\d+\.\d+)\|" r"(?P\w+\/\w+)\^(?P[BPD])\|\|" r"(?P(\w+\/)?\w+)\|\|" - r"(?P\d+)$" + r"(?P\d+)" ) +_PATIENT_RECORD_RE = re.compile(r"P\|\d+") + +_TERMINATOR_RECORD_RE = re.compile(r"L\|1\|\|[NTERQIF]?") + class FrameError(Exception): pass @@ -76,6 +78,8 @@ class Term(enum.IntEnum): STX = 0x02 EOT = 0x04 NAK = 0x15 + ETB = 0x17 + ETX = 0x03 @enum.unique @@ -305,14 +309,12 @@ def sync(self) -> Generator[str, None, None]: # Got an , send tometer = Term.ACK self.currecno = None - # continue + continue if self.state == Mode.DATA: - if data[-1] == Term.EOT: - # Got an , done + if data[-5] == Term.ETX: self.state = Mode.PRECOMMAND tometer = Term.EOT - self.read() break # Search for start of frame @@ -341,7 +343,6 @@ def _get_multirecord(self) -> list[dict[str, str]]: A list of dictionaries, each representing a record from the record file. """ records = [] - finished = False for rec in self.sync(): match rec[0]: From 95909c9b33d011e7eb7148e2963c1e8e92823e66 Mon Sep 17 00:00:00 2001 From: Lorenz Gillner Date: Sat, 13 Jun 2026 23:57:34 +0200 Subject: [PATCH 6/8] Allow mmol/L as the default unit --- glucometerutils/common.py | 3 ++- glucometerutils/drivers/contourcare.py | 16 ++++++++-------- 2 files changed, 10 insertions(+), 9 deletions(-) diff --git a/glucometerutils/common.py b/glucometerutils/common.py index efb8a64..3a86a90 100644 --- a/glucometerutils/common.py +++ b/glucometerutils/common.py @@ -67,6 +67,7 @@ class GlucoseReading: validator=attr.validators.in_(MeasurementMethod), ) extra_data: dict[str, Any] = attr.Factory(dict) + unit: Unit = attr.ib(default=Unit.MG_DL, validator=attr.validators.in_(Unit)) def get_value_as(self, to_unit: Unit) -> float: """Returns the reading value as the given unit. @@ -74,7 +75,7 @@ def get_value_as(self, to_unit: Unit) -> float: Args: to_unit: The unit to return the value to. """ - return convert_glucose_unit(self.value, Unit.MG_DL, to_unit) + return convert_glucose_unit(self.value, self.unit, to_unit) def as_csv(self, unit: Unit) -> str: """Returns the reading as a formatted comma-separated value string.""" diff --git a/glucometerutils/drivers/contourcare.py b/glucometerutils/drivers/contourcare.py index d805ce7..a54c04b 100644 --- a/glucometerutils/drivers/contourcare.py +++ b/glucometerutils/drivers/contourcare.py @@ -28,6 +28,11 @@ "F": common.Meal.FASTING, } +_UNIT_CODES = { + "mg/dL": common.Unit.MG_DL, + "mmol/L": common.Unit.MMOL_L, +} + class Device(contourcare.ContourCareHidDevice): """Glucometer driver for Contour Care devices.""" @@ -51,18 +56,13 @@ def get_readings(self) -> Generator[common.AnyReading, None, None]: """ for parsed_record in self._get_multirecord(): timestamp = self.parse_timestamp(parsed_record["datetime"]) - # Apparently the GlucoseReadings expect mg/dL values, so convert if necessary - value = common.convert_glucose_unit( - float(parsed_record["value"]), - self.get_glucose_unit(), - common.Unit.MG_DL, - ) - meal = _MEAL_CODES[parsed_record["meal"][0]] + value = float(parsed_record["value"]) yield common.GlucoseReading( timestamp, value, - meal, + _MEAL_CODES[parsed_record["meal"][0]], measure_method=common.MeasurementMethod.BLOOD_SAMPLE, + unit=_UNIT_CODES[parsed_record["unit"]], ) def get_serial_number(self) -> str: From 34e0b38557328f55392416541bee9feaf4619946 Mon Sep 17 00:00:00 2001 From: Lorenz Gillner Date: Sun, 14 Jun 2026 15:40:08 +0200 Subject: [PATCH 7/8] Add terminator record parsing --- .gitignore | 1 + README.md | 1 + glucometerutils/driver.py | 8 +++ glucometerutils/drivers/contourcare.py | 10 ++- glucometerutils/support/contourcare.py | 93 ++++++++++++++++---------- 5 files changed, 76 insertions(+), 37 deletions(-) diff --git a/.gitignore b/.gitignore index 0a7fcee..225375e 100644 --- a/.gitignore +++ b/.gitignore @@ -16,3 +16,4 @@ build .idea/ *venv/ *.swp +.DS_Store diff --git a/README.md b/README.md index beca0ba..38032cf 100644 --- a/README.md +++ b/README.md @@ -69,6 +69,7 @@ supported. | Menarini | GlucoMen Nexus | `td42xx` | [construct] [pyserial]² [hidapi] | | Aktivmed | GlucoCheck XL | `td42xx` | [construct] [pyserial]² [hidapi] | | Ascensia | ContourUSB | `contourusb` | [construct] [hidapi]‡ | +| Ascensia | Contour Care | `contourcare` | [construct] [hidapi]‡ | | Menarini | GlucoMen areo³ | `glucomenareo` | [pyserial] [crcmod] | † Untested. diff --git a/glucometerutils/driver.py b/glucometerutils/driver.py index 5f524e9..b1530b8 100644 --- a/glucometerutils/driver.py +++ b/glucometerutils/driver.py @@ -70,6 +70,14 @@ def zero_log(self) -> None: def get_readings(self) -> Generator[common.AnyReading, None, None]: pass + @abc.abstractmethod + def get_patient_name(self) -> Optional[str]: + pass + + @abc.abstractmethod + def set_patient_name(self, name: str) -> None: + pass + @dataclasses.dataclass class Driver: diff --git a/glucometerutils/drivers/contourcare.py b/glucometerutils/drivers/contourcare.py index a54c04b..f608f0a 100644 --- a/glucometerutils/drivers/contourcare.py +++ b/glucometerutils/drivers/contourcare.py @@ -33,12 +33,14 @@ "mmol/L": common.Unit.MMOL_L, } +PRODUCT_ID: int = 0x7950 # Contour Care + class Device(contourcare.ContourCareHidDevice): """Glucometer driver for Contour Care devices.""" def __init__(self, device: Optional[str]) -> None: - super().__init__(device) + super().__init__(PRODUCT_ID, device) def get_meter_info(self) -> common.MeterInfo: self._get_info_record() @@ -82,3 +84,9 @@ def _set_device_datetime(self, date: datetime.datetime) -> NoReturn: def zero_log(self) -> NoReturn: raise NotImplementedError + + def get_patient_name(self) -> str: + raise NotImplementedError + + def set_patient_name(self, name: str) -> None: + raise NotImplementedError diff --git a/glucometerutils/support/contourcare.py b/glucometerutils/support/contourcare.py index fd0d700..490c3a1 100644 --- a/glucometerutils/support/contourcare.py +++ b/glucometerutils/support/contourcare.py @@ -50,7 +50,7 @@ ) _RESULT_RECORD_RE = re.compile( - r"R\|(?P\d+)\|" + r"R\|(?P\d+)\|" r"\^\^\^Glucose\|" r"(?P\d+\.\d+)\|" r"(?P\w+\/\w+)\^(?P[BPD])\|\|" @@ -58,9 +58,12 @@ r"(?P\d+)" ) -_PATIENT_RECORD_RE = re.compile(r"P\|\d+") +_PATIENT_RECORD_RE = re.compile(r"P\|(?P\d+)\|") -_TERMINATOR_RECORD_RE = re.compile(r"L\|1\|\|[NTERQIF]?") +_TERMINATOR_RECORD_RE = re.compile(r"L\|1\|\|(?P[NTERQIF]?)") + +VENDOR_ID: int = 0x1A79 # Bayer Health Care LLC +TIMEOUT_MS: int = 200 class FrameError(Exception): @@ -68,18 +71,31 @@ class FrameError(Exception): @enum.unique -class Term(enum.IntEnum): - """ASTM E1394-97 vocabulary.""" +class Control(enum.IntEnum): + """ASCII Control characters.""" PAD = 0x00 - WAK = 0x58 - ENQ = 0x05 - ACK = 0x06 STX = 0x02 + ETX = 0x03 EOT = 0x04 + ENQ = 0x05 + ACK = 0x06 NAK = 0x15 ETB = 0x17 - ETX = 0x03 + WAK = 0x58 + + +@enum.unique +class TerminatorCode(enum.Enum): + """Terminator record return codes.""" + + NORMAL = "N" + SENDER_ABORT = "T" + RECEIVER_ABORT = "R" + UNKNOWN = "E" + REQUEST_ERROR = "Q" + NO_INFORMATION = "I" + LAST_REQUEST = "F" @enum.unique @@ -95,18 +111,14 @@ class Mode(enum.Enum): class ContourCareHidDevice(driver.GlucometerDevice): """Base class implementing the Contour Care device.""" - USB_VENDOR_ID: int = 0x1A79 # Bayer Health Care LLC - USB_PRODUCT_ID: int = 0x7950 # Contour Care - blocksize: int = 64 state: Optional[Mode] = None currecno: Optional[int] = None - def __init__(self, device_path: Optional[str]) -> None: + def __init__(self, product_id: int, device_path: Optional[str]) -> None: super().__init__(device_path) - hidid = (self.USB_VENDOR_ID, self.USB_PRODUCT_ID) - timeout = 200 - self._hid_session = hiddevice.HidSession(hidid, device_path, timeout) + hidid = (VENDOR_ID, product_id) + self._hid_session = hiddevice.HidSession(hidid, device_path, TIMEOUT_MS) def connect(self): """Connect to the device; handled by `hiddevice`.""" @@ -131,8 +143,8 @@ def read(self, r_size=blocksize) -> bytes: return result - def write(self, message: Term) -> None: - pad = bytes([Term.PAD]) + def write(self, message: Control) -> None: + pad = bytes([Control.PAD]) data = 4 * pad data += chr(1).encode() data += bytes([message]) @@ -219,6 +231,12 @@ def parse_result_record(self, text: str) -> dict[str, str]: assert result is not None return result.groupdict() + def parse_terminator_record(self, text: str) -> TerminatorCode: + """Parse a terminator record and return it as a dictionary.""" + result = _TERMINATOR_RECORD_RE.search(text) + assert result is not None + return TerminatorCode(result.group("code")) + def parse_timestamp(self, datetime_str: str) -> datetime.datetime: """Extract the timestamp from a parsed record.""" return datetime.datetime( @@ -238,16 +256,17 @@ def _get_info_record(self) -> None: try: while True: # Send EOT to suppress further output; device will answer with a header anyway - self.write(Term.EOT) + self.write(Control.EOT) res = self.read() - if res[0] == Term.EOT and res[-1] == Term.ENQ: + if res[0] == Control.EOT and res[-1] == Control.ENQ: # We are connected and just got a header - stx = res.find(Term.STX) + stx = res.find(Control.STX) if stx != -1: header_record = res[stx:-1].decode() - result = _RECORD_FORMAT_RE.match(header_record).group("text") - self.parse_header_record(result) + result = _RECORD_FORMAT_RE.match(header_record) + assert result is not None + self.parse_header_record(result.group("text")) break else: @@ -282,9 +301,9 @@ def sync(self) -> Generator[str, None, None]: try: # Send "wake up call" - self.write(Term.WAK) + self.write(Control.WAK) - tometer = Term.ACK + tometer = Control.ACK result = None # Repeat until all records have been sent @@ -300,38 +319,38 @@ def sync(self) -> Generator[str, None, None]: if self.state == Mode.ESTABLISH: match data[-1]: - case Term.NAK: + case Control.NAK: # Got a , send - tometer = Term.EOT + tometer = Control.EOT continue - case Term.ENQ: + case Control.ENQ: # Got an , send - tometer = Term.ACK + tometer = Control.ACK self.currecno = None continue if self.state == Mode.DATA: - if data[-5] == Term.ETX: + if data[-5] == Control.ETX: self.state = Mode.PRECOMMAND - tometer = Term.EOT + tometer = Control.EOT break # Search for start of frame - stx = data.find(Term.STX) + stx = data.find(Control.STX) if stx != -1: # Got , parse frame try: frame = bytes.decode(data[stx:]) result = self.checkframe(frame) - tometer = Term.ACK + tometer = Control.ACK self.state = Mode.DATA except FrameError: - tometer = Term.NAK # Couldn't parse, send + tometer = Control.NAK # Couldn't parse, send else: # Got something we don't understand, it - tometer = Term.NAK + tometer = Control.NAK except Exception as e: raise e @@ -350,7 +369,9 @@ def _get_multirecord(self) -> list[dict[str, str]]: record = self.parse_result_record(rec) records.append(record) case "L": - # TODO handle L records properly + code = self.parse_terminator_record(rec) + if code != TerminatorCode.NORMAL: + raise ValueError(f"Unexpected terminator code: {code}") break case _: continue From bac2752f44029b04ac4ec8a91eddd218bc40b90a Mon Sep 17 00:00:00 2001 From: Lorenz Gillner Date: Sun, 14 Jun 2026 16:11:50 +0200 Subject: [PATCH 8/8] Add `contourcare` dependencies --- setup.py | 1 + 1 file changed, 1 insertion(+) diff --git a/setup.py b/setup.py index 88c17e3..55f903e 100644 --- a/setup.py +++ b/setup.py @@ -12,6 +12,7 @@ # listed as mandatory for the feature. "accucheck_reports": [], "contourusb": ["construct", "hidapi"], + "contourcare": ["construct", "hidapi"], "fsfreedomlite": ["pyserial"], "fsinsulinx": ["freestyle-hid>=1.0.2"], "fslibre": ["freestyle-hid>=1.0.2"],