Vielleicht hilft das hier noch dazu.
Ein Code aus dem RuuviTag Forum, wo ein Kollege es wohl mit Python hinbekommen hat.
"""
Verbindung zu RuuviTag über Bluetooth herstellen und Sensorprotokolldaten lesen
# Code-Stack
* python (Version 3.x, getestet mit python 3.10.7)
* bleak (python bluetooth library)
https://bleak.readthedocs.io/en/latest/usage.html
Getestet auf Linux (64-Bit) mit Kernel 5.19.7 und bluez 5.65-3
# Einrichtung
Einige Python-Bibliotheken (bleak) werden benötigt, installieren Sie diese mit Ihrem Paketmanager
ODER
Erstellen Sie eine virtuelle Umgebung:
python -m venv env
Quelle env/bin/activate
pip install bleak
# Ausführen
python ./ruuvitag_read_log_data.py
# Worum geht es hier eigentlich?
Der RuuviTag ist ein drahtloser Sensor für Temperatur, Luftfeuchtigkeit, Luftdruck und Bewegung.
*
https://ruuvi.com/ruuvitag/
Er speichert Messdaten (Temperatur, Luftfeuchtigkeit und Luftdruck) im internen Speicher.
Das Standardintervall für die Aufzeichnung beträgt 5 Minuten.
Die Longlife/Longmem-Firmware für den RuuviTag erhöht das Protokollintervall auf 15 Minuten.
Um die Firmware aus dem RuuviTag auszulesen, muss zunächst eine Bluetooth-Verbindung hergestellt werden.
Der RuuviTag stellt den NUS (Nordic UART Service) zur Verfügung, um Daten vom/zum Gerät zu lesen/schreiben.
*
https://docs.ruuvi.com/communication/bluetooth-connection/nordic-uart-service-nus
Mit dem Befehl read log data sendet der Tag die geloggten Daten zurück.
*
https://docs.ruuvi.com/communication/bluetooth-connection/nordic-uart-service-nus/log-read
Die Daten werden beim Lesen des Logs _nicht_ vom RuuviTag gelöscht. Sie werden automatisch
überschrieben, wenn der Speicherplatz aufgebraucht ist (sollte bei der Standard-Firmware 10 Tage betragen).
"""
## Die Sensormessungen, die vom RuuviTag abgefragt werden können (unten von DATATYPE_LOG verwendet)
TEMPERATURE = 0x30 # \x30 Temperature [0,01°C]
HUMIDITY = 0x31 # \x31 Humidity [0,01%rH]
AIR_PRESSURE = 0x32 # \x32 Air pressure [1Pa]
ALL_SENSORS = 0x3A # \x3A All of the above
## Nordic UART Service UUIDs, ändern Sie diese _nicht_.
UART_SERVICE_UUID = "6E400001-B5A3-F393-E0A9-E50E24DCCA9E"
UART_RX_CHAR_UUID = "6E400002-B5A3-F393-E0A9-E50E24DCCA9E"
UART_TX_CHAR_UUID = "6E400003-B5A3-F393-E0A9-E50E24DCCA9E"
## Kopfdatentyp für Anzeigendaten, _nicht_ ändern
DATATYPE_ADV = 0x05
### SETTINGS AREA BEGIN ###
## Enter the MAC address of your RuuviTag
DEVICE_MAC = "00:00:00:00:00:00"
## Anzahl der Sekunden zum Abrufen von Protokolldaten
LOG_SECONDS = 7200 # 7200 = last 2 hours
## Wählen Sie die vom RuuviTag zu lesenden Protokolldaten (TEMPERATUR oder FEUCHTIGKEIT oder LUFTDRUCK oder ALLE_SENSOREN)
DATATYPE_LOG = ALL_SENSORS
### SETTINGS AREA END ###
import asyncio
import sys
import calendar
import time
import struct
import bleak
from datetime import datetime, timezone
log_data_end_of_data = False
async def ruuvi_read_log_data():
global log_data_end_of_data
def handle_disconnect(_: bleak.BleakClient):
print("Device was disconnected", flush=True)
# Das Abbrechen aller Aufgaben beendet das Programm effektiv.
for task in asyncio.all_tasks():
task.cancel()
# vom Bluetooth-Client empfangene Daten, angefordert von start_notify()
def handle_rx(sender_handle: int, data: bytearray):
global log_data_end_of_data
if len(data) <= 0:
print(f"no data (size == {len(data)})")
else:
if (data[0] == DATATYPE_LOG):
if len(data) != 11:
print("Header = log data, but wrong data size", flush=True)
else:
# unpack binary data
dat = struct.unpack('>BBBII', data)
if (dat[3] == 0xFFFFFFFF and dat[4] == 0xFFFFFFFF):
print(" Log: end of output received", flush=True)
log_data_end_of_data = True
else:
if (dat[1] == TEMPERATURE):
print(f" Log data: {time.strftime('%Y-%m-%d %H:%M:%S', time.localtime(dat[3]))} {dat[4]/100.0:.2f}°C", flush=True)
elif (dat[1] == HUMIDITY):
print(f" Log data: {time.strftime('%Y-%m-%d %H:%M:%S', time.localtime(dat[3]))} {dat[4]/100.0}%rH", flush=True)
elif (dat[1] == AIR_PRESSURE):
print(f" Log data: {time.strftime('%Y-%m-%d %H:%M:%S', time.localtime(dat[3]))} {dat[4]/100.0}mBar", flush=True)
elif data[0] == DATATYPE_ADV:
if len(data) != 18:
print("Header = advertisement data, but wrong size", flush=True)
else:
dat = struct.unpack('>BhHHhhhHBH', data)
print(f" Advertisement data received: "
f"T={dat[1]/200.0:.2f}°C "
f"H={dat[2]/400.0:.1f}%rH "
f"P={(dat[3] + 50000)/100.0:.2f}mBar "
f"[x={dat[4]:+5d} y={dat[5]:+5d} z={dat[6]:+5d}] "
f"U={((dat[7]/32)+1600)/1000.0:.2f}V "
f"Tx_strength={(dat[7]%32)*2 - 40:+d}dBm "
f"movement_counter={dat[8]:3d} "
f"sequence_counter={dat[9]:5d}", flush=True)
else:
print(f"unknown first byte: {hex(data[0])}")
# Dieser Code wird beim Aufruf von ruuvi_read_log_data() ausgeführt
print(f"Searching for RuuviTag {DEVICE_MAC}", flush=True)
async with bleak.BleakClient(DEVICE_MAC, disconnected_callback=handle_disconnect, timeout=30) as client:
print(f"Connected to Ruuvi, starting notify on UART channel", flush=True)
await client.start_notify(UART_TX_CHAR_UUID, handle_rx)
# Anforderungsdatenkopf erstellen
data_tx = b''
data_tx += bytes([DATATYPE_LOG])
data_tx += bytes([DATATYPE_LOG])
data_tx += b'\x11'
# timestampts
timenow = int(time.time()) # aktuelle Systemzeit ermitteln
timeprv = timenow - LOG_SECONDS
data_tx += struct.pack('>I', timenow)
data_tx += struct.pack('>I', timeprv)
print(f"Requesting log data for the last {LOG_SECONDS} seconds")
await client.write_gatt_char(UART_RX_CHAR_UUID, data_tx)
# wait for data
while True:
if log_data_end_of_data:
print(f"All log data received, closing connection")
client.close()
break
await asyncio.sleep(1)
if __name__ == "__main__":
try:
asyncio.run(ruuvi_read_log_data())
except asyncio.CancelledError:
# Aufgabe wird beim Trennen der Verbindung abgebrochen, daher ignorieren wir diesen Fehler
pass
except bleak.exc.BleakDBusError as error:
## bleak.exc.BleakDBusError: [org.bluez.Error.NotReady] Resource Not Ready
## bluetoothctl power on
## bleak.exc.BleakDBusError: [org.freedesktop.systemd1.NoSuchUnit] Unit dbus-org.bluez.service not found.
## systemctl start bluetooth.service
print("bleak.exc.BleakDBusError")
print(error)
except bleak.exc.BleakError as error:
## bleak.exc.BleakError: Not connected
print("bleak.exc.BleakError")
print(error)
except:
print("unknown exception")
raise # Wiedererhöhung der Ausnahme für den Benutzer zur Fehlersuche
github.com
B
