1
sudo apt install python3-venv
sudo mkdir -p /opt/shellyem_exporter
sudo python3 -m venv /opt/shellyem_exporter/venv
sudo /opt/shellyem_exporter/venv/bin/pip install requests prometheus_client
sudo vi /opt/shellyem_exporter/shellyem_exporter.py
#!/usr/bin/env python3
"""
Polls one or more Shelly Pro 3EM EM nodes-> Prometheus exporter (localmulti-target, on-demand)
Works like blackbox_exporter / snmp_exporter: the exporter itself takes
no fixed list of devices. Prometheus tells it which device to scrape via
query parameters on each scrape request, and the exporter fetches that
device's live RPC API)data synchronously and exposesreturns theirit readings on /metrics inas Prometheus text format. Every metric is
labeled with `shelly_ip` and `em_id` so multiple devices/nodes can be
told apart.metrics.
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
#Scrape request format:
GET /metrics?target=<shelly-ip>[:port]&id=<em_id>&username=<user>&password=<pass>
target required --config.file defaultsIP/hostname to(optionally "shellyem_exporter.yml"host:port") nextof tothe thisShelly scriptdevice
#id optional --web.listen-address defaultsEM tocomponent ":9091"id (alldefault: interfaces,0)
portusername 9091)
# use e.g.optional - RPC auth username, if the device has authentication enabled
password optional -web.listen-address= RPC auth password, if the device has authentication enabled
Example Prometheus scrape config:
scrape_configs:
- job_name: shellyem
metrics_path: /metrics
static_configs:
- targets:
- 192.168.1.50
- 192.168.1.51
params:
id: ["0"]
relabel_configs:
- source_labels: [__address__]
target_label: __param_target
- source_labels: [__param_target]
target_label: instance
- target_label: __address__
replacement: 127.0.0.1:9091 to bind a specific IP
# --web.telemetry-pathexporter's defaultsown 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>/metricsaddress
"""
import argparse
import logging
import threading
import time
from pathlib import Path
from socketserver import ThreadingMixIn
from urllib.parse import parse_qs
from wsgiref.simple_server import WSGIServer, WSGIRequestHandler, make_server
import requests
import yaml
from prometheus_client import make_wsgi_app,CollectorRegistry, GaugeGauge, generate_latest, CONTENT_TYPE_LATEST
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")
class _ThreadingWSGIServer(ThreadingMixIn, WSGIServer):
daemon_threads = True
class _QuietWSGIRequestHandler(WSGIRequestHandler):
def log_message(self, fmt, *args):
pass # suppress default access log; we log per-scrape ourselves
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(build_metrics(shelly_ip: str, em_id: int, data:timeout: dict)float, auth=None) -> None:tuple:
"""Fetch one device's data and return (registry, success: bool)."""
registry = CollectorRegistry()
labels_phase = ["shelly_ip", "em_id", "phase"]
labels_node = ["shelly_ip", "em_id"]
current = Gauge("shelly_em_current_amps", "Current in amps", labels_phase, registry=registry)
voltage = Gauge("shelly_em_voltage_volts", "Voltage in volts", labels_phase, registry=registry)
act_power = Gauge("shelly_em_active_power_watts", "Active power in watts", labels_phase, registry=registry)
aprt_power = Gauge("shelly_em_apparent_power_va", "Apparent power in VA", labels_phase, registry=registry)
power_factor = Gauge("shelly_em_power_factor", "Power factor", labels_phase, registry=registry)
frequency = Gauge("shelly_em_frequency_hz", "Line frequency in Hz", labels_phase, registry=registry)
neutral_current = Gauge("shelly_em_neutral_current_amps", "Neutral current in amps (0 if not measured)",
labels_node, registry=registry)
total_current = Gauge("shelly_em_total_current_amps", "Total current in amps", labels_node, registry=registry)
total_act_power = Gauge("shelly_em_total_active_power_watts", "Total active power in watts",
labels_node, registry=registry)
total_aprt_power = Gauge("shelly_em_total_apparent_power_va", "Total apparent power in VA",
labels_node, registry=registry)
scrape_success = Gauge("shelly_scrape_success", "1 if the last scrape of the device succeeded",
labels_node, registry=registry)
scrape_duration = Gauge("shelly_scrape_duration_seconds", "Time taken to fetch data from the device",
labels_node, registry=registry)
start = time.time()
success = True
try:
data = fetch_em_status(shelly_ip, em_id, timeout, auth)
for phase in PHASES:
CURRENT.current.labels(shelly_ip, em_id, phase).set(data.get(f"{phase}_current", 0) or 0)
VOLTAGE.voltage.labels(shelly_ip, em_id, phase).set(data.get(f"{phase}_voltage", 0) or 0)
ACT_POWER.act_power.labels(shelly_ip, em_id, phase).set(data.get(f"{phase}_act_power", 0) or 0)
APRT_POWER.aprt_power.labels(shelly_ip, em_id, phase).set(data.get(f"{phase}_aprt_power", 0) or 0)
POWER_FACTOR.power_factor.labels(shelly_ip, em_id, phase).set(data.get(f"{phase}_pf", 0) or 0)
FREQUENCY.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.neutral_current.labels(shelly_ip, em_id).set(data.get("n_current") or 0)
TOTAL_CURRENT.total_current.labels(shelly_ip, em_id).set(data.get("total_current", 0) or 0)
TOTAL_ACT_POWER.total_act_power.labels(shelly_ip, em_id).set(data.get("total_act_power", 0) or 0)
TOTAL_APRT_POWER.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,success label_id).set(0)= False
log.warning("scrape failed [%s id=%s]: %s", shelly_ip, em_id, exc)
finally:scrape_success.labels(shelly_ip, SCRAPE_DURATION.em_id).set(1 if success else 0)
scrape_duration.labels(label_ip,shelly_ip, label_id)em_id).set(time.time() - start)
time.sleep(interval)return registry, success
def make_app(telemetry_path: str, timeout: float):
def app(environ, start_response):
path = environ.get("PATH_INFO", "")
if path == telemetry_path:
params = parse_qs(environ.get("QUERY_STRING", ""))
target = (params.get("target") or [None])[0]
em_id_raw = (params.get("id") or ["0"])[0]
username = (params.get("username") or [None])[0]
password = (params.get("password") or [None])[0]
if not target:
body = b"Missing required 'target' query parameter, e.g. /metrics?target=192.168.1.50&id=0\n"
start_response("400 Bad Request", [("Content-Type", "text/plain"), ("Content-Length", str(len(body)))])
return [body]
try:
em_id = int(em_id_raw)
except ValueError:
body = f"Invalid 'id' query parameter: {em_id_raw!r}\n".encode()
start_response("400 Bad Request", [("Content-Type", "text/plain"), ("Content-Length", str(len(body)))])
return [body]
auth = requests.auth.HTTPDigestAuth(username, password) if username and password else None
registry, _ = build_metrics(target, em_id, timeout, auth)
output = generate_latest(registry)
start_response("200 OK", [("Content-Type", CONTENT_TYPE_LATEST), ("Content-Length", str(len(output)))])
return [output]
if path == "/":
body = (
"<html><body><h1>Shelly EM Exporter</h1>"
f'<p>Usage: <code>{telemetry_path}?target=<shelly-ip>&id=<em_id></code></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 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)str, timeout: float):
app = make_routed_app(telemetry_path)make_app(telemetry_path, timeout)
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 3EMEM Prometheus exporter (multi-node)")target, parser.add_argument("--config.file", dest="config_file", type=Path, default=default_config,
help=f"Path to YAML config file (default: {default_config})on-demand)")
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)")
parser.add_argument("--scrape-timeout", type=float, default=5.0,
help="HTTP request timeout when fetching a target's data (default: 5)")
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 %sexporter -> %s:%s%s"s?target=<ip>&id=<em_id>", len(nodes), args.config_file, listen_host, listen_port, telemetry_path)
server = start_server(listen_host, listen_port, telemetry_path)telemetry_path, threadsargs.scrape_timeout)
=try:
[]while forTrue:
nodetime.sleep(3600)
inexcept nodes:KeyboardInterrupt:
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(server.shutdown()
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 /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/venv/bin/python3 /opt/shellyem_exporter/shellyem_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"
sudo systemctl daemon-reload
sudo systemctl enable --now shellyem_exporter.service
