diff --git a/heat b/heat new file mode 100755 index 0000000..eea2e02 --- /dev/null +++ b/heat @@ -0,0 +1,161 @@ +#!/usr/bin/env python3 +""" +heat: Observatoř Wireless Heating Telemetry CLI +Manages, queries, and inspects 868.95 MHz Wireless M-Bus radiator metrics. +""" + +import os +import sys +import json +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" +SERVICE_NAME = "sdr-heat.service" + +def cmd_status(): + if not os.path.exists(DB_PATH): + print("[-] Žádná databáze měřičů topení zatím neexistuje.") + print(" Spusťte sběrač pomocí: heat start") + return + + conn = sqlite3.connect(DB_PATH) + 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() + conn.close() + + print("╔═══════════════════════════════════════════════════════════════════════════════════════════════════════╗") + print("║ THE OBSERVATOŘ // RADIÁTORY & INDIKÁTORY TOPENÍ (868.95 MHz Wireless M-Bus Mode C&T) ║") + print("╚═══════════════════════════════════════════════════════════════════════════════════════════════════════╝") + + # Check daemon status + is_active = subprocess.run(["systemctl", "--user", "is-active", "--quiet", SERVICE_NAME]).returncode == 0 + status_str = "\033[32m● AKTIVNÍ (démon běží)\033[0m" if is_active else "\033[31m○ NEAKTIVNÍ\033[0m" + print(f" Služba sdr-heat: {status_str} | Tuner: RTL-SDR Fitipower FC0013 (t520) | Zisk: 19.7 dB") + print("─" * 105) + + if not rows: + print(" [!] V databázi zatím není zaznamenán žádný indikátor. Vyčkejte na vysílací interval.") + return + + header = f" {'ID MĚŘIČE':<12} │ {'VÝROBCE':<8} │ {'MÍSTNOST / ŠTÍTEK':<26} │ {'DÍLKY':>8} │ {'MINULE':>8} │ {'POSLEDNÍ SIGNÁL':<19} │ {'PKT':>5}" + print(header) + print("═" * 105) + + total_units = 0.0 + for r in rows: + mid, mfg, label, last_seen, hca, prev_hca, billing_date, pkts = r + total_units += (hca or 0.0) + lbl = label if label else "—" + hca_str = f"{hca:.1f}" if hca is not None else "0.0" + prev_str = f"{prev_hca:.1f}" if prev_hca is not None else "0.0" + print(f" {mid:<12} │ {mfg:<8} │ {lbl:<26} │ {hca_str:>8} │ {prev_str:>8} │ {last_seen:<19} │ {pkts:>5}") + + print("─" * 105) + print(f" Celkem indikátorů: {len(rows)} │ Celková naměřená spotřeba: {total_units:.1f} dílků") + print(" Tip: Pro pojmenování radiátoru použijte: heat label \"Kuchyň okno\"") + print() + +def cmd_label(args): + if len(args) < 2: + print("Použití: heat label ") + print("Příklad: heat label 18466419 \"Kuchyň velký radiátor\"") + return + + meter_id = args[0] + label = " ".join(args[1:]) + + conn = sqlite3.connect(DB_PATH) + cur = conn.cursor() + cur.execute("UPDATE meters SET room_label = ? WHERE id = ?", (label, meter_id)) + if cur.rowcount > 0: + conn.commit() + print(f"[✓] Indikátor #{meter_id} úspěšně označen jako: '{label}'") + else: + # Insert placeholder if not seen yet + now_ts = datetime.now().strftime("%Y-%m-%d %H:%M:%S") + cur.execute(""" + INSERT INTO meters (id, manufacturer, model, room_label, first_seen, last_seen, last_hca, prev_hca, billing_date, total_packets) + VALUES (?, 'UNK', 'HCA', ?, ?, ?, 0.0, 0.0, '', 0) + """, (meter_id, label, now_ts, now_ts)) + conn.commit() + print(f"[✓] Vytvořen nový záznam pro #{meter_id} s označením: '{label}'") + conn.close() + +def cmd_live(): + print("[*] Spouštím živý odposlech paketů radiátorů na 868.95 MHz (Ctrl+C pro ukončení)...") + cmd = ["rtl_433", "-f", "868.95M", "-s", "1200k", "-R", "104", "-g", "19.7", "-F", "json"] + proc = subprocess.Popen(cmd, stdout=subprocess.PIPE, stderr=subprocess.DEVNULL, text=True) + try: + for line in proc.stdout: + line = line.strip() + if not line: + continue + try: + pkt = json.loads(line) + mid = pkt.get("id", "UNK") + mfg = pkt.get("M", "UNK") + hca = pkt.get("inst_hca_0", "0.0").strip() + t_str = pkt.get("time", "") + data_hex = pkt.get("data", "") + print(f"[{t_str}] \033[36m#{mid}\033[0m ({mfg}) ➔ dílky: \033[32m{hca}\033[0m │ raw: {data_hex[:24]}...") + except json.JSONDecodeError: + pass + except KeyboardInterrupt: + proc.terminate() + print("\n[*] Živý odposlech ukončen.") + +def cmd_log(n=10): + if not os.path.exists(JSONL_PATH): + print("[-] Žádný archiv paketů zatím neexistuje.") + return + cmd = ["tail", "-n", str(n), JSONL_PATH] + subprocess.run(cmd) + +def cmd_service(action): + if action in ["start", "stop", "restart", "status", "enable", "disable"]: + subprocess.run(["systemctl", "--user", action, SERVICE_NAME]) + else: + print(f"Neznámá akce služby: {action}") + +def main(): + args = sys.argv[1:] + if not args or args[0] in ["status", "s"]: + cmd_status() + elif args[0] in ["label", "l"]: + cmd_label(args[1:]) + elif args[0] in ["live", "watch", "w"]: + cmd_live() + elif args[0] in ["log", "tail"]: + n = int(args[1]) if len(args) > 1 else 10 + cmd_log(n) + elif args[0] in ["json", "summary"]: + if os.path.exists(SHM_SUMMARY): + with open(SHM_SUMMARY) as f: + print(f.read()) + else: + print("{}") + elif args[0] in ["start", "stop", "restart", "enable", "disable"]: + cmd_service(args[0]) + else: + print("Observatoř Heat Telemetry CLI") + print("Použití:") + print(" heat status Zobrazí přehled všech zachycených radiátorů") + print(" heat live Živý streaming příchozích paketů v terminálu") + print(" heat label Pojmenuje radiátor (např. 'Kuchyň')") + print(" heat log [N] Vypíše posledních N surových JSON paketů") + print(" heat json Vypíše agregovaný JSON z /dev/shm") + print(" heat start|stop|restart Ovládání systemd služby") + +if __name__ == "__main__": + main() diff --git a/sdr-heat-daemon b/sdr-heat-daemon new file mode 100755 index 0000000..ffa384b --- /dev/null +++ b/sdr-heat-daemon @@ -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()