A simple script for receiving water meter readings from a Badger ORION water meter using an Software Defined Radio (SDR).

badger_orion.py

#!/usr/bin/env python3
# A simply tool for receiving water meter readings from a Badger ORION water meter.
# Device radio details can be found at https://fccid.io/GIF2006B
# Requires rtl_433 to be installed. See https://github.com/merbanan/rtl_433

import sys
import json
import subprocess


RTL_PATH = "/usr/local/bin/rtl_433"

nibble_codes = [0x16, 0x0D, 0x0E, 0x0B, 0x1c, 0x19, 0x1A, 0x13, 0x2C, 0x25, 0x26, 0x23, 0x34, 0x31, 0x32, 0x29]
nibble_map = {code: i for i, code in enumerate(nibble_codes)}


def crc16dnp(data):
    poly = 0x13d65
    crc = 0
    for d in data:
        crc = crc << 8 | d
        for i in range(7, -1, -1):
            crc ^= (poly << i) * ((crc >> 16 + i) & 1)
    return crc


def decode_byte(digits):
    # The 4:6 encoding means that one byte is encoded in three hex digits
    raw = int(digits, 16)
    try:
        return (nibble_map[raw >> 6] << 4) + nibble_map[raw & 0x3f]
    except KeyError as kerr:
        raise ValueError("Unknown encoding symbol") from kerr


def decode_record(r):
    timestamp = int(r['time'])
    rssi = r["rssi"]

    data_length = r['rows'][0]['len']
    if data_length < 120:
        raise ValueError("Short packet")

    data_hex = r['rows'][0]['data'][:30]
    packet = bytearray(decode_byte(data_hex[i * 3:(i + 1) * 3]) for i in range(10))

    if crc16dnp(packet) != 0xffff:
        raise ValueError("Bad checksum")

    device_id = int.from_bytes(packet[:4], "little")
    reading = int.from_bytes(packet[4:7], "little")

    return {"timestamp": timestamp, "device_id": device_id, "reading": reading, "rssi": rssi}


protocol_spec = {
    "name": "badger",
    "modulation": "FSK_PCM",
    "short": 10,
    "long": 10,
    "reset": 1000,
    "preamble": "{16}543d"  # 6 bits of sync (010101), 10 bits pre-amble (0000111101)
}
frequency = 916450000
sample_rate = 1000000


command = [
    RTL_PATH,
    "-f", str(frequency),
    "-s", str(sample_rate),
    "-R", "0",
    "-X", ",".join(f"{key}={value}" for key, value in protocol_spec.items()),
    "-F", "json",
    "-M", "time:unix",
    "-M", "level",
]


def main_loop():
    print("Starting radio:", " ".join(command), file=sys.stderr)
    proc = subprocess.Popen(command, stdout=subprocess.PIPE)

    try:
        for line in proc.stdout:
            try:
                raw_record = json.loads(line.strip())
            except json.JSONDecodeError:
                print("Bad JSON", file=sys.stderr)
                continue

            try:
                record = decode_record(raw_record)
            except ValueError as e:
                print("Bad data:", e, file=sys.stderr)
                continue
                
            print(record)
    except KeyboardInterrupt:
        print("Exiting", file=sys.stderr)
    finally:
        proc.kill()
        proc.wait()
        print("Done.", file=sys.stderr)

        
if __name__ == "__main__":
    main_loop()

@jaydeethree Also, since I wrote the script above, I also submitted a PR for rtl_433 to include protocol decoding for the Orion Badger water meters as standard. As such, if you have a recent (post August 2022) version of rtl_433, you don't need to do any protocol decoding and you can capture the messages directly using something like:

rtl_433 -f 916450000 -s 1600000 -R 219

Here -R 219 selects protocol index 219, which is the Badger protocol. You can then play with -F and -M to get the most useful format and metadata for each record. For data logging I would suggest something like -F json -M time:unix:usec:utc.

Thanks so much for the quick and detailed response, I really appreciate it! I'll pick up an SDR dongle and try this out :)