The single-module monitor read one circuit. A whole home has more — mains plus branches, or three phases. On a Raspberry Pi the answer is the same as on a microcontroller: a bus. Two or three rbAmp modules share the Pi's two I²C wires, and one broadcast tells them all to snapshot at the same instant — so the per-circuit numbers add up to an honest total.
What you'll build
- Per-circuit voltage / current / real power from each module, read by the Pi.
- A coherent total active power — the sum of the modules, all latched on the same instant, not three readings taken at three different moments (why that matters).
- Everything on one I²C bus, one Pi, in plain Python.
Bill of materials
| Item | Qty | Notes |
|---|---|---|
| rbAmp Basic Wattmeter (UI1) | 3 | one per circuit · product |
| SCT-013 CT | 3 | one per module, rating per circuit · product |
| Raspberry Pi | 1 | any model with the 40-pin I²C header |
| Jumper wires | — | the shared I²C bus |
Wire the shared bus
All modules sit on the same Pi I²C pins — they differ only by address:
Raspberry Pi module A (0x50) module B (0x51) module C (0x52)
------------ --------------- --------------- ---------------
SDA (GPIO2) ────┬────── SDA ────┬────── SDA ────┬────── SDA
SCL (GPIO3) ────┼────── SCL ────┼────── SCL ────┼────── SCL
5V ────┼────── VCC ├────── VCC ├────── VCC
GND ────┴────── GND ────┴────── GND ────┴────── GND
│ │ │
CT on circuit A CT on B CT on C- Power each module at 5 V (the Pi's 5 V pins); the bus is 3.3 V logic (why?). Each module has on-board pull-ups — on a multi-module bus keep them on the first module and disable them on the rest (one pull-up pair per bus).
- CTs clamp around their conductors — non-invasive on the current side. Mains work is dangerous; if you're not comfortable in a live panel, get a qualified person.
Give each module an address
Modules ship at 0x50, so assign the second and third a unique address (0x51, 0x52) before they share a bus.
On current firmware this is a plain production operation — no special mode or tooling: stage, arm, commit; the
module reboots at its new address. Do it one module at a time, with the others disconnected, so there's no
clash — the library's prepare_address_change() / commit_address_change() handle it.
Synchronize, then read
Instead of reading three modules one after another (each caught at a slightly different moment), you send one broadcast "latch now" to the whole bus — every module freezes its window on the same instant. Then you read each module's frozen snapshot in turn. The reads are sequential, but the values all belong to the same moment, so summing them is honest. It's the three-phase coherence principle, here in Python.
The script
Set FLEET_ADDRS and MY_CT for your modules and run python tp2_multimodule_pi_monitor.py:
"""
T-P2 — Multi-module Raspberry Pi bus (per-circuit + coherent whole-panel total).
Three rbAmp UI1 modules share the Pi's one I2C bus, each measuring its own
circuit's mains. The v1.3+ General-Call broadcast latches all three period
accumulators on the same instant so the summed Wh total is honest -- every
module integrates the same wall-clock window, no per-module drift.
Wiring (Raspberry Pi, hardware I2C, three modules in parallel on the same bus):
Every rbAmp: SDA -> GPIO2 (Pi pin 3)
SCL -> GPIO3 (Pi pin 5)
GND -> any GND pin
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).
Optional external 4.7 kOhm pull-ups on SDA and SCL to 3.3V. The Pi has
on-board 1.8 kOhm pull-ups which are usually enough for two or three
modules on a short bus; add external pull-ups if you see NACKs on longer
runs or with more devices.
Unique addresses: modules ship at 0x50. To make three of them coexist on
one bus, provision the extras once with the address-change flow
(`prepare_address_change` -> `commit_address_change`). That is a separate
one-time step; this tutorial assumes the three addresses are already
0x50 / 0x51 / 0x52.
Enable I2C once on the Pi (same as T-P1):
sudo raspi-config # Interface Options -> I2C -> Enable
The stock 100 kHz Broadcom I2C kernel driver is clean; no
`dtparam=i2c_arm_baudrate` override needed.
Install:
pip install rbamp smbus2
Run:
python tp2_multimodule_pi_monitor.py
# or with a custom fleet:
# python tp2_multimodule_pi_monitor.py --addr 0x50 --addr 0x51 --addr 0x52
Accuracy: +/-0.5% of reading per channel with matched rbAmp CT.
Summed total across three matched channels: approximately +/-1%.
Ctrl-C to stop.
"""
import argparse
import time
from smbus2 import SMBus
from rbamp import (
RbAmp,
RbAmpFleet,
RbAmpError,
RbAmpStaleError,
RbAmpSensorClass,
)
# ---- configuration --------------------------------------------------------
FLEET_ADDRS = [0x50, 0x51, 0x52] # provisioned addresses of the 3 modules
MY_CT = 3 # SCT-013-030 for every module (edit per-mod if needed)
LIVE_S = 1.0 # refresh per-module live V/I/P each second
ENERGY_S = 30.0 # broadcast a GC latch + roll total Wh every 30 s
GC_GROUP = 0 # 0 = all-call (every GC-enabled module latches)
def ensure_ct_preset(dev, ct_code):
"""Verify-then-set: rewrite CT preset only if it actually differs. See T-P1."""
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 enrol_fleet(bus, addrs, ct_code):
"""Bring up N modules at explicit addresses (no bus scan). Any module that
fails begin() is reported and skipped -- the fleet keeps the rest."""
fleet = RbAmpFleet(bus)
for a in addrs:
try:
dev = RbAmp(bus, addr=a)
dev.begin()
except RbAmpError as e:
print(f"[0x{a:02X}] enrol FAILED: {type(e).__name__} -- {e}")
continue
rewrote = ensure_ct_preset(dev, ct_code)
note = "CT rewritten" if rewrote else "CT already set"
print(f"[0x{a:02X}] enrolled fw=0x{dev.firmware_version:02X} "
f"topology={dev.topology_name} ch={dev.channels} ({note})")
fleet.add(dev)
return fleet
def print_live(fleet):
"""One live row: per-module V / I / P plus a fleet total P line."""
total_p = 0.0
parts = []
for m in fleet:
try:
v = m.voltage
i = m.current[0]
p = m.power[0]
except RbAmpError:
parts.append(f"[0x{m.address:02X}] --- V --- A ---- W")
continue
total_p += p
parts.append(f"[0x{m.address:02X}] {v:6.1f}V {i:5.2f}A {p:6.0f}W")
total_wh = fleet.total_energy_wh()
print(" ".join(parts) + f" | totP={total_p:7.0f}W totE={total_wh:9.4f}Wh")
def roll_gc_energy(fleet):
"""One GC-synchronised period window across the whole fleet.
1. gc_latch() broadcasts the 5-byte GC frame `A5 27 group tick_lo tick_hi`
at I2C general-call address 0x00. Every GC-enabled module with a
matching group latches its period accumulator on the same clock edge.
2. Sleep the settle window (inside gc_latch).
3. check_sync() reads REG_GC_TICK per module; any stale/NACK -> missed.
4. Per in-sync module, read_period_snapshot(settle_ms=0, skip_latch=True)
reads the already-latched values -- no per-module latch, no settle.
"""
tick = fleet.gc_latch(group=GC_GROUP) # auto-increments internal counter
sync = fleet.check_sync(expected_tick=tick)
missed = [s for s in sync if not s.in_sync]
if missed:
for s in missed:
print(f" * [0x{s.addr:02X}] MISSED gc tick "
f"(got=0x{s.gc_tick:04X} expected=0x{tick:04X} reachable={s.reachable})")
for m, s in zip(fleet, sync):
if not s.in_sync:
continue
try:
snap = m.read_period_snapshot(settle_ms=0, skip_latch=True)
except RbAmpStaleError:
print(f" * [0x{m.address:02X}] period STALE (in_sync but slave says stale)")
continue
except RbAmpError as e:
print(f" * [0x{m.address:02X}] period skipped: {type(e).__name__} -- {e}")
continue
print(f" * [0x{m.address:02X}] avg_P={snap.avg_p[0]:6.1f}W "
f"dt={snap.master_dt_ms:5d}ms E={m.energy.wh(0):8.4f}Wh")
print(f" * tick=0x{tick:04X} in_sync={sum(1 for s in sync if s.in_sync)}/{len(sync)} "
f"fleet totE={fleet.total_energy_wh():.4f}Wh")
def run(bus, addrs=None, ct_code=MY_CT,
live_s=LIVE_S, energy_s=ENERGY_S, max_iters=None):
if addrs is None:
addrs = FLEET_ADDRS
fleet = enrol_fleet(bus, addrs, ct_code)
if len(fleet) == 0:
print("no modules enrolled -- check wiring and addresses; exiting")
return
enabled = fleet.enable_gc_all(group=GC_GROUP)
print(f"GC latch enabled on {enabled}/{len(fleet)} module(s), group={GC_GROUP}")
print("Accuracy: +/-0.5%/ch matched CT; summed total ~ +/-1%. Ctrl-C to stop.\n")
next_live_t = time.monotonic()
next_energy_t = next_live_t + energy_s
iters = 0
while True:
now = time.monotonic()
if now >= next_live_t:
next_live_t = now + live_s
print_live(fleet)
if now >= next_energy_t:
next_energy_t = now + energy_s
roll_gc_energy(fleet)
iters += 1
if max_iters is not None and iters >= max_iters:
return fleet
sleep_for = min(next_live_t, next_energy_t) - time.monotonic()
if sleep_for > 0:
time.sleep(sleep_for)
def main():
ap = argparse.ArgumentParser(
description="Multi-module Raspberry Pi rbAmp bus with fleet GC-latch energy sync."
)
ap.add_argument("--bus", type=int, default=1,
help="I2C bus number (Raspberry Pi default: 1)")
ap.add_argument("--addr", type=lambda s: int(s, 0), action="append",
default=None,
help="rbAmp address (repeatable; default: 0x50 0x51 0x52)")
ap.add_argument("--ct", type=int, default=MY_CT,
help="CT model code (1=SCT-013-005, 2=-010, 3=-030, "
"4=-050, 6=-020)")
args = ap.parse_args()
addrs = args.addr if args.addr else FLEET_ADDRS
with SMBus(args.bus) as bus:
run(bus, addrs=addrs, ct_code=args.ct)
if __name__ == "__main__":
try:
main()
except KeyboardInterrupt:
print("\nstopped.")(Run-verified against the rbAmp Python library's mock backend on CPython 3.12 — rbamp 1.3.0 + smbus2 0.6.1.
Uses the v1.3+ RbAmpFleet broadcast latch, not per-module sequential latching. check_sync() reports any
module that missed the broadcast so a mis-wired or un-provisioned module is obvious.)
What you'll see
Per-second live rows with a running total, and a synchronized energy line every 30 s:
[0x50] enrolled fw=0x04 topology=SINGLE ch=1 (CT already set)
[0x51] enrolled fw=0x04 topology=SINGLE ch=1 (CT already set)
[0x52] enrolled fw=0x04 topology=SINGLE ch=1 (CT already set)
GC latch enabled on 3/3 module(s), group=0
[0x50] 230.5V 3.52A 812W [0x51] 230.5V 1.93A 445W [0x52] 230.5V 5.37A 1237W | totP= 2494W totE= 0.0000Wh
* [0x50] avg_P= 795.8W dt= 3253ms E= 0.7191Wh
* [0x51] avg_P= 436.1W dt= 3203ms E= 0.3880Wh
* [0x52] avg_P=1212.3W dt= 3152ms E= 1.0614Wh
* tick=0x0001 in_sync=3/3 fleet totE=2.1685WhWhat the script is doing
RbAmpFleet does the fleet bookkeeping; three functions run the show — enrol_fleet() at startup, then
print_live() and roll_gc_energy() on their own cadences.
enrol_fleet() — bring up the modules
fleet = RbAmpFleet(bus)
for a in [0x50, 0x51, 0x52]:
dev = RbAmp(bus, addr=a); dev.begin()
ensure_ct_preset(dev, MY_CT)
fleet.add(dev)Brings up each module at its explicit address (no bus scan, so you see the addresses in the tutorial),
verify-then-sets its CT preset (no needless flash write), and adds it to the fleet. A module that fails begin()
is reported and skipped — the fleet keeps the rest, so one bad connection doesn't sink the whole run.
print_live() — per-module + running total
for m in fleet:
v = m.voltage; i = m.current[0]; p = m.power[0]
total_p += pWalks the fleet and reads each module's live V / I / P for display, summing the per-module power into a live whole-panel total. These live reads don't need to be synchronized — they're for the readout.
roll_gc_energy() — the coherent part
tick = fleet.gc_latch(group=0) # ONE broadcast latches every module together
sync = fleet.check_sync(expected_tick=tick) # did each accept the tick?
for m, s in zip(fleet, sync):
if not s.in_sync: # dropped / GC-disabled / wrong group
continue
snap = m.read_period_snapshot(settle_ms=0, skip_latch=True) # read the frozen windowThis is why the total is trustworthy. gc_latch() broadcasts one General-Call frame to address 0x00 — every
enabled module freezes its period window on the same instant. check_sync() then reads each module's
REG_GC_TICK and flags any that missed the broadcast (prints a MISSED gc tick line and skips that module for
the window — the healthy ones keep rolling). Then read_period_snapshot(skip_latch=True) reads each
already-frozen window without re-latching, and fleet.total_energy_wh() sums the fleet. Because every module
latched together, the summed Wh is coherent — not three readings taken at three different moments. That
check_sync health line is exactly what you'd instrument in a real install.
Make it your own
- More circuits — add modules on the bus, assign each an address; they join the fleet. Mind the bus-loading / pull-up rules as the count grows.
- Log the panel — one row per module per interval to CSV or SQLite (the logger build).
- Serve it — expose the per-circuit + total as a Flask endpoint or publish over MQTT — locally, no cloud.
- Per-circuit actions — trip a relay per module over its own limit (the relay build).
- Independent fleets — different
group_ids let unrelated groups latch on their own cadence, all on one bus.
Where to go next
- Act on it — trip a GPIO relay when a circuit crosses a current limit.
- Log it — periodic per-circuit energy to CSV or SQLite.
- Prefer Home Assistant? The same multi-module bus is the whole panel on one ESP32 with ESPHome.