wmbusmeters JSON output — parsing og feltmapping i Python

wmbusmeters konverterer rå wM-Bus telegrammer til struktureret JSON. Her er det komplette format og Python-integration.

---

JSON output format

// Eksempel: Techem HCA telegram (T1 mode)
{
  "media": "heat cost allocator",
  "meter": "techem_hca",
  "name": "hca_stue_lej1",
  "id": "12345678",
  "status": "OK",
  "current_hca": 1247,
  "previous_hca": 4821,
  "current_date": "2026-04-30",
  "previous_date": "2025-04-30",
  "rssi_dbm": -78,
  "timestamp": "2026-05-24T06:00:00Z"
}

// Kamstrup Multical 302 (varme, C1 mode)
{
  "media": "heat",
  "meter": "multical302",
  "name": "varmemaaler_kaelder",
  "id": "87654321",
  "status": "OK",
  "total_energy_consumption_kwh": 52483.2,
  "volume_m3": 2847.1,
  "power_kw": 12.4,
  "flow_temperature_c": 72.3,
  "return_temperature_c": 41.8,
  "rssi_dbm": -65,
  "timestamp": "2026-05-24T06:00:01Z"
}

---

Konfigurationsfil pr. måler

# /etc/wmbusmeters.d/hca_lej1.conf
name=hca_stue_lej1
id=12345678
driver=techem_hca
key=A1B2C3D4E5F6708192A3B4C5D6E7F800

# /etc/wmbusmeters.d/heat_main.conf
name=varmemaaler_kaelder
id=87654321
driver=multical302
# Ingen key — ikke krypteret

---

Python parsing af wmbusmeters output

# gateway/src/wmbus/parser.py

import json
import subprocess
from dataclasses import dataclass
from datetime import datetime
from typing import Optional

@dataclass
class WMBusReading:
    meter_id: str
    driver: str
    name: str
    timestamp: datetime
    rssi_dbm: Optional[float]
    status: str
    # Type-specifikke felter
    current_hca: Optional[float] = None
    total_energy_kwh: Optional[float] = None
    volume_m3: Optional[float] = None
    flow_temp_c: Optional[float] = None
    return_temp_c: Optional[float] = None
    power_kw: Optional[float] = None
    battery_level_pct: Optional[float] = None

def parse_wmbus_json(raw: str) -> Optional[WMBusReading]:
    """Parse ét JSON-objekt fra wmbusmeters stdout."""
    try:
        data = json.loads(raw.strip())
    except json.JSONDecodeError:
        return None

    # wmbusmeters sætter status="OK" eller fejlkode
    status = data.get("status", "UNKNOWN")

    return WMBusReading(
        meter_id=data["id"],
        driver=data.get("meter", "unknown"),
        name=data.get("name", data["id"]),
        timestamp=datetime.fromisoformat(data["timestamp"].replace("Z", "+00:00")),
        rssi_dbm=data.get("rssi_dbm"),
        status=status,
        current_hca=data.get("current_hca"),
        total_energy_kwh=data.get("total_energy_consumption_kwh"),
        volume_m3=data.get("volume_m3"),
        flow_temp_c=data.get("flow_temperature_c"),
        return_temp_c=data.get("return_temperature_c"),
        power_kw=data.get("power_kw"),
        battery_level_pct=_estimate_battery(data),
    )

def _estimate_battery(data: dict) -> Optional[float]:
    """wmbusmeters leverer battery_v eller battery_remaining_days afhængig af driver."""
    if "remaining_battery_days" in data:
        days = data["remaining_battery_days"]
        return min(100.0, round(days / 3650 * 100, 1))  # antag 10 år = 100%
    if "battery_v" in data:
        v = data["battery_v"]
        # Typisk HCA batteri: 3.6V = 100%, 2.8V = 0%
        return min(100.0, max(0.0, round((v - 2.8) / (3.6 - 2.8) * 100, 1)))
    return None

---

AES-dekrypteringsfejl

// AES-fejl: forkert nøgle eller manglende nøgle
{
  "media": "heat cost allocator",
  "meter": "techem_hca",
  "name": "hca_stue_lej2",
  "id": "AABBCCDD",
  "status": "DEC_ERR",
  "rssi_dbm": -72,
  "timestamp": "2026-05-24T06:00:02Z"
}

// status-koder:
// "OK"       → telegram modtaget og dekrypteret korrekt
// "DEC_ERR"  → AES-dekrypteringsfejl (forkert nøgle)
// "STUB"     → driver ikke implementeret for denne fabrikant/type
// "ERR"      → generel fejl (checksumfejl, trunkeret telegram)

def is_decryption_error(reading: WMBusReading) -> bool:
    return reading.status == "DEC_ERR"

def is_valid_reading(reading: WMBusReading) -> bool:
    return reading.status == "OK" and (
        reading.current_hca is not None or
        reading.total_energy_kwh is not None
    )

---

wmbusmeters subprocess integration

# gateway/src/wmbus/listener.py

import asyncio
import subprocess
from pathlib import Path

WMBUS_DEVICE = "/dev/wmbus"
WMBUS_MODE = "c1"

async def run_wmbusmeters(callback) -> None:
    """
    Kør wmbusmeters som subprocess og parse JSON output line-by-line.
    Genstarter automatisk ved crash.
    """
    cmd = [
        "wmbusmeters",
        "--format=json",
        f"--logfile=/var/log/wmbusmeters/wmbusmeters.log",
        f"{WMBUS_DEVICE}:{WMBUS_MODE}",
        "CONF_DIR=/etc/wmbusmeters.d/",
    ]

    while True:
        proc = await asyncio.create_subprocess_exec(
            *cmd,
            stdout=asyncio.subprocess.PIPE,
            stderr=asyncio.subprocess.PIPE,
        )
        try:
            async for line in proc.stdout:
                raw = line.decode("utf-8", errors="replace").strip()
                if not raw or not raw.startswith("{"):
                    continue
                reading = parse_wmbus_json(raw)
                if reading:
                    await callback(reading)
        finally:
            proc.kill()
            await proc.wait()
            await asyncio.sleep(5)  # vent inden genstart

---

MQTT publish med parsed data

# gateway/src/wmbus/publisher.py

import msgpack
import zlib
from datetime import datetime, timezone
from .parser import WMBusReading

def reading_to_payload(readings: list[WMBusReading], gateway_id: str) -> bytes:
    """Konvertér liste af WMBusReading til komprimeret MessagePack payload."""
    data = {
        "gw": gateway_id,
        "ts": datetime.now(timezone.utc).isoformat(),
        "r": [
            {
                "id": r.meter_id,
                "drv": r.driver,
                "nm": r.name,
                "st": r.status,
                "ts": r.timestamp.isoformat(),
                "hca": r.current_hca,
                "kwh": r.total_energy_kwh,
                "m3": r.volume_m3,
                "ft": r.flow_temp_c,
                "rt": r.return_temp_c,
                "kw": r.power_kw,
                "bat": r.battery_level_pct,
                "rssi": r.rssi_dbm,
            }
            for r in readings
        ],
    }
    return zlib.compress(msgpack.packb(data, use_bin_type=True))

---

Diagnostik: manglende målere

# Se alle modtagne meter-IDs i realtid:
wmbusmeters --format=json /dev/wmbus:c1 | jq -r '.id + " " + .status + " " + .meter'

# Filtrer kun fejl:
wmbusmeters --format=json /dev/wmbus:c1 | jq 'select(.status != "OK")'

# Tæl modtagne telegrammer pr. ID (10 min):
wmbusmeters --format=json /dev/wmbus:c1 2>/dev/null | \
  jq -r '.id' | sort | uniq -c | sort -rn

# Test specifik AES-nøgle:
wmbusmeters --format=json /dev/wmbus:c1 \
  --field-id=12345678 \
  --field-key=A1B2C3D4E5F6708192A3B4C5D6E7F800

---

Driver-tabel

Driver              Fabrikant        Medie
------------------------------------------------------
techem_hca          Techem           HCA (varmefordeling)
multical302         Kamstrup         Varmemåler
multical403         Kamstrup         Varmemåler
multical21          Kamstrup         Vandmåler
sensostar           Engelmann        Varmemåler + HCA
izar                Diehl            Vandmåler (IZAR R4)
qalcosonic_w1       Axioma           Vandmåler
supercal5           Sontex           Varmemåler
mbus_easy           Zenner           Vandmåler (M-Bus Easy+)

---

Konklusion

wmbusmeters JSON output er konsistent på tværs af drivere — id, status, rssi_dbm og timestamp er altid til stede. Type-specifikke felter som current_hca eller total_energy_consumption_kwh varierer pr. driver. DEC_ERR indikerer forkert AES-nøgle. Python-parsing via json.loads() + dataclass giver typesikker håndtering i gateway-koden.

Se wmbusmeters installation guide eller AES-128 CTR dekryptering guide.