# Copyright (C) 2022 The Qt Company Ltd. # SPDX-License-Identifier: LicenseRef-Qt-Commercial OR BSD-3-Clause import struct from enum import IntEnum from PySide6.QtBluetooth import (QLowEnergyCharacteristic, QLowEnergyController, QLowEnergyDescriptor, QLowEnergyService, QBluetoothUuid) from PySide6.QtQml import QmlNamedElement, QmlUncreatable from PySide6.QtCore import (QByteArray, QDateTime, QRandomGenerator, QTimer, Property, Signal, Slot, QEnum) from bluetoothbaseclass import BluetoothBaseClass from heartrate_global import simulator # To be used on the @QmlElement decorator # (QML_IMPORT_MINOR_VERSION is optional) QML_IMPORT_NAME = "Shared" QML_IMPORT_MAJOR_VERSION = 1 @QmlNamedElement("AddressType") @QmlUncreatable("Enum is not a type") class DeviceHandler(BluetoothBaseClass): @QEnum class AddressType(IntEnum): PUBLIC_ADDRESS = 1 RANDOM_ADDRESS = 2 measuringChanged = Signal() aliveChanged = Signal() statsChanged = Signal() def __init__(self, parent=None): super().__init__(parent) self.m_control = None self.m_service = None self.m_notificationDesc = QLowEnergyDescriptor() self.m_currentDevice = None self.m_foundHeartRateService = False self.m_measuring = False self.m_currentValue = 0 self.m_min = 0 self.m_max = 0 self.m_sum = 0 self.m_avg = 0.0 self.m_calories = 0.0 self.m_start = QDateTime() self.m_stop = QDateTime() self.m_measurements = [] self.m_addressType = QLowEnergyController.PublicAddress self.m_demoTimer = QTimer() if simulator: self.m_demoTimer.setSingleShot(False) self.m_demoTimer.setInterval(2000) self.m_demoTimer.timeout.connect(self.updateDemoHR) self.m_demoTimer.start() self.updateDemoHR() @Property(int) def addressType(self): if self.m_addressType == QLowEnergyController.RandomAddress: return DeviceHandler.AddressType.RANDOM_ADDRESS return DeviceHandler.AddressType.PUBLIC_ADDRESS @addressType.setter def addressType(self, type): if type == DeviceHandler.AddressType.PUBLIC_ADDRESS: self.m_addressType = QLowEnergyController.PublicAddress elif type == DeviceHandler.AddressType.RANDOM_ADDRESS: self.m_addressType = QLowEnergyController.RandomAddress @Slot(QLowEnergyController.Error) def controllerErrorOccurred(self, device): self.error = "Cannot connect to remote device." @Slot() def controllerConnected(self): self.info = "Controller connected. Search services..." self.m_control.discoverServices() @Slot() def controllerDisconnected(self): self.error = "LowEnergy controller disconnected" def setDevice(self, device): self.clearMessages() self.m_currentDevice = device if simulator: self.info = "Demo device connected." return # Disconnect and delete old connection if self.m_control: self.m_control.disconnectFromDevice() m_control = None # Create new controller and connect it if device available if self.m_currentDevice: # Make connections #! [Connect-Signals-1] self.m_control = QLowEnergyController.createCentral(self.m_currentDevice.getDevice(), self) #! [Connect-Signals-1] self.m_control.setRemoteAddressType(self.m_addressType) #! [Connect-Signals-2] m_control.serviceDiscovered.connect(self.serviceDiscovered) m_control.discoveryFinished.connect(self.serviceScanDone) self.m_control.errorOccurred.connect(self.controllerErrorOccurred) self.m_control.connected.connect(self.controllerConnected) self.m_control.disconnected.connect(self.controllerDisconnected) # Connect self.m_control.connectToDevice() #! [Connect-Signals-2] @Slot() def startMeasurement(self): if self.alive: self.m_start = QDateTime.currentDateTime() self.m_min = 0 self.m_max = 0 self.m_avg = 0 self.m_sum = 0 self.m_calories = 0.0 self.m_measuring = True self.m_measurements.clear() self.measuringChanged.emit() @Slot() def stopMeasurement(self): self.m_measuring = False self.measuringChanged.emit() #! [Filter HeartRate service 1] @Slot(QBluetoothUuid) def serviceDiscovered(self, gatt): if gatt == QBluetoothUuid(QBluetoothUuid.ServiceClassUuid.HeartRate): self.info = "Heart Rate service discovered. Waiting for service scan to be done..." self.m_foundHeartRateService = True #! [Filter HeartRate service 1] @Slot() def serviceScanDone(self): self.info = "Service scan done." # Delete old service if available if self.m_service: self.m_service = None #! [Filter HeartRate service 2] # If heartRateService found, create new service if self.m_foundHeartRateService: self.m_service = self.m_control.createServiceObject(QBluetoothUuid(QBluetoothUuid.ServiceClassUuid.HeartRate), self) if self.m_service: self.m_service.stateChanged.connect(self.serviceStateChanged) self.m_service.characteristicChanged.connect(self.updateHeartRateValue) self.m_service.descriptorWritten.connect(self.confirmedDescriptorWrite) self.m_service.discoverDetails() else: self.error = "Heart Rate Service not found." #! [Filter HeartRate service 2] # Service functions #! [Find HRM characteristic] @Slot(QLowEnergyService.ServiceState) def serviceStateChanged(self, switch): if switch == QLowEnergyService.RemoteServiceDiscovering: self.setInfo(tr("Discovering services...")) elif switch == QLowEnergyService.RemoteServiceDiscovered: self.setInfo(tr("Service discovered.")) hrChar = m_service.characteristic(QBluetoothUuid(QBluetoothUuid.CharacteristicType.HeartRateMeasurement)) if hrChar.isValid(): self.m_notificationDesc = hrChar.descriptor(QBluetoothUuid.DescriptorType.ClientCharacteristicConfiguration) if self.m_notificationDesc.isValid(): self.m_service.writeDescriptor(m_notificationDesc, QByteArray.fromHex(b"0100")) else: self.error = "HR Data not found." self.aliveChanged.emit() #! [Find HRM characteristic] #! [Reading value] @Slot(QLowEnergyCharacteristic, QByteArray) def updateHeartRateValue(self, c, value): # ignore any other characteristic change. Shouldn't really happen though if c.uuid() != QBluetoothUuid(QBluetoothUuid.CharacteristicType.HeartRateMeasurement): return data = value.data() flags = int(data[0]) # Heart Rate hrvalue = 0 if flags & 0x1: # HR 16 bit little endian? otherwise 8 bit hrvalue = struct.unpack(" 30 and value < 250: self.m_stop = QDateTime.currentDateTime() self.m_measurements.append(value) self.m_min = value if self.m_min == 0 else min(value, self.m_min) self.m_max = max(value, self.m_max) self.m_sum += value self.m_avg = float(self.m_sum) / len(self.m_measurements) self.m_calories = ((-55.0969 + (0.6309 * self.m_avg) + (0.1988 * 94) + (0.2017 * 24)) / 4.184) * 60 * self.time / 3600 self.statsChanged.emit()