Skip to main content

1

sudo apt install python3-venv

sudo mkdir -p /opt/shellyem
sudo python3 -m venv /opt/shellyem/venv
sudo /opt/shellyem/venv/bin/pip install requests prometheus_client

sudo vi /opt/shellyem/shellyem_exporter.py

#!/usr/bin/env python3
"""
Polls a Shelly Pro 3EM's local RPC API and exposes the readings on /metrics in Prometheus text format.

Requirements:
    pip install requests prometheus_client

Usage:
    python3 shellyem_exporter.py --shelly-ip 192.168.1.50 --port 9091

Then point Prometheus at http://<this-host>:9091/metrics
"""

import argparse
import logging
import time

import requests
from prometheus_client import start_http_server, Gauge

logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s")
log = logging.getLogger("shellyem_exporter")

# Metric definitions. Each has a "phase" label (a/b/c) except the totals.
CURRENT = Gauge("shelly_em_current_amps", "Current in amps", ["phase"])
VOLTAGE = Gauge("shelly_em_voltage_volts", "Voltage in volts", ["phase"])
ACT_POWER = Gauge("shelly_em_active_power_watts", "Active power in watts", ["phase"])
APRT_POWER = Gauge("shelly_em_apparent_power_va", "Apparent power in VA", ["phase"])
POWER_FACTOR = Gauge("shelly_em_power_factor", "Power factor", ["phase"])
FREQUENCY = Gauge("shelly_em_frequency_hz", "Line frequency in Hz", ["phase"])

NEUTRAL_CURRENT = Gauge("shelly_em_neutral_current_amps", "Neutral current in amps (absent/0 if not measured)")
TOTAL_CURRENT = Gauge("shelly_em_total_current_amps", "Total current in amps")
TOTAL_ACT_POWER = Gauge("shelly_em_total_active_power_watts", "Total active power in watts")
TOTAL_APRT_POWER = Gauge("shelly_em_total_apparent_power_va", "Total apparent power in VA")

SCRAPE_SUCCESS = Gauge("shelly_scrape_success", "1 if the last scrape of the Shelly device succeeded")
SCRAPE_DURATION = Gauge("shelly_scrape_duration_seconds", "Time taken to fetch data from the Shelly device")

PHASES = ("a", "b", "c")


def fetch_em_status(shelly_ip: str, em_id: int, timeout: float, auth=None) -> dict:
    url = f"http://{shelly_ip}/rpc/EM.GetStatus?id={em_id}"
    resp = requests.get(url, timeout=timeout, auth=auth)
    resp.raise_for_status()
    return resp.json()


def update_metrics(data: dict) -> None:
    for phase in PHASES:
        CURRENT.labels(phase=phase).set(data.get(f"{phase}_current", 0) or 0)
        VOLTAGE.labels(phase=phase).set(data.get(f"{phase}_voltage", 0) or 0)
        ACT_POWER.labels(phase=phase).set(data.get(f"{phase}_act_power", 0) or 0)
        APRT_POWER.labels(phase=phase).set(data.get(f"{phase}_aprt_power", 0) or 0)
        POWER_FACTOR.labels(phase=phase).set(data.get(f"{phase}_pf", 0) or 0)
        FREQUENCY.labels(phase=phase).set(data.get(f"{phase}_freq", 0) or 0)

    # n_current can be null if the device doesn't measure/report it
    NEUTRAL_CURRENT.set(data.get("n_current") or 0)
    TOTAL_CURRENT.set(data.get("total_current", 0) or 0)
    TOTAL_ACT_POWER.set(data.get("total_act_power", 0) or 0)
    TOTAL_APRT_POWER.set(data.get("total_aprt_power", 0) or 0)


def poll_loop(shelly_ip: str, em_id: int, interval: float, timeout: float, auth=None) -> None:
    while True:
        start = time.time()
        try:
            data = fetch_em_status(shelly_ip, em_id, timeout, auth)
            update_metrics(data)
            SCRAPE_SUCCESS.set(1)
            log.info("scrape ok: total_act_power=%.2fW", data.get("total_act_power", 0) or 0)
        except Exception as exc:
            SCRAPE_SUCCESS.set(0)
            log.warning("scrape failed: %s", exc)
        finally:
            SCRAPE_DURATION.set(time.time() - start)

        time.sleep(interval)


def main() -> None:
    parser = argparse.ArgumentParser(description="Shelly Pro 3EM Prometheus exporter")
    parser.add_argument("--shelly-ip", required=True, help="IP address or hostname of the Shelly Pro 3EM")
    parser.add_argument("--em-id", type=int, default=0, help="EM component id (default: 0)")
    parser.add_argument("--port", type=int, default=9091, help="Port to expose /metrics on (default: 9091)")
    parser.add_argument("--interval", type=float, default=5.0, help="Poll interval in seconds (default: 5)")
    parser.add_argument("--timeout", type=float, default=5.0, help="HTTP request timeout in seconds (default: 5)")
    parser.add_argument("--username", help="RPC auth username, if the Shelly has authentication enabled")
    parser.add_argument("--password", help="RPC auth password, if the Shelly has authentication enabled")
    args = parser.parse_args()

    auth = None
    if args.username and args.password:
        auth = requests.auth.HTTPDigestAuth(args.username, args.password)

    log.info("starting exporter: shelly=%s em_id=%s -> :%s/metrics (every %.1fs)",
              args.shelly_ip, args.em_id, args.port, args.interval)

    start_http_server(args.port)
    poll_loop(args.shelly_ip, args.em_id, args.interval, args.timeout, auth)


if __name__ == "__main__":
    main()
Back to top