Skip to content

Instantly share code, notes, and snippets.

@idiotandrobot

idiotandrobot/save-sensor-data.py

Last active April 12, 2024 08:33
Show Gist options

  • Select an option

    Learn more about clone URLs
  • Save idiotandrobot/2b2e36d69fbf7293851a6b1c3d59d6e2 to your computer and use it in GitHub Desktop.

Select an option

Learn more about clone URLs
Save idiotandrobot/2b2e36d69fbf7293851a6b1c3d59d6e2 to your computer and use it in GitHub Desktop.
Download ZIP

Revisions

  1. idiotandrobot revised this gist Apr 12, 2024. 1 changed file with 1 addition and 1 deletion.
    2 changes: 1 addition & 1 deletion save-sensor-data.py
    View file
    Original file line number Diff line number Diff line change
    @@ -49,7 +49,7 @@

    data.append((
    name,
    time.mktime(entry.date.timetuple()),
    time.mktime(entry.date.utctimetuple()),
    entry.temperature,
    entry.humidity,
    entry.pressure,
  2. idiotandrobot created this gist Apr 12, 2024.
    63 changes: 63 additions & 0 deletions save-sensor-data.py
    View file
    Original file line number Diff line number Diff line change
    @@ -0,0 +1,63 @@
    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()