06 · Examples

import time
from smbus2 import SMBus
from rbamp import RbAmpFleet, RbAmpSensorClass
def configure_one(dev):
    variant = dev.read_variant()
    if variant == 1:                                    # UI1 mains
        dev.set_sensor_class(RbAmpSensorClass.SCT_013)
        dev.set_ct_model_ch(0, 3)                       # SCT-013-030
        dev.save_user_config()
    elif variant == 5:                                  # I2
        dev.configure_channels(RbAmpSensorClass.SCT_013, [1, 3])
    elif variant == 6:                                  # I3
        dev.configure_channels(RbAmpSensorClass.SCT_013, [1, 3, 6])
with SMBus(1) as bus:
    fleet = RbAmpFleet(bus)
    # 1) discover
    added = fleet.scan(match_product=True)
    print(f"fleet_scan: {added} module(s), {len(fleet.excluded)} excluded")
    # 2) configure per-module (mixed-CT on the sub-meters)
    for dev in fleet:
        configure_one(dev)
    # 3) arm GC across the whole fleet
    gc_ok = fleet.enable_gc_all(group=0)
    print(f"enable_gc_all → {gc_ok}/{len(fleet)} armed")
    # 4) loop: gc_latch → check_sync → poll_all → totals
    while True:
        tick = fleet.gc_latch(group=0)                  # auto-increment
        sync = fleet.check_sync(expected_tick=tick)
        n_missed = sum(1 for s in sync if not s.in_sync)
        snaps = fleet.poll_all()
        n_ok = sum(1 for snap, info in snaps if info.ok)
        p_total = fleet.total_power()
        e_total = fleet.total_energy_wh()
        n_total = len(fleet)
        print(f"ITER {tick}: GC {n_total - n_missed}/{n_total} in_sync | "
              f"poll n_ok={n_ok}/{n_total} | "
              f"P_total={p_total:.1f} W  E_total={e_total:.3f} Wh")
        time.sleep(1.0)

from machine import I2C, Pin
bus = I2C(0, scl=Pin(22), sda=Pin(21), freq=50_000)
fleet = RbAmpFleet(bus)
# ... the rest is the same ...

fleet_scan: 3 module(s), 0 excluded
  [0] @0x50  channels=1  voltage=yes   (mains meter: U + 1 current)
  [1] @0x51  channels=2  voltage=no    (2-channel current sub-meter)
  [2] @0x52  channels=3  voltage=no    (3-channel current sub-meter)
configure_channels(2ch) → mirrors [1 2 -]
configure_channels(3ch) → mirrors [1 2 3]
enable_gc_all → 3/3 armed
ITER N: GC 3/3 in_sync | poll n_ok=3/3 | P_total≈1260 W  E_total: 0 → 27.84 Wh
   @0x50  V=232.6  I=[5.79]                f=50.00
   @0x51  V= —     I=[11.58, 3.87]         f=50.00
   @0x52  V= —     I=[ 3.84, 2.55, 5.84]   f=50.00  (3rd input unused)

import time
from smbus2 import SMBus
from rbamp import RbAmp, RbAmpFleet, RbAmpSensorClass, RbAmpError
def provision_new_module(bus, fleet, desired_addr, label=None):
    """
    PRECONDITION: exactly one virgin on the bus at 0x50.
    MUST be one-virgin-at-a-time — see ch.04 + ch.10.
    """
    try:
        # 1) handle on the factory-default 0x50
        dev = RbAmp(bus, 0x50)
        dev.begin()
        # 2) two-phase address commit
        dev.prepare_address_change(desired_addr)   # arms candidate, 5 s window
        dev.commit_address_change()                # magic + commit + reset
        time.sleep(0.3)                             # boot window
        # 3) optional config — label + sensor class + CT model
        if label:
            dev.set_label(label)
        dev.set_sensor_class(RbAmpSensorClass.SCT_013)
        dev.set_ct_model_ch(0, 3)                  # SCT-013-030
        dev.save_user_config()                      # persist
        # 4) add to the fleet
        fleet.add(dev)
        print(f"provisioned virgin → 0x{desired_addr:02X} (+saved)")
        return dev
    except RbAmpError as e:
        print(f"provision @0x{desired_addr:02X} failed: {e}")
        return None
with SMBus(1) as bus:
    fleet = RbAmpFleet(bus)
    fleet.scan()
    provision_new_module(bus, fleet, 0x52, label="kitchen")

before: present(0x50)=yes  present(0x52)=no
virgin @0x50: variant=3-channel, channels=3
prepare_address_change(0x52) OK
commit_address_change() OK (lib log: "provisioned virgin → 0x52 (+saved)")
post:   handle now answers @0x52, is_provisioned()=True
after:  present(0x50)=no   present(0x52)=yes

import time
from smbus2 import SMBus
from rbamp import RbAmp, RbAmpSensorClass, RbAmpParamError
with SMBus(1) as bus:
    dev = RbAmp(bus, 0x52)                          # I3 sub-meter
    dev.begin()
    # (a) Configure — three CT models in one call
    models = [1, 3, 6]
    # 1=SCT-013-005 (5A), 3=SCT-013-030 (30A), 6=SCT-013-020 (20A)
    t0 = time.monotonic()
    try:
        dev.configure_channels(RbAmpSensorClass.SCT_013, models)
    except RbAmpParamError as e:
        print(f"configure rejected: {e}")
        raise
    t1 = time.monotonic()
    print(f"configure_channels latency = {(t1-t0)*1000:.0f} ms")
    # (b) Verify the mirror via registers 0x51 / 0x52 / 0x53 (RAM)
    m = [dev.read_ct_model_ch(ch) for ch in range(3)]
    print(f"applied: {m}  (expected [1, 3, 6])")
    # (c) Persist verify — reset + re-read the mirror after boot
    dev.reset()
    time.sleep(0.3)
    m = [dev.read_ct_model_ch(ch) for ch in range(3)]
    print(f"after reboot: {m}")

configure_channels(SCT_013, [1, 3, 6]) = OK
  → read_ct_model_ch 0/1/2 = [1, 3, 6]   raw mirror 0x51/52/53 = [01 03 06]
  → ch0 NOT clobbered (Fix-A canon: bind order-independent)
configure_channels latency ≈ 1400 ms
after reboot: [1, 3, 6]  (persisted through reset)

import time
from smbus2 import SMBus
from rbamp import RbAmpFleet
with SMBus(1) as bus:
    fleet = RbAmpFleet(bus)
    fleet.scan()
    # Step 1: enable GC on every module in the fleet
    gc_ok = fleet.enable_gc_all(group=0)
    print(f"GC armed: {gc_ok}/{len(fleet)}")
    # Steps 2-3: loop with broadcast latch + witness check
    for round_idx in range(10):
        tick = fleet.gc_latch(group=0)             # auto-increment tick
        sync = fleet.check_sync(expected_tick=tick)
        n_in_sync = sum(1 for s in sync if s.in_sync)
        print(f"round {round_idx}  tick={tick}  in_sync={n_in_sync}/{len(fleet)}")
        time.sleep(0.1)

enable_gc_all → 3/3 modules armed
round 0   tick=0   in_sync=3/3
round 1   tick=1   in_sync=3/3
...
round 9   tick=9   in_sync=3/3      skew=0 (validated at tick=0xABCD)

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, RbAmpStaleError
ADDRS = [0x50, 0x51, 0x52]
with SMBus(1) as bus:
    devs = [RbAmp(bus, addr) for addr in ADDRS]
    for dev in devs:
        dev.begin()
    while True:
        time.sleep(60)
        # Phase 1: sequential LATCH on each device, measure the skew
        t_start = time.monotonic()
        for dev in devs:
            dev.latch_period()
        sync_ms = (time.monotonic() - t_start) * 1000
        # Phase 2: a single shared settle for all 3 modules
        time.sleep(0.050)
        # Phase 3: read the snapshots with skip_latch=True
        print(f"sync_ms={sync_ms:.1f}")
        for i, dev in enumerate(devs):
            try:
                snap = dev.read_period_snapshot(settle_ms=0, skip_latch=True)
            except RbAmpStaleError:
                continue
            print(f"  mod{i} 0x{ADDRS[i]:02X}  P={snap.avg_p[0]:.0f} W  "
                  f"Wh={dev.energy.wh(0):.3f}")

import time
from machine import I2C, Pin
from rbamp import RbAmp, RbAmpStaleError
ADDRS = [0x50, 0x51, 0x52]
i2c = I2C(0, scl=Pin(22), sda=Pin(21), freq=50_000)
devs = [RbAmp(i2c, addr) for addr in ADDRS]
for dev in devs:
    dev.begin()
while True:
    time.sleep(60)
    t_start = time.ticks_ms()
    for dev in devs:
        dev.latch_period()
    sync_ms = time.ticks_diff(time.ticks_ms(), t_start)
    time.sleep_ms(50)
    print("sync_ms=", sync_ms)
    for i, dev in enumerate(devs):
        try:
            snap = dev.read_period_snapshot(settle_ms=0, skip_latch=True)
        except RbAmpStaleError:
            continue
        print("  mod{} 0x{:02X}  P={:.0f} W  Wh={:.3f}".format(
            i, ADDRS[i], 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, RbAmpStaleError, RbAmpSensorClass
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()
    mqtt = mqtt_client.Client("home-balance")
    mqtt.connect("192.168.1.10", 1883, keepalive=60)
    mqtt.loop_start()
    while True:
        time.sleep(60)
        # Sequential LATCH + shared settle (the Scenario 7 pattern — legacy sequential)
        for d in devs:
            d.latch_period()
        time.sleep(0.050)
        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)

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
# Stub for publishing. Replace with real WiFi+MQTT logic —
# see Scenario 8 (mqtt_publisher) for a template. In the 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: retained between 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 "dirty" data.
        # Save the flag and go straight back to sleep — the next wake in
        # SLEEP_MS will get a full accumulator over 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 —
    # that is, the accumulator over 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 the 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