08 · Cloud integrations

import ssl
ctx = ssl.create_default_context(cafile="/etc/ssl/certs/ca-certificates.crt")
# For mutual TLS (AWS IoT) — add the device cert + key:
ctx.load_cert_chain(certfile="device.cert.pem", keyfile="device.private.key")

import json, ssl, time
from smbus2 import SMBus
import paho.mqtt.client as mqtt_client
from rbamp import RbAmp, RbAmpSensorClass, RbAmpStaleError
AWS_ENDPOINT  = "xxxxxx-ats.iot.eu-west-1.amazonaws.com"
AWS_CLIENT_ID = "rbamp-main"
ctx = ssl.create_default_context(cafile="AmazonRootCA1.pem")
ctx.load_cert_chain(certfile="device.cert.pem",
                    keyfile="device.private.key")
mqtt = mqtt_client.Client(AWS_CLIENT_ID, protocol=mqtt_client.MQTTv311)
mqtt.tls_set_context(ctx)
mqtt.connect(AWS_ENDPOINT, 8883, keepalive=60)
mqtt.loop_start()
with SMBus(1) as bus, RbAmp(bus, 0x50) as dev:
    dev.set_sensor_class(RbAmpSensorClass.SCT_013)
    dev.set_ct_model(3)
    while True:
        time.sleep(60)
        try:
            snap = dev.read_period_snapshot()
        except RbAmpStaleError:
            continue
        payload = {
            "voltage":  round(dev.voltage, 1),
            "power":    round(snap.avg_p[0], 1),
            "energy":   round(dev.energy.wh(0), 3),
            "freq":     round(dev.frequency, 1),
        }
        mqtt.publish("rbamp/main/state", json.dumps(payload), qos=1)

import time, ujson, ssl
from machine import I2C, Pin
from umqtt.simple import MQTTClient
from rbamp import RbAmp, RbAmpSensorClass, RbAmpStaleError
AWS_ENDPOINT  = b"xxxxxx-ats.iot.eu-west-1.amazonaws.com"
# Cert files in `/lib/certs/` (copied over via mpremote cp)
with open("/lib/certs/AmazonRootCA1.pem", "rb") as f:
    AWS_ROOT_CA = f.read()
with open("/lib/certs/device.cert.pem", "rb") as f:
    DEV_CERT = f.read()
with open("/lib/certs/device.private.key", "rb") as f:
    DEV_KEY = f.read()
ctx = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT)
ctx.load_verify_locations(cadata=AWS_ROOT_CA)
ctx.load_cert_chain(certfile="/lib/certs/device.cert.pem",
                    keyfile="/lib/certs/device.private.key")
mqtt = MQTTClient(b"rbamp-main", AWS_ENDPOINT, port=8883,
                  keepalive=60, ssl=ctx)
mqtt.connect()
i2c = I2C(0, scl=Pin(22), sda=Pin(21), freq=50_000)
with RbAmp(i2c, 0x50) as dev:
    dev.set_sensor_class(RbAmpSensorClass.SCT_013)
    dev.set_ct_model(3)
    while True:
        time.sleep(60)
        try:
            snap = dev.read_period_snapshot()
        except RbAmpStaleError:
            continue
        payload = ujson.dumps({
            "voltage":  round(dev.voltage, 1),
            "power":    round(snap.avg_p[0], 1),
            "energy":   round(dev.energy.wh(0), 3),
            "freq":     round(dev.frequency, 1),
        })
        mqtt.publish(b"rbamp/main/state", payload.encode(), qos=1)
        mqtt.ping()   # keepalive

import boto3, json
iot = boto3.client("iot-data", region_name="eu-west-1")
iot.publish(topic="rbamp/main/state",
            qos=1, payload=json.dumps(payload))

import json, ssl, time
import paho.mqtt.client as mqtt_client
AZ_HOST      = "foo.azure-devices.net"
AZ_DEVICE    = "rbamp-main"
AZ_USERNAME  = f"{AZ_HOST}/{AZ_DEVICE}/?api-version=2021-04-12"
AZ_SAS       = "SharedAccessSignature sr=...&sig=...&se=..."   # see above
ctx = ssl.create_default_context()   # Uses the system CAs
mqtt = mqtt_client.Client(AZ_DEVICE, protocol=mqtt_client.MQTTv311)
mqtt.username_pw_set(AZ_USERNAME, password=AZ_SAS)
mqtt.tls_set_context(ctx)
mqtt.connect(AZ_HOST, 8883, keepalive=60)
mqtt.loop_start()
# Inside your 60-second loop:
mqtt.publish(f"devices/{AZ_DEVICE}/messages/events/",
             json.dumps(payload), qos=1)

from umqtt.simple import MQTTClient
import ssl
ctx = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT)
# Azure root CA (Baltimore CyberTrust or DigiCert Global G2):
with open("/lib/certs/azure_ca.pem", "rb") as f:
    ctx.load_verify_locations(cadata=f.read())
mqtt = MQTTClient(b"rbamp-main", b"foo.azure-devices.net", port=8883,
                  user=AZ_USERNAME.encode(),
                  password=AZ_SAS.encode(),
                  keepalive=60, ssl=ctx)
mqtt.connect()
# In the loop:
mqtt.publish(b"devices/rbamp-main/messages/events/",
             ujson.dumps(payload).encode(), qos=1)

import requests, time
from smbus2 import SMBus
from rbamp import RbAmp, RbAmpSensorClass, RbAmpStaleError
INFLUX_URL   = "https://us-east-1-1.aws.cloud2.influxdata.com/api/v2/write?org=MyOrg&bucket=energy&precision=s"
INFLUX_TOKEN = "your-rw-token"
def push_influx(u, p, e_wh):
    body = f"rbamp,device=main voltage={u:.1f},power={p:.1f},energy={e_wh:.3f}"
    try:
        r = requests.post(INFLUX_URL, data=body,
                          headers={
                              "Authorization": f"Token {INFLUX_TOKEN}",
                              "Content-Type":  "text/plain",
                          },
                          timeout=10)   # TLS handshake can take up to 3 s
        if r.status_code != 204:
            print(f"influx HTTP {r.status_code}: {r.text}")
    except requests.RequestException as e:
        print("influx request failed:", e)
with SMBus(1) as bus, RbAmp(bus, 0x50) as dev:
    dev.set_sensor_class(RbAmpSensorClass.SCT_013)
    dev.set_ct_model(3)
    while True:
        time.sleep(60)
        try:
            snap = dev.read_period_snapshot()
        except RbAmpStaleError:
            continue
        push_influx(dev.voltage, snap.avg_p[0], dev.energy.wh(0))

import urequests as requests, ujson, ssl, time
from machine import I2C, Pin
from rbamp import RbAmp, RbAmpSensorClass, RbAmpStaleError
INFLUX_URL   = "https://us-east-1-1.aws.cloud2.influxdata.com/api/v2/write?org=MyOrg&bucket=energy&precision=s"
INFLUX_TOKEN = "your-rw-token"
# Custom SSL context with embedded CA (DigiCert Global G2)
ctx = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT)
with open("/lib/certs/digicert_g2.pem", "rb") as f:
    ctx.load_verify_locations(cadata=f.read())
def push_influx(u, p, e_wh):
    body = "rbamp,device=main voltage={:.1f},power={:.1f},energy={:.3f}".format(u, p, e_wh)
    try:
        r = requests.post(INFLUX_URL, data=body,
                          headers={
                              "Authorization": "Token " + INFLUX_TOKEN,
                              "Content-Type":  "text/plain",
                          })
        if r.status_code != 204:
            print("influx HTTP", r.status_code)
        r.close()   # MicroPython socket cleanup
    except OSError as e:
        print("influx request failed:", e)

import requests, time, json
WEBHOOK_URL = "https://your-api.example.com/ingest"
API_KEY     = "Bearer your-token-here"
def push_webhook(u, p, e_wh):
    body = {
        "ts":      time.time(),
        "voltage": u,
        "power":   p,
        "energy":  e_wh,
    }
    try:
        r = requests.post(WEBHOOK_URL, json=body, headers={
            "Authorization": API_KEY,
        }, timeout=10)
        if not (200 <= r.status_code < 300):
            print(f"webhook HTTP {r.status_code}: {r.text}")
    except requests.RequestException as e:
        print("webhook failed:", e)

import urequests as requests, ujson, time
def push_webhook(u, p, e_wh):
    body = ujson.dumps({
        "ts":      time.time(),
        "voltage": u,
        "power":   p,
        "energy":  e_wh,
    })
    try:
        r = requests.post(WEBHOOK_URL, data=body, headers={
            "Authorization": API_KEY,
            "Content-Type":  "application/json",
        })
        if not (200 <= r.status_code < 300):
            print("webhook HTTP", r.status_code)
        r.close()
    except OSError as e:
        print("webhook failed:", e)

import logging, time, csv, requests
from logging.handlers import RotatingFileHandler
from smbus2 import SMBus
from rbamp import RbAmp, RbAmpSensorClass, 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)
def sync_to_cloud_if_due():
    """Once an hour: read rbamp.csv, push the new rows to InfluxDB Cloud."""
    # ...uses push_influx from the section above + a state file for the offset...
with SMBus(1) as bus, RbAmp(bus, 0x50) as dev:
    dev.set_sensor_class(RbAmpSensorClass.SCT_013)
    dev.set_ct_model(3)
    last_sync = 0
    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}")
        if time.time() - last_sync > 3600:
            sync_to_cloud_if_due()
            last_sync = time.time()

import time, ujson
def append_local(snap, dev):
    with open("/lib/log.csv", "a") as f:
        f.write("{:.0f},{:.1f},{:.4f},{}\n".format(
            time.time(), snap.avg_p[0], dev.energy.wh(0),
            snap.master_dt_ms))
def sync_if_due():
    """Once an hour: read the accumulated log, push it to the cloud."""
    try:
        with open("/lib/log.csv") as f:
            for line in f:
                # ...push_webhook(line) with retry on failure...
                pass
        # archive successful upload
        import os
        os.rename("/lib/log.csv", "/lib/log_pushed.csv")
    except OSError:
        pass   # offline / no file yet