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 pyyaml
sudo vi /opt/shellyem/shellyem_exporter.py
#!/usr/bin/env python3
"""
Polls aone or more Shelly Pro 3EM's3EM EM nodes (local RPC APIAPI) and exposes
thetheir readings on /metrics in Prometheus text format. Every metric is
labeled with `shelly_ip` and `em_id` so multiple devices/nodes can be
told apart.
Requirements:
pip install requests prometheus_client pyyaml
Usage:
python3 shellyem_exporter.shelly_exporter.py --shelly-ipconfig.file shellyem_exporter.yml --web.listen-address=:9091 --web.telemetry-path=/metrics
# --config.file defaults to "shellyem_exporter.yml" next to this script
# --web.listen-address defaults to ":9091" (all interfaces, port 9091)
# use e.g. --web.listen-address=127.0.0.1:9091 to bind a specific IP
# --web.telemetry-path defaults to "/metrics"
Config file format:
interval: 5 # optional, default poll interval (seconds)
timeout: 5 # optional, default HTTP timeout (seconds)
nodes:
- ip: 192.168.1.50
id: 0
--port 9091ip: 192.168.1.51
id: 0
interval: 10 # optional per-node override
timeout: 3 # optional per-node override
username: admin # optional, if RPC auth is enabled
password: secret
Then point Prometheus at http://<this-host>:9091/<port>/metrics
"""
import argparse
import logging
import threading
import time
from pathlib import Path
from socketserver import ThreadingMixIn
from wsgiref.simple_server import WSGIServer, WSGIRequestHandler, make_server
import requests
import yaml
from prometheus_client import start_http_server,make_wsgi_app, Gauge
logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s")
log = logging.getLogger("shellyem_exporter"shelly_exporter")
LABELS_PHASE = ["shelly_ip", "em_id", "phase"]
LABELS_NODE = ["shelly_ip", "em_id"]
# Metric definitions. Each has ashelly_ip/em_id labels, phase metrics also have "phase" label (a/b/c) except the totals..
CURRENT = Gauge("shelly_em_current_amps", "Current in amps", ["phase"])LABELS_PHASE)
VOLTAGE = Gauge("shelly_em_voltage_volts", "Voltage in volts", ["phase"])LABELS_PHASE)
ACT_POWER = Gauge("shelly_em_active_power_watts", "Active power in watts", ["phase"])LABELS_PHASE)
APRT_POWER = Gauge("shelly_em_apparent_power_va", "Apparent power in VA", ["phase"])LABELS_PHASE)
POWER_FACTOR = Gauge("shelly_em_power_factor", "Power factor", ["phase"])LABELS_PHASE)
FREQUENCY = Gauge("shelly_em_frequency_hz", "Line frequency in Hz", ["phase"])LABELS_PHASE)
NEUTRAL_CURRENT = Gauge("shelly_em_neutral_current_amps", "Neutral current in amps (absent/0 if not measured)"), LABELS_NODE)
TOTAL_CURRENT = Gauge("shelly_em_total_current_amps", "Total current in amps"), LABELS_NODE)
TOTAL_ACT_POWER = Gauge("shelly_em_total_active_power_watts", "Total active power in watts"), LABELS_NODE)
TOTAL_APRT_POWER = Gauge("shelly_em_total_apparent_power_va", "Total apparent power in VA"), LABELS_NODE)
SCRAPE_SUCCESS = Gauge("shelly_scrape_success", "1 if the last scrape of the Shelly devicenode succeeded"), LABELS_NODE)
SCRAPE_DURATION = Gauge("shelly_scrape_duration_seconds", "Time taken to fetch data from the Shellynode", device")LABELS_NODE)
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(shelly_ip: str, em_id: int, data: dict) -> None:
for phase in PHASES:
CURRENT.labels(phase=shelly_ip, em_id, phase).set(data.get(f"{phase}_current", 0) or 0)
VOLTAGE.labels(phase=shelly_ip, em_id, phase).set(data.get(f"{phase}_voltage", 0) or 0)
ACT_POWER.labels(phase=shelly_ip, em_id, phase).set(data.get(f"{phase}_act_power", 0) or 0)
APRT_POWER.labels(phase=shelly_ip, em_id, phase).set(data.get(f"{phase}_aprt_power", 0) or 0)
POWER_FACTOR.labels(phase=shelly_ip, em_id, phase).set(data.get(f"{phase}_pf", 0) or 0)
FREQUENCY.labels(phase=shelly_ip, em_id, 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.labels(shelly_ip, em_id).set(data.get("n_current") or 0)
TOTAL_CURRENT.labels(shelly_ip, em_id).set(data.get("total_current", 0) or 0)
TOTAL_ACT_POWER.labels(shelly_ip, em_id).set(data.get("total_act_power", 0) or 0)
TOTAL_APRT_POWER.labels(shelly_ip, em_id).set(data.get("total_aprt_power", 0) or 0)
def poll_loop(shelly_ip:poll_node(node: str,dict, em_id: int, interval:default_interval: float, timeout:default_timeout: float, auth=None)float) -> None:
shelly_ip = node["ip"]
em_id = int(node.get("id", 0))
interval = float(node.get("interval", default_interval))
timeout = float(node.get("timeout", default_timeout))
auth = None
if node.get("username") and node.get("password"):
auth = requests.auth.HTTPDigestAuth(node["username"], node["password"])
label_ip, label_id = shelly_ip, em_id
while True:
start = time.time()
try:
data = fetch_em_status(shelly_ip, em_id, timeout, auth)
update_metrics(label_ip, label_id, data)
SCRAPE_SUCCESS.labels(label_ip, label_id).set(1)
log.info("scrape ok:ok [%s id=%s]: total_act_power=%.2fW",
shelly_ip, em_id, data.get("total_act_power", 0) or 0)
except Exception as exc:
SCRAPE_SUCCESS.labels(label_ip, label_id).set(0)
log.warning("scrape failed:failed [%s id=%s]: %s", shelly_ip, em_id, exc)
finally:
SCRAPE_DURATION.labels(label_ip, label_id).set(time.time() - start)
time.sleep(interval)
def parse_listen_address(value: str) -> tuple:
"""Parse a Prometheus-exporter-style listen address like ':9091' or
'127.0.0.1:9091' into (host, port). An empty host means all interfaces."""
if ":" not in value:
raise ValueError(f"Invalid --web.listen-address {value!r}, expected HOST:PORT or :PORT")
host, _, port_str = value.rpartition(":")
try:
port = int(port_str)
except ValueError:
raise ValueError(f"Invalid port in --web.listen-address {value!r}")
return host or "0.0.0.0", port
class _ThreadingWSGIServer(ThreadingMixIn, WSGIServer):
daemon_threads = True
class _QuietWSGIRequestHandler(WSGIRequestHandler):
def log_message(self, fmt, *args):
pass # keep our own structured logging; suppress default access log
def make_routed_app(telemetry_path: str):
"""Wrap prometheus_client's WSGI app so metrics are only served on the
configured path; everything else (e.g. "/") gets a small landing page."""
metrics_app = make_wsgi_app()
def app(environ, start_response):
if environ.get("PATH_INFO") == telemetry_path:
return metrics_app(environ, start_response)
if environ.get("PATH_INFO") == "/":
body = f'<html><body><h1>Shelly EM Exporter</h1><p><a href="{telemetry_path}">Metrics</a></p></body></html>'.encode()
start_response("200 OK", [("Content-Type", "text/html"), ("Content-Length", str(len(body)))])
return [body]
start_response("404 Not Found", [("Content-Type", "text/plain")])
return [b"404 Not Found"]
return app
def start_server(listen_host: str, listen_port: int, telemetry_path: str):
app = make_routed_app(telemetry_path)
server = make_server(listen_host, listen_port, app,
server_class=_ThreadingWSGIServer,
handler_class=_QuietWSGIRequestHandler)
t = threading.Thread(target=server.serve_forever, daemon=True)
t.start()
return server
def load_config(config_path: Path) -> dict:
if not config_path.exists():
raise FileNotFoundError(
f"Config file not found: {config_path}. "
f"Create it (see script docstring for format) or pass --config.file."
)
with config_path.open("r") as f:
config = yaml.safe_load(f) or {}
nodes = config.get("nodes") or []
if not nodes:
raise ValueError(f"No nodes defined in {config_path} (expected a top-level 'nodes' list)")
for node in nodes:
if "ip" not in node:
raise ValueError(f"Node missing required 'ip' field: {node}")
return config
def main() -> None:
default_config = Path(__file__).resolve().parent / "shellyem_exporter.yml"
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 idexporter (default: 0)multi-node)")
parser.add_argument("--port"config.file", dest="config_file", type=int,Path, default=9091,default_config,
help=f"Path to YAML config file (default: {default_config})")
parser.add_argument("--web.listen-address", dest="listen_address", default=":9091",
help="PortAddress to expose /metrics onon, HOST:PORT or :PORT (default: :9091)")
parser.add_argument("--interval"web.telemetry-path", type=float,dest="telemetry_path", default=5.0,"/metrics",
help="PollPath intervalunder inwhich secondsto expose metrics (default: 5)/metrics)")
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()
authconfig = Noneload_config(args.config_file)
listen_host, listen_port = parse_listen_address(args.listen_address)
telemetry_path = args.telemetry_path
if args.usernamenot andtelemetry_path.startswith("/"):
args.password:
authtelemetry_path = requests.auth.HTTPDigestAuth(args.username,"/" args.password)+ telemetry_path
default_interval = float(config.get("interval", 5.0))
default_timeout = float(config.get("timeout", 5.0))
nodes = config["nodes"]
log.info("starting exporter: shelly=%sd em_id=node(s) from %s -> :%s/metricss:%s%s", (every %.1fs)"len(nodes), args.shelly_ip,config_file, args.em_id,listen_host, args.port,listen_port, args.interval)telemetry_path)
start_http_server(args.port)start_server(listen_host, poll_loop(args.shelly_ip,listen_port, args.em_id,telemetry_path)
args.interval,threads args.timeout,= auth)[]
for node in nodes:
t = threading.Thread(
target=poll_node,
args=(node, default_interval, default_timeout),
daemon=True,
)
t.start()
threads.append(t)
# keep main thread alive
for t in threads:
t.join()
if __name__ == "__main__":
main()
