Posted on Sat 04 March 2023

Saving Aranet4 data to Raspberry Pi

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

© Julian Schrittwieser. Built using 開板. Theme by Giulio Fidente on github. .