bin/sdr-heat-daemon

#!/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()