Recently I was curious about the amount of CO2 present while sleeping and working; I ended up buying an Aranet4 monitor. It's great to see the reading live, but I would also like to compare trends over time. The Aranet4 itself can only store a maximum of 7 days of data, so for longer term analysis I need to periodically export the data.
This is where a Raspberry Pi comes in handy! First, install the necessary packages:
sudo apt install bluetooth pi-bluetooth bluez blueman
Use bluetoothctl
to pair the Aranet4:
sudo bluetoothctl
> scan on
> pair $MAC
> scan off
We can use the excellent Aranet4-Python library to connect to the Aranet4 through Bluetooth and save measurement data.
pip3 install aranet4
, then you can use a script similar to the following to periodically save data to sqlite. The retry logic is necessary to deal with flaky connections when the Raspberry Pi and the Aranet are in different rooms.
from datetime import datetime
import os
import sqlite3
import time
import aranet4
NUM_RETRIES = 10
DEVICES = {
'bedroom': 'xx:xx:xx:xx:xx:xx'
}
db_path = os.path.join(os.path.expanduser('~'), 'aranet4.db')
con = sqlite3.connect(db_path)
cur = con.cursor()
cur.execute('''CREATE TABLE IF NOT EXISTS measurements(
device TEXT,
timestamp INTEGER,
temperature REAL,
humidity INTEGER,
pressure REAL,
CO2 INTEGER,
PRIMARY KEY(device, timestamp)
)''')
con.commit()
for name, mac in DEVICES.items():
entry_filter = {}
res = cur.execute('''SELECT timestamp FROM measurements WHERE device = ?
ORDER BY timestamp DESC LIMIT 1''', (name,))
row = res.fetchone()
if row is not None:
entry_filter['start'] = datetime.utcfromtimestamp(row[0])
for attempt in range(NUM_RETRIES):
entry_filter['end'] = datetime.now()
try:
history = aranet4.client.get_all_records(mac, entry_filter)
break
except Exception as e:
print('attempt', attempt, 'failed, retrying:', e)
data = []
for entry in history.value:
if entry.co2 < 0:
continue
data.append((
name,
time.mktime(entry.date.timetuple()),
entry.temperature,
entry.humidity,
entry.pressure,
entry.co2
))
print('fetched', len(data), 'measurements', entry_filter)
cur.executemany(
'INSERT OR IGNORE INTO measurements VALUES(?, ?, ?, ?, ?, ?)', data)
con.commit()
con.close()
Tags: automation, programming