|
| 1 | +import datetime |
| 2 | +import hashlib |
| 3 | +import json |
| 4 | +import urllib.request |
| 5 | +import webbrowser |
| 6 | +from string import Template |
| 7 | +from typing import Callable |
| 8 | + |
| 9 | +from PyQt6.QtCore import QThread, QTimer, pyqtSignal |
| 10 | +from PyQt6.QtWidgets import QFrame, QHBoxLayout |
| 11 | + |
| 12 | +from core.utils.tooltip import set_tooltip |
| 13 | +from core.utils.utilities import build_widget_label |
| 14 | +from core.validation.widgets.yasb.glucose_monitor import VALIDATION_SCHEMA |
| 15 | +from core.widgets.base import BaseWidget |
| 16 | + |
| 17 | + |
| 18 | +class GlucoseMonitorWorker(QThread): |
| 19 | + @classmethod |
| 20 | + def get_instance(cls): |
| 21 | + return cls() |
| 22 | + |
| 23 | + status_updated = pyqtSignal(int, float, str, str) |
| 24 | + error_signal = pyqtSignal(str) |
| 25 | + |
| 26 | + def __init__(self, parent=None) -> None: |
| 27 | + super().__init__(parent) |
| 28 | + |
| 29 | + self._url: str | None = None |
| 30 | + self.running = True |
| 31 | + |
| 32 | + def set_url(self, host: str, secret: str) -> None: |
| 33 | + secret_hash = hashlib.sha1(secret.encode()).hexdigest() |
| 34 | + self._url = f"{host}/api/v1/entries/current.json?secret={secret_hash}" |
| 35 | + |
| 36 | + def stop(self) -> None: |
| 37 | + self.running = False |
| 38 | + self.wait() |
| 39 | + |
| 40 | + def run(self) -> None: |
| 41 | + if not self.running: |
| 42 | + return |
| 43 | + |
| 44 | + try: |
| 45 | + with urllib.request.urlopen(self._url) as response: |
| 46 | + data = json.loads(response.read().decode("utf-8")) |
| 47 | + status = response.status |
| 48 | + |
| 49 | + if status != 200: |
| 50 | + raise RuntimeError(f"Response status code should be 200 but got {status}") |
| 51 | + |
| 52 | + resp_json = data[0] |
| 53 | + self.status_updated.emit( |
| 54 | + resp_json["sgv"], |
| 55 | + resp_json["delta"], |
| 56 | + resp_json["dateString"], |
| 57 | + resp_json["direction"], |
| 58 | + ) |
| 59 | + except Exception as e: |
| 60 | + self.error_signal.emit(str(e)) |
| 61 | + |
| 62 | + |
| 63 | +class GlucoseMonitor(BaseWidget): |
| 64 | + validation_schema = VALIDATION_SCHEMA |
| 65 | + |
| 66 | + update_interval_in_milliseconds = 1 * 60 * 1_000 |
| 67 | + datetime_format = "%Y-%m-%dT%H:%M:%S.%f%z" |
| 68 | + |
| 69 | + direction_icons_mapping = { |
| 70 | + "double_up": "DoubleUp", |
| 71 | + "single_up": "SingleUp", |
| 72 | + "forty_five_up": "FortyFiveUp", |
| 73 | + "flat": "Flat", |
| 74 | + "forty_five_down": "FortyFiveDown", |
| 75 | + "single_down": "SingleDown", |
| 76 | + "double_down": "DoubleDown", |
| 77 | + } |
| 78 | + |
| 79 | + def __init__( |
| 80 | + self, |
| 81 | + label: str, |
| 82 | + tooltip: str, |
| 83 | + host: str, |
| 84 | + secret: str, |
| 85 | + direction_icons: dict[str, str], |
| 86 | + sgv_measurement_units: str, |
| 87 | + callbacks: dict[str, str], |
| 88 | + container_padding: dict[str, str], |
| 89 | + ) -> None: |
| 90 | + super().__init__(timer_interval=self.update_interval_in_milliseconds, class_name="cgm-widget") |
| 91 | + |
| 92 | + self._label_content = label |
| 93 | + self._label_template = Template(label) |
| 94 | + |
| 95 | + self._tooltip_label_content = tooltip |
| 96 | + self._tooltip_label_template = Template(tooltip) |
| 97 | + |
| 98 | + self._host = host |
| 99 | + self._secret = secret |
| 100 | + |
| 101 | + self._direction_icons = {self.direction_icons_mapping[key]: value for key, value in direction_icons.items()} |
| 102 | + |
| 103 | + self._sgv_measurement_units = sgv_measurement_units |
| 104 | + self._available_sgv_measurement_units: dict[str, Callable[[int | float], str]] = { |
| 105 | + "mg/dl": lambda sgv: str(round(sgv)), |
| 106 | + "mmol/l": lambda sgv: f"{sgv / 18:.1f}", |
| 107 | + } |
| 108 | + self._padding = container_padding |
| 109 | + |
| 110 | + self._widget_container_layout = QHBoxLayout() |
| 111 | + self._widget_container_layout.setSpacing(0) |
| 112 | + self._widget_container_layout.setContentsMargins( |
| 113 | + self._padding["left"], |
| 114 | + self._padding["top"], |
| 115 | + self._padding["right"], |
| 116 | + self._padding["bottom"], |
| 117 | + ) |
| 118 | + |
| 119 | + self._widget_container = QFrame() |
| 120 | + self._widget_container.setLayout(self._widget_container_layout) |
| 121 | + self._widget_container.setProperty("class", "widget-container") |
| 122 | + self.widget_layout.addWidget(self._widget_container) |
| 123 | + |
| 124 | + self._label_content = self._label_template.substitute( |
| 125 | + { |
| 126 | + "sgv": "...", |
| 127 | + "sgv_delta": "...", |
| 128 | + "delta_time_in_minutes": "..", |
| 129 | + "direction": "..", |
| 130 | + } |
| 131 | + ) |
| 132 | + build_widget_label(self, self._label_content) |
| 133 | + |
| 134 | + self.register_callback("open_cgm", self._open_cgm) |
| 135 | + |
| 136 | + self.callback_left = callbacks["on_left"] |
| 137 | + self.callback_right = callbacks["on_right"] |
| 138 | + self.callback_middle = callbacks["on_middle"] |
| 139 | + |
| 140 | + self._worker = GlucoseMonitorWorker.get_instance() |
| 141 | + self._worker.set_url(self._host, self._secret) |
| 142 | + self._worker.status_updated.connect(self._handle_status_update) |
| 143 | + self._worker.error_signal.connect(self._handle_error_signal) |
| 144 | + |
| 145 | + self._update_timer = QTimer() |
| 146 | + self._update_timer.timeout.connect(self._worker.start) |
| 147 | + self._update_timer.start(self.update_interval_in_milliseconds) |
| 148 | + |
| 149 | + self._worker.start() |
| 150 | + |
| 151 | + def _open_cgm(self) -> None: |
| 152 | + webbrowser.open(self._host) |
| 153 | + |
| 154 | + def _update_label(self, text: str) -> None: |
| 155 | + self._widgets[0].setText(text) |
| 156 | + |
| 157 | + def _handle_error_signal(self, message: str) -> None: |
| 158 | + message = f"❌{message}❌" |
| 159 | + self._update_label(message) |
| 160 | + set_tooltip( |
| 161 | + widget=self._widget_container, |
| 162 | + text=message, |
| 163 | + ) |
| 164 | + |
| 165 | + def _handle_status_update( |
| 166 | + self, |
| 167 | + sgv: int, |
| 168 | + sgv_delta: float, |
| 169 | + date_string: str, |
| 170 | + direction: str, |
| 171 | + ) -> None: |
| 172 | + now = datetime.datetime.now(tz=datetime.timezone.utc) |
| 173 | + last_update_time = datetime.datetime.strptime(date_string, self.datetime_format) |
| 174 | + delta_time_in_minutes = int((now - last_update_time).total_seconds() // 60) |
| 175 | + direction = self._direction_icons[direction] |
| 176 | + |
| 177 | + if not (convert_sgv := self._available_sgv_measurement_units.get(self._sgv_measurement_units)): |
| 178 | + self._handle_error_signal("Wrong measurement units") |
| 179 | + |
| 180 | + sgv = convert_sgv(sgv) |
| 181 | + sgv_delta = convert_sgv(sgv_delta) |
| 182 | + |
| 183 | + self._label_content = self._label_template.substitute( |
| 184 | + { |
| 185 | + "sgv": sgv, |
| 186 | + "sgv_delta": sgv_delta, |
| 187 | + "delta_time_in_minutes": delta_time_in_minutes, |
| 188 | + "direction": direction, |
| 189 | + } |
| 190 | + ) |
| 191 | + self._update_label(self._label_content) |
| 192 | + |
| 193 | + self._tooltip_label_content = self._tooltip_label_template.safe_substitute( |
| 194 | + { |
| 195 | + "sgv": sgv, |
| 196 | + "sgv_delta": sgv_delta, |
| 197 | + "delta_time_in_minutes": delta_time_in_minutes, |
| 198 | + "direction": direction, |
| 199 | + } |
| 200 | + ) |
| 201 | + set_tooltip( |
| 202 | + widget=self._widget_container, |
| 203 | + text=self._tooltip_label_content, |
| 204 | + ) |
0 commit comments