feat(heat): add autonomous wireless heat meter daemon and CLI manager
This commit is contained in:
Executable
+232
@@ -0,0 +1,232 @@
|
||||
#!/usr/bin/env python3
|
||||
"""
|
||||
sdr-heat-daemon: Autonomous Wireless Heating Meter Telemetry Collector
|
||||
Captures Wireless M-Bus Mode C/T packets on 868.95 MHz using RTL-SDR and rtl_433.
|
||||
Maintains SQLite database, JSONL archive, /dev/shm fast IPC, and SuperCollider OSC relay.
|
||||
|
||||
Part of The Observatoř environmental telemetry suite.
|
||||
"""
|
||||
|
||||
import os
|
||||
import sys
|
||||
import json
|
||||
import time
|
||||
import socket
|
||||
import struct
|
||||
import signal
|
||||
import sqlite3
|
||||
import subprocess
|
||||
from datetime import datetime
|
||||
|
||||
HOME = os.path.expanduser("~")
|
||||
DATA_DIR = os.path.join(HOME, "src", "observator", "data")
|
||||
DB_PATH = os.path.join(DATA_DIR, "heat_meters.db")
|
||||
JSONL_PATH = os.path.join(DATA_DIR, "heat_meters.jsonl")
|
||||
SHM_LATEST = "/dev/shm/latest_heat_meter.json"
|
||||
SHM_SUMMARY = "/dev/shm/heat_fleet_summary.json"
|
||||
|
||||
OSC_HOST = "127.0.0.1"
|
||||
OSC_PORT_SCLANG = 57120
|
||||
|
||||
os.makedirs(DATA_DIR, exist_ok=True)
|
||||
|
||||
# OSC UDP Socket
|
||||
sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
|
||||
|
||||
def make_osc_message(address, args):
|
||||
def pad4(b):
|
||||
return b + b"\x00" * ((4 - (len(b) % 4)) % 4)
|
||||
msg = pad4(address.encode("utf-8") + b"\x00")
|
||||
tags = ","
|
||||
payload = bytearray()
|
||||
for arg in args:
|
||||
if isinstance(arg, int):
|
||||
tags += "i"
|
||||
payload.extend(struct.pack(">i", arg))
|
||||
elif isinstance(arg, float):
|
||||
tags += "f"
|
||||
payload.extend(struct.pack(">f", arg))
|
||||
elif isinstance(arg, str):
|
||||
tags += "s"
|
||||
payload.extend(pad4(arg.encode("utf-8") + b"\x00"))
|
||||
msg += pad4(tags.encode("utf-8") + b"\x00")
|
||||
msg += payload
|
||||
return bytes(msg)
|
||||
|
||||
def init_db():
|
||||
conn = sqlite3.connect(DB_PATH)
|
||||
cur = conn.cursor()
|
||||
cur.execute("""
|
||||
CREATE TABLE IF NOT EXISTS meters (
|
||||
id INTEGER PRIMARY KEY,
|
||||
manufacturer TEXT,
|
||||
model TEXT,
|
||||
room_label TEXT DEFAULT '',
|
||||
first_seen TEXT,
|
||||
last_seen TEXT,
|
||||
last_hca REAL,
|
||||
prev_hca REAL,
|
||||
billing_date TEXT,
|
||||
total_packets INTEGER DEFAULT 0
|
||||
)
|
||||
""")
|
||||
cur.execute("""
|
||||
CREATE TABLE IF NOT EXISTS readings (
|
||||
id INTEGER,
|
||||
timestamp TEXT,
|
||||
hca_val REAL,
|
||||
hca_prev REAL,
|
||||
billing_date TEXT,
|
||||
raw_hex TEXT,
|
||||
FOREIGN KEY(id) REFERENCES meters(id)
|
||||
)
|
||||
""")
|
||||
cur.execute("CREATE INDEX IF NOT EXISTS idx_readings_id_time ON readings(id, timestamp)")
|
||||
conn.commit()
|
||||
conn.close()
|
||||
|
||||
def update_summary(conn):
|
||||
cur = conn.cursor()
|
||||
cur.execute("""
|
||||
SELECT id, manufacturer, room_label, last_seen, last_hca, prev_hca, billing_date, total_packets
|
||||
FROM meters ORDER BY last_seen DESC
|
||||
""")
|
||||
rows = cur.fetchall()
|
||||
meters_list = []
|
||||
total_consumption = 0.0
|
||||
for r in rows:
|
||||
val = r[4] or 0.0
|
||||
total_consumption += val
|
||||
meters_list.append({
|
||||
"id": r[0],
|
||||
"manufacturer": r[1],
|
||||
"room_label": r[2] if r[2] else f"Meter #{r[0]}",
|
||||
"last_seen": r[3],
|
||||
"last_hca": val,
|
||||
"prev_hca": r[5] or 0.0,
|
||||
"billing_date": r[6],
|
||||
"packets": r[7]
|
||||
})
|
||||
|
||||
summary = {
|
||||
"timestamp": datetime.now().isoformat(),
|
||||
"total_meters": len(meters_list),
|
||||
"total_units": round(total_consumption, 3),
|
||||
"meters": meters_list
|
||||
}
|
||||
tmp_path = SHM_SUMMARY + ".tmp"
|
||||
with open(tmp_path, "w") as f:
|
||||
json.dump(summary, f, indent=2)
|
||||
os.replace(tmp_path, SHM_SUMMARY)
|
||||
|
||||
def process_packet(pkt):
|
||||
meter_id = pkt.get("id")
|
||||
if not meter_id:
|
||||
return
|
||||
|
||||
mfg = pkt.get("M", "UNK")
|
||||
now_ts = pkt.get("time", datetime.now().strftime("%Y-%m-%d %H:%M:%S"))
|
||||
data_hex = pkt.get("data", "")
|
||||
billing_date = pkt.get("inst_date_17", "")
|
||||
|
||||
# Try parsing current consumption
|
||||
last_hca = 0.0
|
||||
try:
|
||||
last_hca = float(pkt.get("inst_hca_0", "0.0").strip())
|
||||
except (ValueError, TypeError):
|
||||
pass
|
||||
|
||||
# Try parsing previous billing period
|
||||
prev_hca = 0.0
|
||||
try:
|
||||
prev_hca = float(pkt.get("inst_hca_17", "0.0").strip())
|
||||
except (ValueError, TypeError):
|
||||
pass
|
||||
|
||||
# 1. Update SQLite DB
|
||||
conn = sqlite3.connect(DB_PATH)
|
||||
cur = conn.cursor()
|
||||
cur.execute("SELECT id, total_packets FROM meters WHERE id = ?", (meter_id,))
|
||||
row = cur.fetchone()
|
||||
if row:
|
||||
pkts = row[1] + 1
|
||||
cur.execute("""
|
||||
UPDATE meters
|
||||
SET last_seen = ?, last_hca = ?, prev_hca = ?, billing_date = ?, total_packets = ?
|
||||
WHERE id = ?
|
||||
""", (now_ts, last_hca, prev_hca, billing_date, pkts, meter_id))
|
||||
else:
|
||||
cur.execute("""
|
||||
INSERT INTO meters (id, manufacturer, model, first_seen, last_seen, last_hca, prev_hca, billing_date, total_packets)
|
||||
VALUES (?, ?, ?, ?, ?, ?, ?, ?, 1)
|
||||
""", (meter_id, mfg, pkt.get("type_string", "HCA"), now_ts, now_ts, last_hca, prev_hca, billing_date))
|
||||
|
||||
cur.execute("""
|
||||
INSERT INTO readings (id, timestamp, hca_val, hca_prev, billing_date, raw_hex)
|
||||
VALUES (?, ?, ?, ?, ?, ?)
|
||||
""", (meter_id, now_ts, last_hca, prev_hca, billing_date, data_hex))
|
||||
conn.commit()
|
||||
|
||||
# 2. Update SHM summary and latest
|
||||
update_summary(conn)
|
||||
conn.close()
|
||||
|
||||
tmp_latest = SHM_LATEST + ".tmp"
|
||||
with open(tmp_latest, "w") as f:
|
||||
json.dump(pkt, f, indent=2)
|
||||
os.replace(tmp_latest, SHM_LATEST)
|
||||
|
||||
# 3. Append to JSONL
|
||||
with open(JSONL_PATH, "a") as f:
|
||||
f.write(json.dumps(pkt) + "\n")
|
||||
|
||||
# 4. Relay to SuperCollider via OSC
|
||||
try:
|
||||
osc_data = make_osc_message("/observator/heat", [int(meter_id), float(last_hca), float(prev_hca), str(mfg)])
|
||||
sock.sendto(osc_data, (OSC_HOST, OSC_PORT_SCLANG))
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
print(f"[{now_ts}] Heat Meter #{meter_id} ({mfg}): current={last_hca} prev={prev_hca} (logged)")
|
||||
sys.stdout.flush()
|
||||
|
||||
def run_collector():
|
||||
init_db()
|
||||
print("[*] sdr-heat-daemon started. Listening on 868.95 MHz Mode C&T...")
|
||||
sys.stdout.flush()
|
||||
|
||||
cmd = [
|
||||
"rtl_433",
|
||||
"-f", "868.95M",
|
||||
"-s", "1200k",
|
||||
"-R", "104",
|
||||
"-g", "19.7",
|
||||
"-F", "json"
|
||||
]
|
||||
|
||||
while True:
|
||||
try:
|
||||
proc = subprocess.Popen(cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True, bufsize=1)
|
||||
for line in proc.stdout:
|
||||
line = line.strip()
|
||||
if not line:
|
||||
continue
|
||||
try:
|
||||
pkt = json.loads(line)
|
||||
process_packet(pkt)
|
||||
except json.JSONDecodeError:
|
||||
pass
|
||||
proc.wait()
|
||||
print("[-] rtl_433 exited with code", proc.returncode, "- restarting in 3s...")
|
||||
time.sleep(3)
|
||||
except Exception as e:
|
||||
print("[-] Collector exception:", e, "- retrying in 5s...")
|
||||
time.sleep(5)
|
||||
|
||||
if __name__ == "__main__":
|
||||
def sig_handler(sig, frame):
|
||||
print("\n[*] Stopping sdr-heat-daemon...")
|
||||
sys.exit(0)
|
||||
signal.signal(signal.SIGINT, sig_handler)
|
||||
signal.signal(signal.SIGTERM, sig_handler)
|
||||
run_collector()
|
||||
Reference in New Issue
Block a user