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 one or more Shelly Pro 3EM EM nodes (local RPC API) and exposes
their 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 shelly_exporter.py --config.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
- ip: 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>:<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 make_wsgi_app, Gauge
logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s")
log = logging.getLogger("shelly_exporter")
LABELS_PHASE = ["shelly_ip", "em_id", "phase"]
LABELS_NODE = ["shelly_ip", "em_id"]
# Metric definitions. Each has shelly_ip/em_id labels, phase metrics also have "phase".
CURRENT = Gauge("shelly_em_current_amps", "Current in amps", LABELS_PHASE)
VOLTAGE = Gauge("shelly_em_voltage_volts", "Voltage in volts", LABELS_PHASE)
ACT_POWER = Gauge("shelly_em_active_power_watts", "Active power in watts", LABELS_PHASE)
APRT_POWER = Gauge("shelly_em_apparent_power_va", "Apparent power in VA", LABELS_PHASE)
POWER_FACTOR = Gauge("shelly_em_power_factor", "Power factor", LABELS_PHASE)
FREQUENCY = Gauge("shelly_em_frequency_hz", "Line frequency in Hz", LABELS_PHASE)
NEUTRAL_CURRENT = Gauge("shelly_em_neutral_current_amps", "Neutral current in amps (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 node succeeded", LABELS_NODE)
SCRAPE_DURATION = Gauge("shelly_scrape_duration_seconds", "Time taken to fetch data from the node", 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(shelly_ip, em_id, phase).set(data.get(f"{phase}_current", 0) or 0)
VOLTAGE.labels(shelly_ip, em_id, phase).set(data.get(f"{phase}_voltage", 0) or 0)
ACT_POWER.labels(shelly_ip, em_id, phase).set(data.get(f"{phase}_act_power", 0) or 0)
APRT_POWER.labels(shelly_ip, em_id, phase).set(data.get(f"{phase}_aprt_power", 0) or 0)
POWER_FACTOR.labels(shelly_ip, em_id, phase).set(data.get(f"{phase}_pf", 0) or 0)
FREQUENCY.labels(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_node(node: dict, default_interval: float, default_timeout: 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 [%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 [%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 (multi-node)")
parser.add_argument("--config.file", dest="config_file", type=Path, default=default_config,
help=f"Path to YAML config file (default: {default_config})")
parser.add_argument("--web.listen-address", dest="listen_address", default=":9091",
help="Address to expose metrics on, HOST:PORT or :PORT (default: :9091)")
parser.add_argument("--web.telemetry-path", dest="telemetry_path", default="/metrics",
help="Path under which to expose metrics (default: /metrics)")
args = parser.parse_args()
config = load_config(args.config_file)
listen_host, listen_port = parse_listen_address(args.listen_address)
telemetry_path = args.telemetry_path
if not telemetry_path.startswith("/"):
telemetry_path = "/" + 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: %d node(s) from %s -> %s:%s%s", len(nodes), args.config_file, listen_host, listen_port, telemetry_path)
start_server(listen_host, listen_port, telemetry_path)
threads = []
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()
sudo groupadd --system shellyem_exporter
sudo useradd --system --no-create-home --shell /usr/sbin/nologin --gid shellyem_exporter shellyem_exporter
sudo vi /opt/shellyem/shellyem_exporter.yml
sudo chgrp shellyem_exporter /opt/shellyem/shellyem_exporter.yml
sudo chmod 0640 /opt/shellyem/shellyem_exporter.yml
sudo vi /etc/systemd/system/shellyem_exporter.service
[Unit]
Description=Shelly EM Prometheus Exporter
After=network-online.target
Wants=network-online.target
[Service]
Type=simple
User=shellyem_exporter
Group=shellyem_exporter
EnvironmentFile=-/etc/default/shellyem_exporter
ExecStart=/opt/shellyem_exporter/shelly_exporter.py $SHELLYEM_OPTS
WorkingDirectory=/opt/shellyem_exporter
Restart=on-failure
RestartSec=5
# Hardening
NoNewPrivileges=true
ProtectSystem=strict
ProtectHome=true
PrivateTmp=true
ReadWritePaths=-/opt/shellyem_exporter
ProtectKernelTunables=true
ProtectKernelModules=true
ProtectControlGroups=true
RestrictSUIDSGID=true
LockPersonality=true
[Install]
WantedBy=multi-user.target
sudo vi /etc/default/shellyem_exporter
# Options passed to shelly_exporter.py, e.g.:
SHELLYEM_OPTS="--config.file=/opt/shellyem_exporter/shellyem_exporter.yml --web.listen-address=:9091 --web.telemetry-path=/metrics"
