How Gem & I made a SDR sensor fallback for TX6U (changes its ID on battery replacement)

[TheButterZone]

Unfortunately I could only proof of concept this to myself on Wunderground+APRS CWOP - turns out there is no place, no inexpensive way, that I can put the sensors around my home where they won't delta out of QC with neighboring stations for hours every day. Now they're inside 2 refrigerators being received on their base stations & decoded by rtl_433.

/Users/Username/weewx/weewx-data/weewx.conf:

station_type = SDR (instead of simulator)

[SDR]

    driver = user.sdr
    cmd = /usr/local/bin/rtl_433 -M utc -F json -f 433.92M -s 1024k -g 42.1
    [[sensor_map]]
        outTemp = temperature.0C23.AcuriteTowerPacketV2
        outHumidity = humidity.0C23.AcuriteTowerPacketV2

        extraTemp1 = temperature.112.LaCrosseTXPacket

process_services = weewx.engine.StdConvert, weewx.engine.StdCalibrate, user.temp_fallback.TempFallback, weewx.engine.StdQC, weewx.wxservices.StdWXCalculate

weewx-data/bin/user/sdr.py (from https://github.com/matthewwall/weewx-sdr):

IDENTIFIER = "LaCrosse-TX" (instead of  IDENTIFIER = "LaCrosse TX Sensor" which was out of date)

/Users/Username/weewx/weewx-venv/bin/weewxd-rebind:

#!/usr/bin/env python3
import subprocess
import json
import re
import os

CONF_PATH = "/Users/Username/weewx/weewx-data/weewx.conf"
WEEWXD_PATH = "/Users/Username/weewx/weewx-venv/bin/weewxd"
RTL_CMD = [
    "/usr/local/bin/rtl_433",
    "-M", "utc",
    "-F", "json",
    "-f", "433.92M",
    "-s", "1024k",
    "-g", "42.1"
]

def get_lacrosse_id():
    print("Sniffing SDR stream for active LaCrosse sensor ID...")
    proc = subprocess.Popen(RTL_CMD, stdout=subprocess.PIPE, stderr=subprocess.DEVNULL, text=True)
    try:
        stdout, _ = proc.communicate(timeout=10)
    except subprocess.TimeoutExpired:
        proc.terminate()
        stdout, _ = proc.communicate()

    for line in stdout.splitlines():
        if not line.strip():
            continue
        try:
            data = json.loads(line)
            model = data.get("model", "")
            if "LaCrosse" in model and "id" in data:
                sensor_id = str(data["id"])
                print(f"Detected LaCrosse sensor ID: {sensor_id}")
                return sensor_id
        except json.JSONDecodeError:
            continue
    return None

def update_weewx_conf(new_id):
    if not os.path.exists(CONF_PATH):
        print(f"Config file not found at {CONF_PATH}")
        return

    with open(CONF_PATH, "r") as f:
        content = f.read()

    pattern = r"(extraTemp1\s*=\s*temperature\.)([^.]+)(\.LaCrosseTXPacket)"
    match = re.search(pattern, content)

    if match:
        old_id = match.group(2)
        if old_id == new_id:
            print(f"LaCrosse ID {new_id} is already up to date in weewx.conf.")
            return

        new_content = re.sub(pattern, rf"\g<1>{new_id}\g<3>", content)
        with open(CONF_PATH, "w") as f:
            f.write(new_content)
        print(f"Updated weewx.conf: swapped ID '{old_id}' -> '{new_id}'.")
    else:
        print("Warning: Could not find 'extraTemp1 = temperature.[id].LaCrosseTXPacket' pattern in weewx.conf.")

if __name__ == "__main__":
    sensor_id = get_lacrosse_id()
    if sensor_id:
        update_weewx_conf(sensor_id)
    else:
        print("Warning: LaCrosse sensor ID not detected within timeout; launching with existing configuration.")

    os.execvp(WEEWXD_PATH, [WEEWXD_PATH, CONF_PATH])

Library/LaunchAgents/com.weewx.plist:

<?xml version="1.0" encoding="UTF-8"?>
<!DOCTYPE plist PUBLIC "-//Apple//DTD PLIST 1.0//EN" "http://www.apple.com/DTDs/PropertyList-1.0.dtd">
<plist version="1.0">
<dict>
    <key>Label</key>
    <string>com.weewx</string>
    <key>ProgramArguments</key>
    <array>
        <string>/Users/Username/weewx/weewx-venv/bin/weewxd-rebind</string>
        <string>/Users/Username/weewx/weewx-data/weewx.conf</string>
    </array>
    <key>RunAtLoad</key>
    <true/>
    <key>KeepAlive</key>
    <true/>
    <key>StandardOutPath</key>
    <string>/Users/Username/weewx/weewx-data/weewx.log</string>
    <key>StandardErrorPath</key>
    <string>/Users/Username/weewx/weewx-data/weewx.log</string>
</dict>
</plist>

[TheButterZone]

Whoops, I left the most important one out...

/Users/Username/weewx/weewx-data/bin/user/temp_fallback.py:

import weewx
import weedb
from weewx.engine import StdService

class TempFallback(StdService):
    def __init__(self, engine, config_dict):
        super(TempFallback, self).__init__(engine, config_dict)
        self.bind(weewx.NEW_LOOP_PACKET, self.new_loop_packet)

    def new_loop_packet(self, event):
        packet = event.packet
        # If primary outTemp is missing/None, but extraTemp1 has data, use it as fallback
        if packet.get('outTemp') is None and packet.get('extraTemp1') is not None:
            packet['outTemp'] = packet['extraTemp1']