giglez/scripts/experiment_rtl433_augment.py

#!/usr/bin/env python3
"""
Honest experiment: does rtl_433 IQ data help the category classifier?
=====================================================================

The statistical category classifier is starved on minority classes
(train counts ~ Doorbell 39, Weather 12, Security 2). Local Flipper .sub
corpora are byte-identical mirrors, so they add nothing. The rtl_433 project
ships genuinely-independent labelled OOK captures (``.cu8`` IQ), heavy on
exactly the starved classes (weather + PIR/door/smoke security).

This script measures — honestly — whether adding those demodulated captures to
TRAINING helps, without fooling ourselves:

  * Flipper captures are split group-disjoint into train/test (the real
    product distribution; prod input is Flipper .sub RAW).
  * rtl_433 captures are demodulated (``rtl433_iq_demod.demod_file``), split
    group-disjoint by device folder, and used to AUGMENT training only.
  * We compare, on the SAME held-out FLIPPER test set:
        baseline  = train on Flipper-only
        augmented = train on Flipper + rtl_433
    Regression check: do the well-populated Flipper classes
    (Garage/Remote/Fan/Doorbell) get WORSE? If augmentation tanks them, reject.
  * New-capability check: on a held-out rtl_433 test set, can the augmented
    model recognise Weather/Security at all (classes Flipper can't measure —
    too few held-out samples)?

Nothing here touches the serving path; it only decides whether to retrain.
"""

from __future__ import annotations

import random
import warnings
from collections import Counter
from pathlib import Path

import numpy as np

warnings.filterwarnings("ignore")

from scripts.rtl433_iq_demod import demod_file
from scripts.train_category_classifier import (
    collect as collect_flipper, DEFAULT_DATASET, FEATURE_NAMES, extract_features,
)
from src.matcher.timing_analyzer import get_timing_analyzer
from src.matcher.preamble_detector import get_preamble_detector

RTL433_ROOT = (Path(__file__).parent.parent
               / "data/rf_test_datasets/rtl_433_tests/tests")

# Conservative folder -> category map. Only clearly in-taxonomy devices; the
# demod quality-gate additionally drops any FSK/degenerate captures. TPMS,
# energy/utility meters, thermostats and unknowns are intentionally omitted.
RTL433_FOLDER_TO_CATEGORY = {
    # Weather Sensor (the biggest starved-class win)
    **{f: "Weather Sensor" for f in [
        "acurite", "alectov1", "alecto_ws_1200", "ambient_weather", "auriol",
        "bresser_3ch", "bresser_5in1", "bresser_6in1", "bt_rain", "calibeur",
        "companion_wtr001", "conrad_pool_thermometer", "cotech", "cresta_ws688",
        "ecowitt", "EcoWitt-WH40", "esperanza_ews", "Eurochron-EFTH800",
        "eurochron-th", "fineoffset", "froggit_wh1080_Pass14c", "ft004b",
        "hideki", "holman_ws5029", "imagintronix_wh5", "infactory_PV-8796-675",
        "InFactroy-Temp-Humidity-Sensor-T05K-THC", "inkbird", "inovalley-kw9015b",
        "lacrosse", "lacrosse_ltv", "lacrosse_ws7000", "maverick_et-73",
        "maverick_et733", "mebus_te204nl", "misol", "missil_ml0757", "nexus",
        "Opus-XT300", "oregon_scientific", "prologue", "proove", "rubicson",
        "s3318p", "sharp_spc344", "sharp_spc775", "solight_te44", "springfield",
        "TFA_30.3196_TempHumid", "tfa_30_3211_02", "TFA-Drop-30.3233.01",
        "TFA_Marbella", "TFA-Pool-thermometer-30.3160", "TFA-Twin-Plus-30.3049",
        "thermopro-tp11", "thermopro-tp12", "thermopro-tx2", "TS-FT002",
        "ttx201", "tx22-it", "wssensor", "wt0124", "XC-0324", "xc0348",
        "Globaltronics", "PoolWT0122", "rh787t",
    ]},
    # Security Sensor (motion / PIR / smoke / door / alarm)
    **{f: "Security Sensor" for f in [
        "generic_door_sensor", "generic_motion", "honeywell", "honeywell_5816",
        "honeywell_5890PI", "honeywell_activlink", "dsc", "simplisafe",
        "skylink_motion", "smoke_gs558", "cavius", "KIDDE_RF-SM-ACDC",
        "visonic_powercode", "interlogix", "x10_sec", "EV1527-PIR-SGOOWAY",
        "Kerui-D026", "chuango",
    ]},
    # Doorbell
    **{f: "Doorbell" for f in [
        "Byron-BY101", "Byron-BY34", "Byron_db304", "door_bell", "quhwa",
    ]},
    # Remote Control (remotes, outlet switches, key fobs)
    **{f: "Remote Control" for f in [
        "akhan_remote", "dish_remote_6.3", "directv", "EV1527-Universal-Remote",
        "fordremote", "generic-4ch-black-remote-315mhz",
        "generic-4ch-black-remote-433mhz", "generic_remote",
        "ge-coloreffects-remote", "honda_remote", "HT680_remote",
        "hyundai_remote", "intertechno", "nexa_LMST-606", "newkaku", "PT2262",
        "rayrun_rm03", "waveman", "x10", "philips", "Philips-AJ7010",
        "status-rc202", "kangtai", "kedsum", "etekcity", "elro", "blyss",
        "brennstuhl_rcs_2044", "telldus", "smarthome", "silvercrest",
        "uni-com-66125", "valeo",
    ]},
    # Garage Door Opener / gate
    **{f: "Garage Door Opener" for f in [
        "cardin", "LiftMaster_4330E", "secplus", "keeloq", "Microchip-HCS200",
    ]},
    # Fan Controller
    **{f: "Fan Controller" for f in [
        "lucci-air-ceiling-fan-remote", "SQMLtdFan",
    ]},
}


def collect_rtl433(seed: int, per_folder: int = 40):
    """Demodulate rtl_433 captures into (X, y, groups) using the folder map."""
    rng = random.Random(seed)
    ta, pd = get_timing_analyzer(), get_preamble_detector()
    X, y, groups = [], [], []
    skipped = Counter()
    for folder, category in RTL433_FOLDER_TO_CATEGORY.items():
        fpath = RTL433_ROOT / folder
        if not fpath.is_dir():
            skipped["missing_folder"] += 1
            continue
        cu8s = sorted(fpath.rglob("*.cu8"))
        rng.shuffle(cu8s)
        taken = 0
        for c in cu8s:
            if taken >= per_folder:
                break
            pulses, freq = demod_file(c)
            if pulses is None:
                skipped["rejected_demod"] += 1
                continue
            feats = extract_features(pulses, freq, ta, pd)
            if feats is None:
                skipped["no_timing"] += 1
                continue
            X.append(feats)
            y.append(category)
            groups.append(f"rtl433:{folder}")  # group per device folder
            taken += 1
    return np.array(X), np.array(y), np.array(groups), skipped


def per_class_recall(y_true, y_pred, labels):
    out = {}
    for lab in labels:
        idx = y_true == lab
        n = int(idx.sum())
        out[lab] = (int((y_pred[idx] == lab).sum()), n)
    return out


def main():
    from sklearn.model_selection import GroupShuffleSplit
    from sklearn.ensemble import GradientBoostingClassifier
    from sklearn.metrics import balanced_accuracy_score, accuracy_score

    seed = 42
    print("Collecting Flipper RAW ...")
    Xf, yf, gf, ff, _ = collect_flipper(DEFAULT_DATASET, 400, seed)
    print(f"  Flipper: {len(Xf)} samples  {dict(Counter(yf))}")

    print("Demodulating rtl_433 IQ ...")
    Xr, yr, gr, sk = collect_rtl433(seed)
    print(f"  rtl_433: {len(Xr)} samples  {dict(Counter(yr))}  (skipped {dict(sk)})")

    labels = sorted(set(yf) | set(yr))

    # Group-disjoint Flipper train/test (the real product distribution).
    gss = GroupShuffleSplit(n_splits=1, test_size=0.25, random_state=seed)
    ftr, fte = next(gss.split(Xf, yf, gf))
    Xf_tr, Xf_te = Xf[ftr], Xf[fte]
    yf_tr, yf_te = yf[ftr], yf[fte]

    # Group-disjoint rtl_433 train/test.
    rtr, rte = next(GroupShuffleSplit(1, test_size=0.30, random_state=seed)
                    .split(Xr, yr, gr))
    Xr_tr, Xr_te = Xr[rtr], Xr[rte]
    yr_tr, yr_te = yr[rtr], yr[rte]

    def fit(X, y):
        return GradientBoostingClassifier(random_state=seed).fit(X, y)

    base = fit(Xf_tr, yf_tr)
    aug = fit(np.vstack([Xf_tr, Xr_tr]), np.concatenate([yf_tr, yr_tr]))

    print("\n" + "=" * 70)
    print("A) REGRESSION CHECK — held-out FLIPPER test (does augmentation hurt?)")
    print("=" * 70)
    for name, model in [("baseline (Flipper-only)", base),
                        ("augmented (Flipper+rtl433)", aug)]:
        p = model.predict(Xf_te)
        print(f"\n{name}:  bal={balanced_accuracy_score(yf_te, p):.3f} "
              f"top1={accuracy_score(yf_te, p):.3f}")
        for lab, (ok, n) in per_class_recall(yf_te, p, labels).items():
            if n:
                print(f"    {lab:20} {ok:3d}/{n:<3d}")

    print("\n" + "=" * 70)
    print("B) NEW-CAPABILITY — held-out rtl_433 test (classes Flipper can't measure)")
    print("=" * 70)
    for name, model in [("baseline (Flipper-only)", base),
                        ("augmented (Flipper+rtl433)", aug)]:
        p = model.predict(Xr_te)
        print(f"\n{name}:  bal={balanced_accuracy_score(yr_te, p):.3f} "
              f"top1={accuracy_score(yr_te, p):.3f}")
        for lab, (ok, n) in per_class_recall(yr_te, p, labels).items():
            if n:
                print(f"    {lab:20} {ok:3d}/{n:<3d}")


if __name__ == "__main__":
    main()