384 lines
15 KiB
Python
384 lines
15 KiB
Python
#!/usr/bin/env python3
|
|
import json
|
|
import logging
|
|
import math
|
|
import os
|
|
import queue
|
|
import re
|
|
import signal
|
|
import threading
|
|
import time
|
|
import urllib.error
|
|
import urllib.parse
|
|
import urllib.request
|
|
from collections import Counter
|
|
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
|
|
|
|
import paho.mqtt.client as mqtt
|
|
|
|
logging.basicConfig(level=os.getenv("LOG_LEVEL", "INFO"), format="%(asctime)s %(levelname)s %(message)s")
|
|
LOG = logging.getLogger("home-monitor")
|
|
|
|
MQTT_HOST = os.getenv("MQTT_HOST", "10.10.1.1")
|
|
MQTT_PORT = int(os.getenv("MQTT_PORT", "1883"))
|
|
MQTT_TOPICS = [x.strip() for x in os.getenv("MQTT_TOPICS", "#,$SYS/#").split(",") if x.strip()]
|
|
MQTT_USERNAME = os.getenv("MQTT_USERNAME", "")
|
|
MQTT_PASSWORD = os.getenv("MQTT_PASSWORD", "")
|
|
INFLUX_URL = os.getenv("INFLUX_URL", "http://influxdb:8086").rstrip("/")
|
|
INFLUX_ORG = os.getenv("INFLUX_ORG", "influx.sascha-lutz.de")
|
|
INFLUX_BUCKET = os.getenv("INFLUX_BUCKET", "telegraf")
|
|
INFLUX_TOKEN = os.getenv("INFLUX_TOKEN", "")
|
|
BUTLER_URL = os.getenv("BUTLER_URL", "http://10.4.1.116:8888").rstrip("/")
|
|
BUTLER_TOKEN = os.getenv("BUTLER_TOKEN", "")
|
|
WRITE_INTERVAL = int(os.getenv("WRITE_INTERVAL", "15"))
|
|
TOPIC_STATS_INTERVAL = int(os.getenv("TOPIC_STATS_INTERVAL", "300"))
|
|
HA_POLL_INTERVAL = int(os.getenv("HA_POLL_INTERVAL", "60"))
|
|
MAX_TOPICS = int(os.getenv("MAX_TOPICS", "500"))
|
|
HEALTH_PORT = int(os.getenv("HEALTH_PORT", "9199"))
|
|
|
|
SENSITIVE = re.compile(r"(^|[_.\-/])(token|secret|password|pass|credential|api[_-]?key|latitude|longitude|lat|lon|gps|location|ssid|bssid|mac|ip|user|person)([_.\-/]|$)", re.I)
|
|
BLOCKED_TOPIC = re.compile(r"(^|/)(camera|image|snapshot|video|audio|recording|clip)(/|$)", re.I)
|
|
BOOLS = {
|
|
"on": 1.0, "off": 0.0, "true": 1.0, "false": 0.0,
|
|
"online": 1.0, "offline": 0.0, "open": 1.0, "closed": 0.0,
|
|
"home": 1.0, "not_home": 0.0, "locked": 1.0, "unlocked": 0.0,
|
|
"available": 1.0, "unavailable": 0.0,
|
|
}
|
|
SAFE_HA_DOMAINS = {
|
|
"sensor", "binary_sensor", "switch", "light", "climate", "cover", "lock",
|
|
"fan", "humidifier", "input_number", "number", "water_heater", "sun",
|
|
}
|
|
|
|
stop_event = threading.Event()
|
|
write_queue: queue.Queue[str] = queue.Queue(maxsize=20000)
|
|
lock = threading.Lock()
|
|
state = {
|
|
"started": time.time(), "mqtt_up": 0, "messages_total": 0, "parse_errors": 0,
|
|
"numeric_samples": 0, "topics": {}, "roots": Counter(), "last_message": 0.0,
|
|
"influx_up": 0, "influx_writes": 0, "influx_errors": 0,
|
|
"ha_up": 0, "ha_http_status": 0, "ha_entities": 0, "ha_numeric": 0,
|
|
"ha_latency_ms": 0.0, "last_ha_poll": 0.0,
|
|
}
|
|
|
|
|
|
def esc_measurement(value):
|
|
return str(value).replace("\\", "\\\\").replace(",", "\\,").replace(" ", "\\ ")
|
|
|
|
|
|
def esc_tag(value):
|
|
return str(value).replace("\\", "\\\\").replace(",", "\\,").replace(" ", "\\ ").replace("=", "\\=")
|
|
|
|
|
|
def esc_field_key(value):
|
|
return esc_tag(value)
|
|
|
|
|
|
def field_value(value):
|
|
if isinstance(value, bool):
|
|
return "true" if value else "false"
|
|
if isinstance(value, int):
|
|
return f"{value}i"
|
|
if isinstance(value, float):
|
|
if not math.isfinite(value):
|
|
raise ValueError("non-finite float")
|
|
return repr(value)
|
|
text = str(value).replace("\\", "\\\\").replace('"', '\\"')
|
|
return f'"{text}"'
|
|
|
|
|
|
def line(measurement, tags, fields, timestamp=None):
|
|
tag_text = "".join(f",{esc_tag(k)}={esc_tag(v)}" for k, v in sorted(tags.items()) if str(v) != "")
|
|
field_text = ",".join(f"{esc_field_key(k)}={field_value(v)}" for k, v in sorted(fields.items()))
|
|
stamp = f" {int(timestamp)}" if timestamp is not None else ""
|
|
return f"{esc_measurement(measurement)}{tag_text} {field_text}{stamp}"
|
|
|
|
|
|
def safe_key(key):
|
|
key = re.sub(r"[^A-Za-z0-9_.-]+", "_", str(key)).strip("_.-")[:100]
|
|
return key and not SENSITIVE.search(key)
|
|
|
|
|
|
def flatten(obj, prefix="", depth=0):
|
|
out = {}
|
|
if depth > 2 or not isinstance(obj, dict):
|
|
return out
|
|
for raw_key, value in obj.items():
|
|
key = str(raw_key)
|
|
full = f"{prefix}.{key}" if prefix else key
|
|
if not safe_key(full):
|
|
continue
|
|
if isinstance(value, bool):
|
|
out[full] = 1.0 if value else 0.0
|
|
elif isinstance(value, (int, float)) and not isinstance(value, bool) and math.isfinite(float(value)):
|
|
out[full] = float(value)
|
|
elif isinstance(value, str) and value.strip().lower() in BOOLS:
|
|
out[full] = BOOLS[value.strip().lower()]
|
|
elif isinstance(value, dict):
|
|
out.update(flatten(value, full, depth + 1))
|
|
return out
|
|
|
|
|
|
def extract_values(payload):
|
|
if len(payload) > 65536:
|
|
return {}
|
|
try:
|
|
text = payload.decode("utf-8").strip()
|
|
except UnicodeDecodeError:
|
|
return {}
|
|
if not text:
|
|
return {}
|
|
low = text.lower()
|
|
if low in BOOLS:
|
|
return {"value": BOOLS[low]}
|
|
try:
|
|
value = float(text)
|
|
if math.isfinite(value):
|
|
return {"value": value}
|
|
except ValueError:
|
|
pass
|
|
if text[:1] == "{" and text[-1:] == "}":
|
|
try:
|
|
return flatten(json.loads(text))
|
|
except (json.JSONDecodeError, TypeError, ValueError):
|
|
return {}
|
|
return {}
|
|
|
|
|
|
def topic_allowed(topic):
|
|
if not topic or len(topic) > 250 or SENSITIVE.search(topic) or BLOCKED_TOPIC.search(topic):
|
|
return False
|
|
# Discovery documents contain names, identifiers and configuration, not time-series values.
|
|
if topic.startswith("homeassistant/") and topic.endswith("/config"):
|
|
return False
|
|
return True
|
|
|
|
|
|
def enqueue(metric_line):
|
|
try:
|
|
write_queue.put_nowait(metric_line)
|
|
except queue.Full:
|
|
with lock:
|
|
state["influx_errors"] += 1
|
|
|
|
|
|
def on_connect(client, userdata, flags, reason_code, properties=None):
|
|
ok = int(reason_code) == 0
|
|
with lock:
|
|
state["mqtt_up"] = 1 if ok else 0
|
|
if ok:
|
|
for topic in MQTT_TOPICS:
|
|
client.subscribe(topic, qos=0)
|
|
LOG.info("MQTT connected; %d read-only subscriptions active", len(MQTT_TOPICS))
|
|
else:
|
|
LOG.warning("MQTT connection rejected with code %s", reason_code)
|
|
|
|
|
|
def on_disconnect(client, userdata, disconnect_flags, reason_code, properties=None):
|
|
with lock:
|
|
state["mqtt_up"] = 0
|
|
if not stop_event.is_set():
|
|
LOG.warning("MQTT disconnected with code %s", reason_code)
|
|
|
|
|
|
def on_message(client, userdata, msg):
|
|
now = int(time.time())
|
|
topic = msg.topic
|
|
values = extract_values(msg.payload) if topic_allowed(topic) else {}
|
|
root = topic.split("/", 1)[0][:80]
|
|
with lock:
|
|
state["messages_total"] += 1
|
|
state["last_message"] = now
|
|
state["roots"][root] += 1
|
|
topics = state["topics"]
|
|
if topic in topics or len(topics) < MAX_TOPICS:
|
|
row = topics.setdefault(topic, {"messages": 0, "last_seen": 0, "retained": 0, "numeric": 0})
|
|
row["messages"] += 1
|
|
row["last_seen"] = now
|
|
row["retained"] = 1 if msg.retain else row["retained"]
|
|
row["numeric"] += len(values)
|
|
state["numeric_samples"] += len(values)
|
|
for field, value in values.items():
|
|
enqueue(line("mqtt_value", {"topic": topic, "root": root, "field": field}, {"value": value}, now))
|
|
if topic.startswith("$SYS/broker/") and values:
|
|
metric = topic.removeprefix("$SYS/broker/").replace("/", "_")
|
|
for value in values.values():
|
|
enqueue(line("mqtt_broker", {"metric": metric}, {"value": value}, now))
|
|
break
|
|
|
|
|
|
def post_influx(lines):
|
|
if not lines or not INFLUX_TOKEN:
|
|
return False
|
|
query = urllib.parse.urlencode({"org": INFLUX_ORG, "bucket": INFLUX_BUCKET, "precision": "s"})
|
|
req = urllib.request.Request(
|
|
f"{INFLUX_URL}/api/v2/write?{query}", data=("\n".join(lines)).encode(), method="POST",
|
|
headers={"Authorization": f"Token {INFLUX_TOKEN}", "Content-Type": "text/plain; charset=utf-8"},
|
|
)
|
|
try:
|
|
with urllib.request.urlopen(req, timeout=10) as response:
|
|
ok = response.status in (204, 200)
|
|
except (urllib.error.URLError, TimeoutError, OSError) as exc:
|
|
LOG.warning("InfluxDB write failed: %s", type(exc).__name__)
|
|
ok = False
|
|
with lock:
|
|
state["influx_up"] = 1 if ok else 0
|
|
state["influx_writes" if ok else "influx_errors"] += 1
|
|
return ok
|
|
|
|
|
|
def snapshot_metrics(include_topics=False):
|
|
now = int(time.time())
|
|
with lock:
|
|
s = dict(state)
|
|
roots = dict(state["roots"])
|
|
topics = {k: dict(v) for k, v in state["topics"].items()}
|
|
age = now - int(s["last_message"]) if s["last_message"] else -1
|
|
lines = [line("home_monitoring", {"source": "home-monitor"}, {
|
|
"mqtt_up": int(s["mqtt_up"]), "influx_up": int(s["influx_up"]),
|
|
"messages_total": int(s["messages_total"]), "topics_seen": len(topics),
|
|
"numeric_samples_total": int(s["numeric_samples"]), "parse_errors_total": int(s["parse_errors"]),
|
|
"last_message_age_seconds": age, "influx_writes_total": int(s["influx_writes"]),
|
|
"influx_errors_total": int(s["influx_errors"]), "homeassistant_up": int(s["ha_up"]),
|
|
"homeassistant_http_status": int(s["ha_http_status"]), "homeassistant_entities": int(s["ha_entities"]),
|
|
"homeassistant_numeric_entities": int(s["ha_numeric"]), "homeassistant_latency_ms": float(s["ha_latency_ms"]),
|
|
}, now)]
|
|
for root, count in roots.items():
|
|
lines.append(line("mqtt_root", {"root": root}, {"messages_total": int(count)}, now))
|
|
if include_topics:
|
|
for topic, row in topics.items():
|
|
lines.append(line("mqtt_topic", {"topic": topic, "root": topic.split('/', 1)[0]}, {
|
|
"messages_total": int(row["messages"]), "last_seen": int(row["last_seen"]),
|
|
"retained": int(row["retained"]), "numeric_samples_total": int(row["numeric"]),
|
|
}, now))
|
|
return lines
|
|
|
|
|
|
def poll_homeassistant():
|
|
started = time.monotonic()
|
|
url = f"{BUTLER_URL}/homeassistant/api/states"
|
|
req = urllib.request.Request(url, headers={"Authorization": f"Bearer {BUTLER_TOKEN}"})
|
|
status = 0
|
|
entities = []
|
|
try:
|
|
with urllib.request.urlopen(req, timeout=15) as response:
|
|
status = response.status
|
|
if status == 200:
|
|
entities = json.load(response)
|
|
except urllib.error.HTTPError as exc:
|
|
status = exc.code
|
|
except (urllib.error.URLError, TimeoutError, OSError):
|
|
status = 0
|
|
now = int(time.time())
|
|
numeric = 0
|
|
if status == 200 and isinstance(entities, list):
|
|
for entity in entities:
|
|
entity_id = str(entity.get("entity_id", ""))
|
|
if "." not in entity_id:
|
|
continue
|
|
domain = entity_id.split(".", 1)[0]
|
|
if domain not in SAFE_HA_DOMAINS or SENSITIVE.search(entity_id):
|
|
continue
|
|
raw = str(entity.get("state", "")).strip()
|
|
if raw.lower() in BOOLS:
|
|
value = BOOLS[raw.lower()]
|
|
else:
|
|
try:
|
|
value = float(raw)
|
|
if not math.isfinite(value):
|
|
continue
|
|
except ValueError:
|
|
continue
|
|
attrs = entity.get("attributes") or {}
|
|
unit = str(attrs.get("unit_of_measurement", ""))[:30]
|
|
device_class = str(attrs.get("device_class", ""))[:40]
|
|
enqueue(line("homeassistant_state", {
|
|
"entity_id": entity_id, "domain": domain, "unit": unit, "device_class": device_class,
|
|
}, {"value": value}, now))
|
|
numeric += 1
|
|
latency = (time.monotonic() - started) * 1000.0
|
|
with lock:
|
|
state["ha_up"] = 1 if status == 200 else 0
|
|
state["ha_http_status"] = status
|
|
state["ha_entities"] = len(entities)
|
|
state["ha_numeric"] = numeric
|
|
state["ha_latency_ms"] = latency
|
|
state["last_ha_poll"] = now
|
|
|
|
|
|
def worker():
|
|
last_status = 0.0
|
|
last_topics = 0.0
|
|
last_ha = 0.0
|
|
batch = []
|
|
while not stop_event.wait(1):
|
|
now = time.monotonic()
|
|
while len(batch) < 5000:
|
|
try:
|
|
batch.append(write_queue.get_nowait())
|
|
except queue.Empty:
|
|
break
|
|
if now - last_ha >= HA_POLL_INTERVAL:
|
|
poll_homeassistant()
|
|
last_ha = now
|
|
include_topics = now - last_topics >= TOPIC_STATS_INTERVAL
|
|
if include_topics:
|
|
last_topics = now
|
|
if now - last_status >= WRITE_INTERVAL:
|
|
batch.extend(snapshot_metrics(include_topics=include_topics))
|
|
post_influx(batch)
|
|
batch.clear()
|
|
last_status = now
|
|
|
|
|
|
class HealthHandler(BaseHTTPRequestHandler):
|
|
def do_GET(self):
|
|
if self.path not in ("/health", "/"):
|
|
self.send_error(404)
|
|
return
|
|
with lock:
|
|
payload = {
|
|
"status": "ok" if state["influx_up"] else "degraded",
|
|
"mqtt_up": bool(state["mqtt_up"]), "influx_up": bool(state["influx_up"]),
|
|
"homeassistant_up": bool(state["ha_up"]), "homeassistant_http_status": state["ha_http_status"],
|
|
"topics_seen": len(state["topics"]), "messages_total": state["messages_total"],
|
|
"numeric_samples_total": state["numeric_samples"],
|
|
"uptime_seconds": int(time.time() - state["started"]),
|
|
}
|
|
body = json.dumps(payload, separators=(",", ":")).encode()
|
|
self.send_response(200 if payload["influx_up"] else 503)
|
|
self.send_header("Content-Type", "application/json")
|
|
self.send_header("Content-Length", str(len(body)))
|
|
self.end_headers()
|
|
self.wfile.write(body)
|
|
|
|
def log_message(self, fmt, *args):
|
|
return
|
|
|
|
|
|
def main():
|
|
if not INFLUX_TOKEN or not BUTLER_TOKEN:
|
|
raise SystemExit("INFLUX_TOKEN and BUTLER_TOKEN are required")
|
|
threading.Thread(target=worker, name="writer", daemon=True).start()
|
|
server = ThreadingHTTPServer(("0.0.0.0", HEALTH_PORT), HealthHandler)
|
|
threading.Thread(target=server.serve_forever, name="health", daemon=True).start()
|
|
client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id="pfannkuchen-home-monitor", clean_session=True)
|
|
if MQTT_USERNAME:
|
|
client.username_pw_set(MQTT_USERNAME, MQTT_PASSWORD)
|
|
client.on_connect = on_connect
|
|
client.on_disconnect = on_disconnect
|
|
client.on_message = on_message
|
|
client.reconnect_delay_set(min_delay=2, max_delay=60)
|
|
client.connect_async(MQTT_HOST, MQTT_PORT, keepalive=60)
|
|
for sig in (signal.SIGTERM, signal.SIGINT):
|
|
signal.signal(sig, lambda *_: stop_event.set())
|
|
client.loop_start()
|
|
LOG.info("Home monitor started: MQTT read-only, raw payload persistence disabled")
|
|
while not stop_event.wait(1):
|
|
pass
|
|
client.disconnect()
|
|
client.loop_stop()
|
|
server.shutdown()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|