Raspberry Pi: log sensor data to SQLite and query it

CSV files are how sensor logging starts. SQLite is how it grows up: one file, zero admin, but real queries (last hour, daily average, the 5 times the basement got humid). It ships inside Python; nothing to install, no server to run.

This tutorial sets up the schema, the insert loop, and the few queries that cover 90% of sensor questions.

What you need

• Raspberry Pi with Raspberry Pi OS
• Python 3 (already installed)
• A sensor writing readings (the Flask API tutorial’s stand-in works)

Why SQLite over CSV: indexed time queries (instant “average per hour last week” that CSV makes you hand-roll), safe concurrent access, and a file you can still just copy to another machine (e.g. it is a real database that fits the same storage habits as a CSV).

The schema

CREATE TABLE IF NOT EXISTS readings (
    ts      INTEGER PRIMARY KEY,   -- unix epoch seconds
    sensor  TEXT NOT NULL,
    value   REAL NOT NULL
);
CREATE INDEX IF NOT EXISTS idx_sensor_time ON readings(sensor, ts);

One table, index on (sensor, time). Resist the urge to add columns per sensor type: the sensor name column is what keeps one table working forever (e.g. “temp_basement”, “temp_attic”, “humidity” all share it).

The logging loop

# ~/logger/logger.py
import sqlite3, time, random   # random = stand-in sensor

DB = "/home/brian/logger/readings.db"

def log(sensor, value, ts=None):
    ts = ts or int(time.time())
    with sqlite3.connect(DB) as conn:
        conn.execute(
            "INSERT INTO readings(sensor, value, ts) VALUES (?, ?, ?)",
            (sensor, value, ts))

def query_last_hour(sensor):
    cutoff = int(time.time()) - 3600
    with sqlite3.connect(DB) as conn:
        return conn.execute(
            "SELECT ts, value FROM readings WHERE sensor=? AND ts>=? ORDER BY ts",
            (sensor, cutoff)).fetchall()

# main loop
import os
if not os.path.exists(DB):
    with sqlite3.connect(DB) as conn:
        conn.executescript("""CREATE TABLE IF NOT EXISTS readings (
            ts INTEGER, sensor TEXT, value REAL);
            CREATE INDEX idx_sensor_time ON readings(sensor, ts);""")

while True:
    log("temp_basement", round(20 + random.random() * 3, 2))
    time.sleep(60)

Two details that carry this pattern: the with block commits (SQLite transactions open per connection), and INSERT ... VALUES (?, ?, ?) parameterization is the difference between a working logger and a quote-bug database.

The queries that answer real questions

-- average per hour for a chart
SELECT ts/3600*3600 AS hour, AVG(value), MIN(value), MAX(value)
FROM readings WHERE sensor='temp_basement'
GROUP BY hour ORDER BY hour;

-- when did it cross the threshold? (e.g. basement humidity events)
SELECT ts, value FROM readings
WHERE sensor='humidity_basement' AND value > 70
ORDER BY ts;

-- daily change (is the trend up?)
SELECT date(ts, 'unixepoch') AS day, AVG(value)
FROM readings WHERE sensor='temp_basement'
GROUP BY day ORDER BY day;

-- retention: keep 90 days
DELETE FROM readings WHERE ts < strftime('%s','now','-90 days');

Run that last one from a cron entry weekly (e.g. crontab -e, one line) and the database stays small forever.

Feeding it from the API

The Flask tutorial’s API and this logger want to be separate processes. The clean join: a small writer task in the API process

import threading, time
from app import read_temp   # same stand-in

def background_logger():
    while True:
        log("temp_basement", read_temp())
        time.sleep(60)

threading.Thread(target=background_logger, daemon=True).start()

Now the database and the API share the file safely (SQLite handles concurrent readers with one writer fine at household rates).

What you learned

• SQLite in Python: with sqlite3.connect() manages transactions.
• One table + one index covers multi-sensor logging forever.
• GROUP BY on time buckets turns raw rows into chart-ready answers.

When something breaks

• “database is locked”: two writers at once (e.g. logger and a migration script). Household rates: one writer is enough; if you really have two, add timeout=30 to connect().
• Queries slow after months: the (sensor, ts) index is missing, or you kept the raw table at 1-second rates for years. The retention DELETE fixes the growth; the index fixes the scan.
• Timestamps look wrong by hours: epoch is UTC. Convert at query time (datetime(ts, 'unixepoch', 'localtime')), never at insert.
• Pi’s SD card fills: 1 row/min is 500 KB/year, nothing. At 10 Hz you will fill a card in months; retention + aggregation is the fix (e.g. roll daily averages into a second table and delete the raw).

What to build next

• The Flask API tutorial exposes these queries over HTTP.
• The InfluxDB + Grafana tutorial is the upgrade when you want dashboards instead of SQL.
• The ESP32 MQTT tutorial is the sensor side that feeds this database from other rooms.