The Pi monitor reads the current; this one acts on it. When a circuit stays over your limit, a Raspberry Pi switches a GPIO relay to shed the load — locally, in its own Python loop, no Home Assistant, no cloud.
⚠️ Read this first — what this is and isn't. This is a comfort-and-convenience layer, not a protection device. Your circuit breaker remains the safety device — never size a circuit or a load assuming this automation will act. And the relay you switch must be rated for the load: a small relay can't break 30–50 A directly. For a high-current circuit, use the GPIO relay to drive a properly rated contactor, and let an electrician size it. Mains work is dangerous; if you're not comfortable in a live panel, get a qualified person.
What you'll build
- A rbAmp module metering one circuit's current on the Pi.
- A GPIO relay that trips when the current stays over
LIMIT_ONA and releases belowLIMIT_OFF(hysteresis). - Inrush debounce so a motor or kettle's startup surge doesn't nuisance-trip.
- Optional LED/buzzer alert. All local, in Python.
Bill of materials
| Item | Qty | Notes |
|---|---|---|
| rbAmp current module (I-module) | 1 | current-only I1 is the lean pick — a UI1 works too · product |
| SCT-013 CT | 1 | rating to match the circuit · product |
| Raspberry Pi | 1 | any model with the 40-pin header |
| Relay module | 1 | isolated, rated for the load — or driving a contactor for high current |
| LED / buzzer (optional) | 1 | local alert |
Wire it up
The rbAmp module goes on the Pi's I²C pins (SDA GPIO2 / SCL GPIO3, VCC 5 V — why 5 V but a 3.3 V
bus?). The relay module's control input goes to GPIO26 (BCM); set RELAY_ACTIVE_HIGH in the
config to match your board (many opto-isolated boards are active-LOW). An optional LED/buzzer alert goes to
GPIO16. Wire the load through the relay's NO contact so a de-energised relay = load OFF. The CT clamps
around the circuit conductor — non-invasive on the current side.
The logic: threshold, hysteresis, debounce
Same three ideas as the microcontroller version:
- Hysteresis — trip at
LIMIT_ONbut only release below a lowerLIMIT_OFF, so the relay doesn't chatter at the setpoint. - Inrush debounce — require the current to stay over the limit for a few seconds before tripping, so a startup surge rides through.
- The library reads the module's current-RMS register each loop (the module computes RMS internally at about 5 Hz) — Python compares it to the limits and switches the relay.
The script
Set your LIMIT_ON / LIMIT_OFF and pins, then pip install rbamp smbus2 gpiozero and run it. Relay on
GPIO26, alert on GPIO16 by default:
python
"""
T-P3 -- Local current-limit -> Raspberry Pi GPIO relay (standalone, no HA).
One rbAmp module meters a circuit's current. When it stays over a limit for a
sustained interval, the Pi drives a GPIO relay to shed the load and lights an
alert LED (or drives a small buzzer). Everything runs in one Python loop --
no network, no Home Assistant round-trip, sub-second reaction once the
overload is confirmed.
=========================== SAFETY -- READ FIRST ===========================
* The relay / switch MUST be rated for the load. A small hobby relay or
smart-plug CANNOT break 30-50 A directly. Switch a high-current circuit
with a properly rated CONTACTOR that this low-current relay drives.
* This is a COMFORT / CONVENIENCE layer, NOT a protection device. The
circuit BREAKER remains the safety device. Never size a circuit or a
load assuming this automation will act. For a hard cut-off use a
hardware overcurrent device (breaker / fuse / thermal cut-out).
* The relay reaction is deliberately delayed by OVERLOAD_S to ride through
inrush -- it is not, and must not be treated as, instantaneous
overcurrent protection.
============================================================================
Module: a current-only rbAmp I-module (I1) at the default address 0x50 is the
lean pick -- only current is needed, no voltage sensing. A UI1 works
too; its voltage channel simply goes unused. Whichever you use,
`dev.current[0]` is the metered channel.
Wiring (Raspberry Pi, hardware I2C + a relay module):
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).
Relay IN -> RELAY_PIN (BCM GPIO26 by default; edit below)
Relay VCC -> 5V (opto-isolated relay boards usually want 5V on the
coil side; check yours)
Relay GND -> Pi GND (common ground)
Wire the LOAD through the relay's NO (normally-open) contact so a
de-energised relay = load OFF (safe default at power-up before this
script even runs).
Alert -> ALERT_PIN (BCM GPIO16 by default) -> LED + series
resistor to GND, and/or an active buzzer.
Relay polarity: set RELAY_ACTIVE_HIGH below to match your module. Many
opto-isolated relay boards are ACTIVE-LOW (IN = LOW energises the coil).
gpiozero's `active_high=` argument handles both; we always call
`relay.on()` to mean "energise" (== load powered via the NO contact).
Enable I2C once on the Pi (same as T-P1/T-P2):
sudo raspi-config # Interface Options -> I2C -> Enable
Install:
pip install rbamp smbus2 gpiozero
Run:
python tp3_pi_gpio_load_relay.py
Set MY_CT / LIMIT_ON / LIMIT_OFF / OVERLOAD_S / COOLDOWN_S below to your
circuit's setpoints. Accuracy: +/-0.5% of reading per channel with the
matched rbAmp CT (current is the metered quantity here).
Ctrl-C to stop -- both the relay and the alert are released cleanly.
"""
import argparse
import time
from smbus2 import SMBus
from gpiozero import OutputDevice
from rbamp import (
RbAmp,
RbAmpError,
RbAmpSensorClass,
)
# ---- configuration --------------------------------------------------------
RBAMP_ADDR = 0x50
METER_CH = 0
MY_CT = 3 # SCT-013-030; 1=-005, 2=-010, 4=-050, 6=-020
# Threshold with hysteresis (amps). Trip when the current stays at/above
# LIMIT_ON; a pending overload only clears when it falls below LIMIT_OFF,
# so the relay does not chatter around the setpoint. Keep LIMIT_OFF < LIMIT_ON.
LIMIT_ON = 10.0
LIMIT_OFF = 8.0
SAMPLE_S = 0.2 # current sample cadence
OVERLOAD_S = 3.0 # sustained over-limit before trip (rides inrush)
COOLDOWN_S = 10.0 # stay shed before an auto re-arm attempt
# BCM GPIO pin numbers (edit to match your wiring).
RELAY_PIN = 26
ALERT_PIN = 16
RELAY_ACTIVE_HIGH = True # set False for active-LOW opto-isolated boards
# ---- state machine --------------------------------------------------------
class LoadState:
ARMED = "ARMED"
OVER_PENDING = "OVER_PENDING"
TRIPPED = "TRIPPED"
def ensure_ct_preset(dev, ct_code):
"""Verify-then-set CT preset, same pattern as T-P1/T-P2 -- no flash write
on repeat runs."""
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 _log_transition(t_ms, old, new, amps):
print(f"[{t_ms:8d}ms] {old:>12s} -> {new:<12s} (I={amps:6.3f} A)")
def run(bus, relay, alert, addr=RBAMP_ADDR, ct_code=MY_CT,
limit_on=LIMIT_ON, limit_off=LIMIT_OFF,
sample_s=SAMPLE_S, overload_s=OVERLOAD_S, cooldown_s=COOLDOWN_S,
max_iters=None):
"""Core loop -- takes actuators as args so a test harness can inject
gpiozero MockFactory-backed OutputDevices."""
# SAFETY: outputs already in the safe armed state (relay energised
# via `initial_value=True`, alert off) BEFORE we touch the bus.
if not relay.value:
relay.on()
alert.off()
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}")
print(f"armed: trip >= {limit_on:.1f} A sustained {overload_s:.1f} s, "
f"clear < {limit_off:.1f} A, cooldown {cooldown_s:.1f} s")
print(f"relay pin={RELAY_PIN} (active_{'high' if RELAY_ACTIVE_HIGH else 'low'}), "
f"alert pin={ALERT_PIN}")
print("Ctrl-C stops (releases relay + alert). Accuracy: +/-0.5% of reading.\n")
t0 = time.monotonic()
def now_ms():
return int((time.monotonic() - t0) * 1000.0)
state = LoadState.ARMED
over_since_s = 0.0
tripped_at_s = 0.0
trip_count = 0
bad_reads = 0
iters = 0
try:
while True:
loop_start = time.monotonic()
try:
a = dev.current[METER_CH]
except RbAmpError as e:
# Graceful: transient read must NOT trigger a false trip
# or a false release. Hold state, skip this sample.
bad_reads += 1
if bad_reads % 10 == 1:
print(f"[{now_ms():8d}ms] read failed "
f"({type(e).__name__}: {e}) -- holding state={state}")
_wait(loop_start, sample_s)
iters += 1
if max_iters is not None and iters >= max_iters:
return {"state": state, "trip_count": trip_count,
"bad_reads": bad_reads}
continue
now = time.monotonic()
if state == LoadState.ARMED:
if a >= limit_on:
over_since_s = now
_log_transition(now_ms(), state, LoadState.OVER_PENDING, a)
state = LoadState.OVER_PENDING
elif state == LoadState.OVER_PENDING:
if a < limit_off:
_log_transition(now_ms(), state, LoadState.ARMED, a)
state = LoadState.ARMED
elif (now - over_since_s) >= overload_s:
relay.off() # de-energise -> load OFF
alert.on()
tripped_at_s = now
trip_count += 1
_log_transition(now_ms(), state, LoadState.TRIPPED, a)
state = LoadState.TRIPPED
print(f" LOAD SHED (trip #{trip_count})")
elif state == LoadState.TRIPPED:
if (now - tripped_at_s) >= cooldown_s:
relay.on() # re-energise -> load ON
alert.off()
_log_transition(now_ms(), state, LoadState.ARMED, a)
state = LoadState.ARMED
print(f" cooldown elapsed -- load re-energised")
_wait(loop_start, sample_s)
iters += 1
if max_iters is not None and iters >= max_iters:
return {"state": state, "trip_count": trip_count,
"bad_reads": bad_reads}
finally:
# SAFETY: leave the outputs in a defined state -- release the alert,
# but LEAVE THE RELAY whichever way the state machine last set it.
# For "always re-energise on exit", change this to `relay.on()`.
alert.off()
def _wait(loop_start, sample_s):
"""Sleep off the remainder of this sample window (no drift)."""
elapsed = time.monotonic() - loop_start
if elapsed < sample_s:
time.sleep(sample_s - elapsed)
def main():
ap = argparse.ArgumentParser(
description="Local current-limit -> Pi GPIO relay (standalone, no HA)."
)
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("--relay-pin", type=int, default=RELAY_PIN)
ap.add_argument("--alert-pin", type=int, default=ALERT_PIN)
args = ap.parse_args()
# SAFETY: construct the outputs BEFORE we touch the I2C bus, and drive
# them to the safe armed state (load powered, alert off) IMMEDIATELY.
# `initial_value=True` + `active_high=RELAY_ACTIVE_HIGH` guarantees the
# physical pin lands on the "energised" level regardless of wiring.
relay = OutputDevice(args.relay_pin,
active_high=RELAY_ACTIVE_HIGH,
initial_value=True)
alert = OutputDevice(args.alert_pin,
active_high=True,
initial_value=False)
try:
with SMBus(args.bus) as bus:
run(bus, relay, alert, addr=args.addr, ct_code=args.ct)
except KeyboardInterrupt:
print("\nstopped.")
finally:
alert.off()
relay.close()
alert.close()
if __name__ == "__main__":
main()(Run-verified against the rbAmp library's mock + gpiozero's MockFactory on CPython 3.12 — every state edge,
plus a transient-NACK case proving a bad read never false-trips and never resets the sustained-overload timer. The
relay is driven to the safe load-ON level at construction, before the bus is touched.)
What you'll see
State transitions logged as they happen — inrush cleared by hysteresis, a sustained overload tripping, cooldown re-arming:
plaintext
rbAmp @ 0x50 fw=0x04 topology=SINGLE ch=1
armed: trip >= 10.0 A sustained 3.0 s, clear < 8.0 A, cooldown 10.0 s
relay pin=26 (active_high), alert pin=16
[ 200ms] ARMED -> OVER_PENDING (I=12.000 A) <- inrush spike
[ 401ms] OVER_PENDING -> ARMED (I= 3.000 A) <- inrush cleared (hysteresis)
[ 601ms] ARMED -> OVER_PENDING (I=11.000 A) <- real overload
[ 902ms] OVER_PENDING -> TRIPPED (I=11.000 A) <- sustained -> trip
LOAD SHED (trip #1)
[ 1403ms] TRIPPED -> ARMED (I=11.000 A) <- cooldown -> re-arm
cooldown elapsed -- load re-energisedWhat the script is doing
It's the same three-state machine as the Arduino relay build — ARMED → OVER_PENDING → TRIPPED — in
Python, driving a gpiozero relay.
Safe by default — before the bus is even touched
python
relay = OutputDevice(RELAY_PIN, active_high=RELAY_ACTIVE_HIGH, initial_value=True)
alert = OutputDevice(ALERT_PIN, active_high=True, initial_value=False)The relay is constructed first, with initial_value=True — so the physical pin lands on the "energised"
(load-ON, via the NO contact) level immediately, before any I²C traffic. active_high hides both relay-board
polarities behind one flag, so relay.on() always means "energise." A finally block releases the pins on exit
so a re-run can reclaim them.
The state machine — the loop
python
a = dev.current[METER_CH] # RbAmpError here -> hold state, skip (no false trip)
if state == ARMED and a >= LIMIT_ON: over_since = now; state = OVER_PENDING
elif state == OVER_PENDING:
if a < LIMIT_OFF: state = ARMED # inrush passed
elif now - over_since >= OVERLOAD_S: relay.off(); alert.on(); state = TRIPPED
elif state == TRIPPED and now - tripped_at >= COOLDOWN_S: relay.on(); alert.off(); state = ARMED- ARMED — load powered; the moment current hits
LIMIT_ONit timestamps and moves toOVER_PENDING. - OVER_PENDING — on a stopwatch. Drops below
LIMIT_OFF→ back to ARMED (no trip — hysteresis); stays over forOVERLOAD_S→ sheds (relay.off()), alert on, → TRIPPED. - TRIPPED — after
COOLDOWN_Sit re-arms and re-energises; if the overload persists it just trips again.
The hysteresis is the two thresholds; the inrush debounce is OVERLOAD_S. And the load-bearing bit: a
transient RbAmpError holds the state and does not reset the over-limit timer — a bus wobble can never cause
a false trip or a false release. (dev.lib-python's verification exercised exactly that path.)
Where to go next
- Log the trips — write each transition to CSV or SQLite (the logger build).
- Priorities — meter several circuits (the multi-module build) and shed the lowest-priority load first to stay under a whole-house limit.
- Prefer Home Assistant? The HA version — sliders, alerts, priority shedding — is real-time load control with rbAmp.