The monitors read it; the relay acts on it — this one keeps it. An rbAmp module and a Raspberry Pi make a standalone logger: timestamped voltage, current, power and kilowatt-hours appended to a CSV file, and — because a Pi has a database built in — optionally straight into SQLite so you can query it. No SD-card breakout, no real-time clock, no cloud: the Pi's own filesystem and NTP-synced clock do the work. It's the fourth and final build in the Python series (after the monitor, the multi-module bus, and the relay).
What you'll build
- A CSV row per interval —
timestamp, voltage, current, power, energy— appended to a file on the Pi. - Optionally the same rows in SQLite (
--sqlite energy.db), so you canSELECTand chart later. - Real timestamps from the Pi's system clock (NTP-synced) — no RTC hardware needed.
- Robust to a bad read, and flushed so a row survives a power cut.
Bill of materials
| Item | Qty | Notes |
|---|---|---|
| rbAmp Basic Wattmeter (UI1) | 1 | voltage + current + on-chip real power · product |
| SCT-013 CT | 1 | rating to match your circuit · product |
| Raspberry Pi | 1 | any model with the 40-pin I²C header |
| Jumper wires | — | the I²C bus (SDA / SCL / 5V / GND) |
(No SD-card module, no RTC — the Pi already has both.)
Wire it up and install
The rbAmp goes on the Pi's I²C pins (SDA GPIO2 / SCL GPIO3, VCC 5 V — why 5 V but a 3.3 V
bus?), CT clamped around one conductor. Enable I²C (sudo raspi-config) and install — CSV and
SQLite are both in Python's standard library, so no extra dependencies:
pip install rbamp smbus2The CSV
Each interval appends one row; the header is written when the file is first created:
timestamp,voltage_V,current_A,power_W,energy_Wh
2026-07-30T14:05:00+01:00,230.70,3.793,875.00,0.0120
2026-07-30T14:05:30+01:00,230.65,3.788,873.40,1.4700(The timestamp carries the Pi's local timezone offset, so the file stays unambiguous a year later on another
host. If your Pi is offline and not NTP-synced, timestamps read 1970-… until you set the clock or connect it.)
Open it in a spreadsheet, or — if you passed --sqlite — query the database: SELECT strftime('%H', timestamp),
AVG(power_W) FROM energy GROUP BY 1;.
The script
Run it as-is for CSV, or add --sqlite energy.db to also write a queryable database:
"""
T-P4 -- Log energy to CSV (and optionally SQLite) on a Raspberry Pi.
One rbAmp module meters a circuit; every interval this script writes a
timestamped row to a CSV file, mirrors it to stdout so you can watch, and
-- if you pass `--sqlite path.db` -- also inserts the same row into an
SQLite table for easy querying (`SELECT AVG(power_W) FROM energy WHERE ...`).
Compared to the Arduino data-logger equivalent this is much simpler: the
Pi has a real filesystem and an NTP-synchronised system clock, so there
is no microSD-card module and no external RTC. Timestamps are just
`datetime.now().astimezone().isoformat(...)` -- the Pi is already telling
you the correct wall-clock time.
CSV columns:
timestamp,voltage_V,current_A,power_W,energy_Wh
Wiring (Raspberry Pi, hardware I2C):
rbAmp SDA -> GPIO2 (Pi pin 3)
rbAmp SCL -> GPIO3 (Pi pin 5)
rbAmp GND -> any GND pin
rbAmp VCC -> 5V (Pi pin 2 or 4)
^ rbAmp needs 5V for its analog front-end. Only the I2C
logic level is 3.3V (handled internally).
Enable I2C once on the Pi (same as T-P1/P2/P3):
sudo raspi-config # Interface Options -> I2C -> Enable
Install:
pip install rbamp smbus2
# `csv` and `sqlite3` are Python stdlib; no extra pip deps.
Run:
python tp4_pi_csv_sqlite_logger.py # CSV only, every 30 s
python tp4_pi_csv_sqlite_logger.py --sqlite energy.db # CSV + SQLite
python tp4_pi_csv_sqlite_logger.py --interval 10 --ct 3
Accuracy: +/-0.5% of reading per channel with matched rbAmp CT.
Ctrl-C to stop -- the CSV file + SQLite DB are flushed and closed cleanly.
Data is safe against power cuts: every CSV row is flushed and every SQLite
insert is committed before the loop moves on.
"""
import argparse
import csv
import os
import sqlite3
import sys
import time
from datetime import datetime
from smbus2 import SMBus
from rbamp import (
RbAmp,
RbAmpError,
RbAmpStaleError,
RbAmpSensorClass,
)
# ---- configuration --------------------------------------------------------
RBAMP_ADDR = 0x50
METER_CH = 0
MY_CT = 3 # SCT-013-030
CSV_PATH = "energy.csv"
LOG_S = 30.0 # one CSV row every 30 s
CSV_HEADER = ["timestamp", "voltage_V", "current_A", "power_W", "energy_Wh"]
SQLITE_DDL = """
CREATE TABLE IF NOT EXISTS energy (
timestamp TEXT NOT NULL,
voltage_V REAL,
current_A REAL,
power_W REAL,
energy_Wh REAL
)
"""
SQLITE_INSERT = "INSERT INTO energy (timestamp, voltage_V, current_A, power_W, energy_Wh) VALUES (?, ?, ?, ?, ?)"
def ensure_ct_preset(dev, ct_code):
"""Verify-then-set CT preset (same pattern as T-P1/P2/P3)."""
try:
applied = dev.read_ct_model_ch(0)
except RbAmpError:
applied = -1
if applied == ct_code:
return False
dev.set_sensor_class(RbAmpSensorClass.SCT_013)
dev.set_ct_model_ch(0, ct_code)
return True
def sample_row(dev):
"""Read one V/I/P + latch a period + roll Wh. Returns a row tuple or None
on any RbAmpError (transient hiccup / stale snapshot). The energy tick
silently skips on stale but V/I/P are still reported."""
try:
v = dev.voltage
i = dev.current[METER_CH]
p = dev.power[METER_CH]
except RbAmpError:
return None
try:
dev.read_period_snapshot() # library auto-rolls the Wh accumulator
except RbAmpStaleError:
pass # keep the V/I/P row; skip the energy tick
except RbAmpError:
pass
wh = dev.energy.wh(METER_CH)
# ISO-8601 with local tz offset; Pi is NTP-synced.
ts = datetime.now().astimezone().isoformat(timespec="seconds")
return (ts, v, i, p, wh)
def open_csv(path):
"""Open (or create) the CSV, write the header only if new, return
(file, writer). Line-buffered so each row hits disk immediately."""
is_new = not os.path.exists(path) or os.path.getsize(path) == 0
f = open(path, "a", newline="", encoding="utf-8", buffering=1)
w = csv.writer(f)
if is_new:
w.writerow(CSV_HEADER)
f.flush()
return f, w
def open_sqlite(path):
"""Open SQLite (create tables if missing), return connection. Explicit
commit per insert gives crash-safe single-row durability."""
con = sqlite3.connect(path)
con.execute(SQLITE_DDL)
con.commit()
return con
def run(bus, addr=RBAMP_ADDR, ct_code=MY_CT,
csv_path=CSV_PATH, sqlite_path=None, log_s=LOG_S,
max_iters=None, stdout=sys.stdout):
"""Core loop. `sqlite_path=None` = CSV only."""
with RbAmp(bus, addr=addr) as dev:
ensure_ct_preset(dev, ct_code)
print(f"rbAmp @ 0x{dev.address:02X} fw=0x{dev.firmware_version:02X} "
f"topology={dev.topology_name} ch={dev.channels}", file=stdout)
print(f"logging every {log_s:.1f} s -> {csv_path}"
+ (f" + sqlite:{sqlite_path}" if sqlite_path else ""),
file=stdout)
print("Accuracy: +/-0.5% of reading with matched CT. Ctrl-C stops.\n",
file=stdout)
csv_file, csv_writer = open_csv(csv_path)
con = open_sqlite(sqlite_path) if sqlite_path else None
rows = 0
skipped = 0
iters = 0
try:
while True:
t_start = time.monotonic()
row = sample_row(dev)
if row is None:
skipped += 1
print(f"[SKIP #{skipped}] read failed -- holding, no row written",
file=stdout)
else:
ts, v, i, p, wh = row
csv_writer.writerow([
ts,
f"{v:.2f}", f"{i:.3f}", f"{p:.2f}", f"{wh:.4f}",
])
csv_file.flush() # crash-safe row
if con is not None:
con.execute(SQLITE_INSERT, (ts, v, i, p, wh))
con.commit() # crash-safe row
rows += 1
print(f"row#{rows:>4d} {ts} U={v:6.1f}V I={i:6.3f}A "
f"P={p:7.1f}W E={wh:9.4f}Wh", file=stdout)
iters += 1
if max_iters is not None and iters >= max_iters:
return {"rows": rows, "skipped": skipped,
"csv": csv_path, "sqlite": sqlite_path}
sleep_for = log_s - (time.monotonic() - t_start)
if sleep_for > 0:
time.sleep(sleep_for)
finally:
csv_file.close()
if con is not None:
con.close()
def main():
ap = argparse.ArgumentParser(
description="Log rbAmp energy to CSV (and optionally SQLite) on a Raspberry Pi."
)
ap.add_argument("--bus", type=int, default=1)
ap.add_argument("--addr", type=lambda s: int(s, 0), default=RBAMP_ADDR)
ap.add_argument("--ct", type=int, default=MY_CT)
ap.add_argument("--csv", type=str, default=CSV_PATH)
ap.add_argument("--sqlite", type=str, default=None,
help="Optional path to an SQLite DB file (adds `energy` table).")
ap.add_argument("--interval", type=float, default=LOG_S,
help="Seconds between rows (default 30).")
args = ap.parse_args()
with SMBus(args.bus) as bus:
run(bus, addr=args.addr, ct_code=args.ct,
csv_path=args.csv, sqlite_path=args.sqlite, log_s=args.interval)
if __name__ == "__main__":
try:
main()
except KeyboardInterrupt:
print("\nstopped.")(Run-verified against the rbAmp library's mock backend + a real filesystem CSV and a real SQLite DB on CPython
3.12 — three scenarios: CSV+SQLite write, append-idempotence on re-run, and graceful SKIP on injected read
failures. Every row is flushed (CSV) and committed (SQLite) before the loop moves on, so a power cut never loses a
written row.)
What the script is doing
Three helpers — sample_row(), open_csv(), and (optionally) open_sqlite() — feed a simple run() loop.
sample_row() — one reading
v = dev.voltage; i = dev.current[0]; p = dev.power[0] # RbAmpError here -> return None (skip the row)
dev.read_period_snapshot() # roll the Wh accumulator (stale -> keep V/I/P, skip tick)
ts = datetime.now().astimezone().isoformat(timespec="seconds")
return (ts, v, i, p, dev.energy.wh(0))Reads the live V/I/P, latches a period to advance the watt-hour total, and stamps it with the Pi's system
time — ISO-8601 with the local timezone offset, no RTC and no manual clock setup because the Pi is NTP-synced. A
transient read error returns None (the loop skips that row); a stale energy window keeps the V/I/P row but
doesn't advance Wh that interval.
open_csv() / open_sqlite() — the sinks
f = open(path, "a", newline="", encoding="utf-8", buffering=1) # line-buffered -> each row hits disk
if is_new: writer.writerow(CSV_HEADER) # header written once
con.execute("CREATE TABLE IF NOT EXISTS energy (...)") # SQLite, only if --sqlite givenCSV opens in append mode (header only when the file is new), line-buffered so each row is on disk immediately.
Pass --sqlite and it also opens an SQLite DB and creates the energy table if missing — both from Python's
standard library, so there's nothing extra to install.
The run() loop
row = sample_row(dev)
if row is None: skipped += 1 # bad read -> log a SKIP, no row
else:
csv_writer.writerow([...]); csv_file.flush() # crash-safe CSV row
if con: con.execute(INSERT, row); con.commit() # crash-safe SQLite rowEvery --interval seconds it takes a sample and — if the read was good — writes one CSV row (and one SQLite row
if enabled), flushing and committing before it sleeps, so a power cut never loses a written row. A finally
block closes the file and DB cleanly on Ctrl-C. That per-row flush/commit is the whole durability story.
Using the data
Pop open the CSV in any spreadsheet — plot power over the day, sum energy per hour, spot the fridge cycling. Or, with the SQLite option, run SQL straight against it — hourly averages, daily totals, the lot. Because the energy column is real host-integrated watt-hours, the totals reconcile with what you'd expect on the bill. Want it graphed live instead? Put it in Home Assistant — but for a fully offline record, these files are yours.
The Python series, recap
You now have the full standalone Pi toolkit — no network required: - Read it — one module in the terminal. - Scale it — several circuits on one bus, coherently. - Act on it — trip a GPIO relay on over-current. - Log it — this build, to CSV / SQLite.