Examples

import time
from smbus2 import SMBus
from rbamp import RbAmp, RbAmpError
with SMBus(1) as bus, RbAmp(bus, 0x50) as dev:
    while True:
        try:
            s = dev.read_all()
        except RbAmpError as e:
            print("read fail:", e)
            time.sleep(1)
            continue
        line = f"U={s.voltage:.1f} V  f={s.frequency} Hz   "
        for ch in range(s.channels):
            line += f"I{ch}={s.current[ch]:.2f}A  P{ch}={s.power[ch]:.1f}W  "
        print(line)
        time.sleep(1)

import time
from machine import I2C, Pin
from rbamp import RbAmp, RbAmpError
i2c = I2C(0, scl=Pin(22), sda=Pin(21), freq=50_000)
with RbAmp(i2c, 0x50) as dev:
    while True:
        try:
            s = dev.read_all()
        except RbAmpError as e:
            print("read fail:", e)
            time.sleep(1)
            continue
        print("U={:.1f} V  f={} Hz".format(s.voltage, s.frequency))
        for ch in range(s.channels):
            print("  I{}={:.2f}A  P{}={:.1f}W".format(
                ch, s.current[ch], ch, s.power[ch]))
        time.sleep(1)

import time
from machine import I2C, Pin
from ssd1306 import SSD1306_I2C
from rbamp import RbAmp, RbAmpStaleError, RbAmpSensorClass
i2c = I2C(0, scl=Pin(22), sda=Pin(21), freq=50_000)
oled = SSD1306_I2C(128, 64, i2c)   # OLED on the same bus, addr 0x3C
with RbAmp(i2c, 0x50) as dev:
    snapshots_ok = 0
    snapshots_bad = 0
    while True:
        time.sleep(60)
        try:
            snap = dev.read_period_snapshot()
            snapshots_ok += 1
        except RbAmpStaleError:
            snapshots_bad += 1
            continue
        oled.fill(0)
        oled.text("rbAmp Energy Meter", 0, 0)
        oled.text("{:.1f} W".format(snap.avg_p[0]), 0, 16, 1)
        oled.text("Wh: {:.4f}".format(dev.energy.wh(0)), 0, 32)
        oled.text("ok={} bad={}".format(snapshots_ok, snapshots_bad), 0, 48)
        oled.show()

import time
from smbus2 import SMBus
from rbamp import RbAmp, RbAmpStaleError
with SMBus(1) as bus, RbAmp(bus, 0x50) as dev:
    snapshots_ok = 0
    snapshots_bad = 0
    while True:
        time.sleep(60)
        try:
            snap = dev.read_period_snapshot()
            snapshots_ok += 1
        except RbAmpStaleError:
            snapshots_bad += 1
            print("stale snapshot — skipping")
            continue
        print(f"P = {snap.avg_p[0]:.2f} W   "
              f"Wh = {dev.energy.wh(0):.4f}   "
              f"ok={snapshots_ok} bad={snapshots_bad}   "
              f"dt={snap.master_dt_ms} ms")

import time
from smbus2 import SMBus
from rbamp import RbAmp, RbAmpFleet, RbAmpError, RbAmpStaleError
ADDRS = [0x50, 0x51, 0x52]
GC_GROUP = 0
with SMBus(1) as bus:
    fleet = RbAmpFleet(bus)
    for addr in ADDRS:
        try:
            dev = RbAmp(bus, addr); dev.begin()
        except RbAmpError as exc:
            print(f"skip 0x{addr:02X}: {exc}"); continue
        fleet.add(dev)
    # one-time: persist the General-Call latch opt-in on every module
    fleet.enable_gc_all(group=GC_GROUP)
    while True:
        time.sleep(60)
        tick = fleet.gc_latch(group=GC_GROUP)     # one all-call frame latches the bus
        time.sleep(0.050)                         # single shared settle
        sync = fleet.check_sync(expected_tick=tick)
        for dev, s in zip(fleet, sync):
            if not s.in_sync:                     # this module missed the frame
                print(f"  0x{s.addr:02X}  MISSED tick 0x{tick:04X}")
                continue
            try:
                snap = dev.read_period_snapshot(settle_ms=0, skip_latch=True)
            except RbAmpStaleError:
                continue
            print(f"  0x{dev.address:02X}  P={snap.avg_p[0]:.0f} W  "
                  f"Wh={dev.energy.wh(0):.3f}")
        print(f"  fleet ΣWh={fleet.total_energy_wh():.3f}  "
              f"in_sync={sum(s.in_sync for s in sync)}/{len(sync)}")

import time
from machine import I2C, Pin
from rbamp import RbAmp, RbAmpFleet, RbAmpError, RbAmpStaleError
ADDRS = [0x50, 0x51, 0x52]
GC_GROUP = 0
i2c = I2C(0, scl=Pin(22), sda=Pin(21), freq=50_000)
fleet = RbAmpFleet(i2c)
for addr in ADDRS:
    try:
        dev = RbAmp(i2c, addr); dev.begin()
    except RbAmpError as exc:
        print("skip 0x{:02X}:".format(addr), exc); continue
    fleet.add(dev)
fleet.enable_gc_all(group=GC_GROUP)           # one-time, persisted
while True:
    time.sleep(60)
    tick = fleet.gc_latch(group=GC_GROUP)     # one all-call frame
    time.sleep_ms(50)                         # single shared settle
    sync = fleet.check_sync(expected_tick=tick)
    for dev, s in zip(fleet, sync):
        if not s.in_sync:
            print("  0x{:02X}  MISSED tick".format(s.addr))
            continue
        try:
            snap = dev.read_period_snapshot(settle_ms=0, skip_latch=True)
        except RbAmpStaleError:
            continue
        print("  0x{:02X}  P={:.0f} W  Wh={:.3f}".format(
            dev.address, snap.avg_p[0], dev.energy.wh(0)))

import time, json
from smbus2 import SMBus
import paho.mqtt.client as mqtt_client
from rbamp import RbAmp, RbAmpStaleError, RbAmpSensorClass
CH_NAMES = ["main", "heatpump", "lights"]
mqtt = mqtt_client.Client("rbamp-ui3")
mqtt.connect("192.168.1.10", 1883, keepalive=60)
mqtt.loop_start()
with SMBus(1) as bus, RbAmp(bus, 0x50) as dev:
    while True:
        time.sleep(60)
        try:
            snap = dev.read_period_snapshot()
        except RbAmpStaleError:
            continue
        for ch in range(dev.channels):
            payload = json.dumps({
                "power":  round(snap.avg_p[ch], 1),
                "energy": round(dev.energy.wh(ch), 4),
            })
            mqtt.publish(f"rbamp/{CH_NAMES[ch]}/state",
                         payload, qos=1, retain=True)

import time, ujson
from machine import I2C, Pin
from umqtt.simple import MQTTClient
from rbamp import RbAmp, RbAmpStaleError, RbAmpSensorClass
CH_NAMES = [b"main", b"heatpump", b"lights"]
mqtt = MQTTClient(b"rbamp-ui3", b"192.168.1.10",
                  port=1883, keepalive=60)
mqtt.connect()
i2c = I2C(0, scl=Pin(22), sda=Pin(21), freq=50_000)
with RbAmp(i2c, 0x50) as dev:
    while True:
        time.sleep(60)
        try:
            snap = dev.read_period_snapshot()
        except RbAmpStaleError:
            continue
        for ch in range(dev.channels):
            payload = ujson.dumps({
                "power":  round(snap.avg_p[ch], 1),
                "energy": round(dev.energy.wh(ch), 4),
            })
            topic = b"rbamp/" + CH_NAMES[ch] + b"/state"
            mqtt.publish(topic, payload.encode(), qos=1, retain=True)

import time
from smbus2 import SMBus
from rbamp import RbAmp, RbAmpIOError
with SMBus(1) as bus, RbAmp(bus, 0x50) as dev:
    consume_wh = 0.0
    export_wh  = 0.0
    t_prev = time.monotonic()
    while True:
        time.sleep(0.2)   # 5 Hz, matches the RT cadence
        t_now = time.monotonic()
        dt_s  = t_now - t_prev
        t_prev = t_now
        try:
            p = dev.read_power(0)
        except RbAmpIOError:
            continue
        dwh = p * dt_s / 3600.0
        if p >= 0:
            consume_wh += dwh
        else:
            export_wh  += -dwh
        # Print once per second
        if int(t_now) % 5 == 0:
            print(f"P={p:.1f} W   "
                  f"cons={consume_wh:.4f} Wh  exp={export_wh:.4f} Wh  "
                  f"net={consume_wh - export_wh:.4f} Wh")

import time
from machine import I2C, Pin
from rbamp import RbAmp, RbAmpIOError
i2c = I2C(0, scl=Pin(22), sda=Pin(21), freq=50_000)
with RbAmp(i2c, 0x50) as dev:
    consume_wh = 0.0
    export_wh  = 0.0
    t_prev = time.ticks_ms()
    while True:
        time.sleep_ms(200)
        t_now = time.ticks_ms()
        dt_s  = time.ticks_diff(t_now, t_prev) / 1000
        t_prev = t_now
        try:
            p = dev.read_power(0)
        except RbAmpIOError:
            continue
        dwh = p * dt_s / 3600.0
        if p >= 0:
            consume_wh += dwh
        else:
            export_wh  += -dwh

import time, json
from smbus2 import SMBus
import paho.mqtt.client as mqtt_client
from rbamp import (RbAmp, RbAmpFleet, RbAmpError,
                   RbAmpStaleError, RbAmpSensorClass)
GC_GROUP = 0
with SMBus(1) as bus:
    mains = RbAmp(bus, 0x50)   # bidirectional mains
    solar = RbAmp(bus, 0x51)   # generation only
    loads = RbAmp(bus, 0x52)   # UI3 — per-appliance
    devs = [mains, solar, loads]
    for d in devs:
        d.begin()
    fleet = RbAmpFleet(bus)
    for d in devs:
        fleet.add(d)
    fleet.enable_gc_all(group=GC_GROUP)   # one-time, persisted
    mqtt = mqtt_client.Client("home-balance")
    mqtt.connect("192.168.1.10", 1883, keepalive=60)
    mqtt.loop_start()
    while True:
        time.sleep(60)
        # One coherent all-call latch for the whole bus (see Scenario 3)
        tick = fleet.gc_latch(group=GC_GROUP)
        time.sleep(0.050)                     # single shared settle
        sync = fleet.check_sync(expected_tick=tick)
        try:
            sm = mains.read_period_snapshot(settle_ms=0, skip_latch=True)
            ss = solar.read_period_snapshot(settle_ms=0, skip_latch=True)
            sl = loads.read_period_snapshot(settle_ms=0, skip_latch=True)
        except RbAmpStaleError:
            continue
        payload = {
            "mains":  round(mains.energy.wh(0), 3),
            "solar":  round(solar.energy.wh(0), 3),
            "hp":     round(loads.energy.wh(0), 3),
            "ac":     round(loads.energy.wh(1), 3),
            "ev":     round(loads.energy.wh(2), 3),
        }
        mqtt.publish("home/energy/balance",
                     json.dumps(payload), qos=1, retain=True)
        # let a dashboard alarm on a partial fleet (a module that missed the frame)
        in_sync = sum(s.in_sync for s in sync)
        mqtt.publish("home/energy/sync/in_sync",
                     f"{in_sync}/{len(sync)}", qos=1, retain=True)

import time
from smbus2 import SMBus
from rbamp import RbAmp, RbAmpIOError
EMA_ALPHA = 0.05
EVENT_THRESHOLD_W = 200.0
with SMBus(1) as bus, RbAmp(bus, 0x50) as dev:
    try:
        p_ema = dev.read_power(0)   # seed so the first read isn't an event
    except RbAmpIOError:
        p_ema = 0.0
    while True:
        time.sleep(0.2)
        try:
            p = dev.read_power(0)
        except RbAmpIOError:
            continue
        delta = p - p_ema
        p_ema = (1 - EMA_ALPHA) * p_ema + EMA_ALPHA * p
        if abs(delta) > EVENT_THRESHOLD_W:
            action = "TURN_ON" if delta > 0 else "TURN_OFF"
            print(f"{time.time():.0f}  {action}  delta={delta:.0f} W  "
                  f"P={p:.0f} W  EMA={p_ema:.0f}")

import time, logging
from logging.handlers import RotatingFileHandler
from smbus2 import SMBus
from rbamp import RbAmp, RbAmpStaleError
log = logging.getLogger("rbamp.csv")
log.setLevel(logging.INFO)
handler = RotatingFileHandler("rbamp.csv", maxBytes=1_000_000, backupCount=5)
handler.setFormatter(logging.Formatter("%(asctime)s,%(message)s",
                                       datefmt="%Y-%m-%dT%H:%M:%S"))
log.addHandler(handler)
with SMBus(1) as bus, RbAmp(bus, 0x50) as dev:
    while True:
        time.sleep(60)
        try:
            snap = dev.read_period_snapshot()
        except RbAmpStaleError:
            log.warning("stale")
            continue
        log.info(f"{snap.avg_p[0]:.1f},{dev.energy.wh(0):.4f},"
                 f"{snap.master_dt_ms}")

import time, machine
from machine import I2C, Pin
from rbamp import RbAmp, RbAmpSensorClass
# Publishing stub. Replace with real WiFi+MQTT logic —
# see Scenario 4 (mqtt_publisher) for a template. In a deep-sleep
# scenario, the WiFi stack is brought up/torn down on every wake,
# which eats up the bulk of the energy budget.
def publish_via_wifi(power_w, total_wh, dt_s):
    print("P=%.1f W  Wh=%.4f  dt=%.1f s" % (power_w, total_wh, dt_s))
SLEEP_MS = 10 * 60 * 1000             # 10 minutes — fixed interval
RTC_MAGIC = b"\xCA\xFE\xFE\xED"
# RTC slow memory: persisted across deep-sleep wakes
rtc = machine.RTC()
state = rtc.memory()                  # bytes; empty on cold boot
def save_state(magic, wh_total, primer_done):
    # Simple layout: 4 bytes magic + 8 bytes float wh + 1 byte primer_done
    import struct
    rtc.memory(magic + struct.pack("dB", wh_total, 1 if primer_done else 0))
def load_state():
    import struct
    if len(state) < 4 + 8 + 1 or state[:4] != RTC_MAGIC:
        return False, 0.0, False
    wh_total, primer_done_byte = struct.unpack("dB", state[4:4+9])
    return True, wh_total, bool(primer_done_byte)
valid, rtc_total_wh, primer_done = load_state()
if not valid:
    rtc_total_wh = 0.0
    primer_done = False
i2c = I2C(0, scl=Pin(22), sda=Pin(21), freq=50_000)
with RbAmp(i2c, 0x50) as dev:    # __enter__ automatically calls begin() — the LATCH primer
    dev.energy.disable()         # the master owns Wh persistence itself
    if not primer_done:
        # First wake: the primer just latched away the "dirty" data.
        # Save the flag and go straight back to sleep — the next wake,
        # SLEEP_MS later, will get a full accumulator for the interval.
        save_state(RTC_MAGIC, rtc_total_wh, primer_done=True)
        machine.deepsleep(SLEEP_MS)
    # skip_latch=True reads what the begin() primer latched —
    # i.e. the accumulator for the elapsed sleep interval.
    snap = dev.read_period_snapshot(settle_ms=0, skip_latch=True)
    # Use the KNOWN sleep duration as dt
    dt_s = SLEEP_MS / 1000
    rtc_total_wh += snap.avg_p[0] * dt_s / 3600.0
    publish_via_wifi(snap.avg_p[0], rtc_total_wh, dt_s)  # stub
    save_state(RTC_MAGIC, rtc_total_wh, primer_done=True)
machine.deepsleep(SLEEP_MS)

import asyncio
from smbus2 import SMBus
from rbamp import RbAmp, RbAmpSensorClass
async def main():
    with SMBus(1) as bus, RbAmp(bus, 0x50) as dev:
        # Once per minute, skip_stale=True (default) — stale snapshots are
        # automatically skipped without raising RbAmpStaleError
        async for snap in dev.stream_period(interval_s=60.0):
            print(f"P={snap.avg_p[0]:.2f} W,  Wh={dev.energy.wh(0):.4f}")
asyncio.run(main())

import uasyncio as asyncio
from machine import I2C, Pin
from rbamp import RbAmp, RbAmpSensorClass
async def reader(dev):
    async for snap in dev.stream_period(interval_s=60.0):
        print("P={:.2f} W,  Wh={:.4f}".format(snap.avg_p[0], dev.energy.wh(0)))
async def main():
    i2c = I2C(0, scl=Pin(22), sda=Pin(21), freq=50_000)
    with RbAmp(i2c, 0x50) as dev:
        # Run reader + other async tasks in parallel
        await asyncio.gather(
            reader(dev),
            # other_async_task(),
        )
asyncio.run(main())

# /usr/local/bin/rbamp_daemon.py
import sys, signal, time, logging
from smbus2 import SMBus
from rbamp import RbAmp, RbAmpSensorClass
logging.basicConfig(level=logging.INFO,
                    format="%(asctime)s %(levelname)s %(message)s")
log = logging.getLogger("rbamp")
shutdown = False
def on_signal(sig, frame):
    global shutdown
    log.info("Received signal %d, shutting down gracefully", sig)
    shutdown = True
signal.signal(signal.SIGTERM, on_signal)
signal.signal(signal.SIGINT,  on_signal)
with SMBus(1) as bus, RbAmp(bus, 0x50) as dev:
    log.info("rbAmp daemon started")
    while not shutdown:
        try:
            snap = dev.read_period_snapshot()
            log.info("P=%.1f W, Wh=%.4f", snap.avg_p[0], dev.energy.wh(0))
        except Exception as e:
            log.warning("snapshot failed: %s", e)
        time.sleep(60)
log.info("rbAmp daemon stopped")

# /etc/systemd/system/rbamp.service
[Unit]
Description=rbAmp I2C AC sensor daemon
After=network.target
[Service]
Type=simple
ExecStart=/usr/local/bin/rbamp_daemon.py
Restart=on-failure
RestartSec=5
User=pi
[Install]
WantedBy=multi-user.target