Merge remote-tracking branch 'church/master'

Trains sklearn RandomForest + GradientBoosting on 14 timing/statistical
features from RAW-only UberGuidoZ captures, with a group-aware split
(GroupShuffleSplit by device sub-folder) so near-duplicate captures never
leak across train/test. The heuristic CategoryRouter is scored on the exact
same held-out files for a fair comparison.

Best model (gradient_boost): 86.8% top-1 accuracy, 62.5% balanced accuracy
vs heuristic 30.0% top-1 / 55.9% routed. Raw accuracy is inflated by the
70%-Garage RAW class balance; balanced accuracy (~2x the heuristic) is the
honest number and meets the Phase 3A 60-75% target. Not yet wired into the
decoder ensemble.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

models/category_classifier_metrics.json

@@ -0,0 +1,47 @@
+{
+  "best_model": "gradient_boost",
+  "n_samples": 803,
+  "n_train": 583,
+  "n_test": 220,
+  "features": [
+    "freq_mhz",
+    "short_pulse_us",
+    "long_pulse_us",
+    "pulse_ratio",
+    "short_gap_us",
+    "long_gap_us",
+    "gap_ratio",
+    "duty_cycle",
+    "pulse_count",
+    "pulse_mean_abs",
+    "pulse_std_abs",
+    "pulse_min_abs",
+    "pulse_max_abs",
+    "preamble_type"
+  ],
+  "class_balance": {
+    "Doorbell": 39,
+    "Garage Door Opener": 565,
+    "Weather Sensor": 12,
+    "Fan Controller": 91,
+    "Security Sensor": 2,
+    "Remote Control": 94
+  },
+  "metrics": {
+    "random_forest": {
+      "accuracy": 0.8272727272727273,
+      "balanced_accuracy": 0.5789748226457088,
+      "macro_f1": 0.5360433604336042
+    },
+    "gradient_boost": {
+      "accuracy": 0.8681818181818182,
+      "balanced_accuracy": 0.6253728690437551,
+      "macro_f1": 0.5340975664713835
+    }
+  },
+  "heuristic_baseline_same_test": {
+    "top1": 0.3,
+    "routed": 0.5590909090909091,
+    "n_test": 220
+  }
+}

scripts/benchmark_realworld.py

@@ -0,0 +1,273 @@
+#!/usr/bin/env python3
+"""
+Real-World Device-Type Validation
+==================================
+
+Unlike ``benchmark_phase0.py`` (which *fabricates* .sub files from the protocol
+DB's own timing parameters and therefore only measures an upper bound), this
+harness runs the **real identification pipeline** against **real community
+Flipper Zero captures** — the UberGuidoZ Sub-GHz corpus — where the folder name
+is the ground-truth device type.
+
+It measures what actually matters for "what kind of device is this?":
+
+  1. Category routing  — does the router put the capture in the right device
+     family? Reported two ways:
+        top1   : router's single best category  == ground truth
+        routed : ground truth ∈ router's allowed_categories (the searched set)
+  2. Coverage          — % of files the pipeline can even act on (RAW-parseable,
+     timing-extractable, ≥1 device match).
+
+Run:
+    python scripts/benchmark_realworld.py                 # default sample
+    python scripts/benchmark_realworld.py --per-category 80 --out /tmp/rw.json
+"""
+
+import argparse
+import json
+import random
+import sys
+import time
+from collections import Counter, defaultdict
+from pathlib import Path
+
+sys.path.insert(0, str(Path(__file__).parent.parent))
+
+from src.parser.sub_parser import parse_sub_file
+from src.matcher.pattern_decoder import get_pattern_decoder
+
+
+# ── Ground truth: UberGuidoZ folder name -> GigLez router category ──────────
+# Only folders with an unambiguous mapping onto a category the router can emit
+# are included. Ambiguous grab-bags (Misc, Jamming, Settings, Pocsag, ...) are
+# intentionally excluded so the denominator stays honest.
+FOLDER_TO_CATEGORY = {
+    "Doorbells":              "Doorbell",
+    "Garages":                "Garage Door Opener",
+    "Gates":                  "Garage Door Opener",
+    "Weather_stations":       "Weather Sensor",
+    "Ceiling_Fans":           "Fan Controller",
+    "Fans":                   "Fan Controller",
+    "Motion_Sensors":         "Security Sensor",
+    "Smoke_Alarm":            "Security Sensor",
+    "Vehicles":               "Remote Control",
+    "Smart_Home_Remotes":     "Remote Control",
+    "Remote_Outlet_Switches": "Remote Control",
+}
+
+DEFAULT_DATASET = (
+    Path(__file__).parent.parent
+    / "data/rf_test_datasets/UberGuidoZ_Flipper/Sub-GHz"
+)
+
+
+def collect_files(dataset_root: Path, per_category: int, seed: int):
+    """Return list of (path, ground_truth_category, folder) sampled per folder."""
+    rng = random.Random(seed)
+    out = []
+    for folder, category in FOLDER_TO_CATEGORY.items():
+        folder_path = dataset_root / folder
+        if not folder_path.is_dir():
+            continue
+        subs = sorted(folder_path.rglob("*.sub"))
+        rng.shuffle(subs)
+        for p in subs[:per_category]:
+            out.append((p, category, folder))
+    return out
+
+
+def evaluate(files, decoder):
+    """Run the routing + decode pipeline over the sampled files."""
+    ta = decoder.timing_analyzer
+    pd = decoder.preamble_detector
+    router = decoder.category_router
+
+    results = []
+    for path, gt_category, folder in files:
+        rec = {
+            "file": str(path),
+            "folder": folder,
+            "ground_truth": gt_category,
+            "status": None,          # ok | no_raw | no_timing | parse_error
+            "file_format": None,
+            "protocol": None,        # Flipper Protocol: field (present on KEY files)
+            "predicted_top1": None,
+            "allowed_categories": [],
+            "routed_hit": False,
+            "top1_hit": False,
+            "n_device_matches": 0,
+            "top_device": None,
+            "top_confidence": None,
+        }
+        try:
+            meta = parse_sub_file(str(path))
+            rec["file_format"] = getattr(meta, "file_format", None)
+            rec["protocol"] = getattr(meta, "protocol", None)
+
+            if not getattr(meta, "has_raw_data", False):
+                # Decoded KEY file: no pulses to time, but a Protocol name is
+                # itself identifying — route by name.
+                if not getattr(meta, "protocol", None):
+                    rec["status"] = "no_raw"   # nothing to go on
+                    results.append(rec)
+                    continue
+                pred = router.route_by_protocol(meta.protocol, meta.frequency)
+                rec["status"] = "ok_key"
+                rec["predicted_top1"] = pred.primary_category
+                rec["allowed_categories"] = list(pred.allowed_categories or [])
+                rec["top1_hit"] = (pred.primary_category == gt_category)
+                rec["routed_hit"] = (
+                    gt_category in rec["allowed_categories"] or pred.use_full_db
+                )
+                matches = decoder.decode(meta)
+                rec["n_device_matches"] = len(matches)
+                if matches:
+                    rec["top_device"] = matches[0].name
+                    rec["top_confidence"] = round(matches[0].confidence, 3)
+                results.append(rec)
+                continue
+
+            pulses = meta.raw_data
+            timing = ta.extract_timing(pulses)
+            short, long = timing.short_pulse_us, timing.long_pulse_us
+            if short == 0 or long == 0:
+                rec["status"] = "no_timing"
+                results.append(rec)
+                continue
+
+            detected = pd.detect(pulses, short, long)
+            ptype = detected.type if detected else "none"
+
+            pred = router.predict(
+                frequency=meta.frequency,
+                short_pulse_us=short,
+                long_pulse_us=long,
+                pulse_count=len(pulses),
+                preamble_type=ptype,
+            )
+            rec["status"] = "ok"
+            rec["predicted_top1"] = pred.primary_category
+            rec["allowed_categories"] = list(pred.allowed_categories or [])
+            rec["top1_hit"] = (pred.primary_category == gt_category)
+            rec["routed_hit"] = (
+                gt_category in rec["allowed_categories"]
+                or (pred.use_full_db)  # full-DB fallback searches everything
+            )
+
+            matches = decoder.decode(meta)
+            rec["n_device_matches"] = len(matches)
+            if matches:
+                rec["top_device"] = matches[0].name
+                rec["top_confidence"] = round(matches[0].confidence, 3)
+
+        except Exception as e:      # noqa: BLE001 — want to bucket, not crash
+            rec["status"] = "parse_error"
+            rec["error"] = str(e)[:200]
+
+        results.append(rec)
+    return results
+
+
+def report(results):
+    total = len(results)
+    status_counts = Counter(r["status"] for r in results)
+    fmt_counts = Counter(r["file_format"] for r in results)
+
+    routable = [r for r in results if r["status"] in ("ok", "ok_key")]
+    n_routable = len(routable)
+    n_raw = sum(1 for r in routable if r["status"] == "ok")
+    n_key = sum(1 for r in routable if r["status"] == "ok_key")
+
+    top1_hits = sum(r["top1_hit"] for r in routable)
+    routed_hits = sum(r["routed_hit"] for r in routable)
+    with_device = sum(1 for r in routable if r["n_device_matches"] > 0)
+
+    print("=" * 74)
+    print("REAL-WORLD DEVICE-TYPE VALIDATION  (UberGuidoZ Sub-GHz corpus)")
+    print("=" * 74)
+    print(f"Files sampled            : {total}")
+    print(f"  status breakdown       : {dict(status_counts)}")
+    print(f"  file formats           : {dict(fmt_counts)}")
+    print(f"Routable (RAW + KEY)     : {n_routable} "
+          f"({n_routable/total:.0%} of sampled)  "
+          f"[RAW timing={n_raw}, KEY protocol={n_key}]")
+    print()
+    if n_routable:
+        print("── Category accuracy (over routable files) ──")
+        print(f"  top-1  (best == truth)      : {top1_hits}/{n_routable} "
+              f"= {top1_hits/n_routable:.1%}")
+        print(f"  routed (truth ∈ allowed set): {routed_hits}/{n_routable} "
+              f"= {routed_hits/n_routable:.1%}")
+        print(f"  device match coverage       : {with_device}/{n_routable} "
+              f"= {with_device/n_routable:.1%}")
+        print()
+
+        # End-to-end (routable AND top-1 correct) over ALL sampled files — the
+        # number a user actually experiences on an arbitrary upload.
+        print("── End-to-end over ALL sampled (incl. unparseable) ──")
+        print(f"  top-1   : {top1_hits}/{total} = {top1_hits/total:.1%}")
+        print(f"  routed  : {routed_hits}/{total} = {routed_hits/total:.1%}")
+        print()
+
+    # Per-category breakdown
+    print("── Per-category (routable only) ──")
+    by_cat = defaultdict(list)
+    for r in routable:
+        by_cat[r["ground_truth"]].append(r)
+    print(f"  {'category':22} {'n':>4} {'top1':>7} {'routed':>7}")
+    for cat in sorted(by_cat):
+        rs = by_cat[cat]
+        n = len(rs)
+        t1 = sum(x["top1_hit"] for x in rs) / n
+        rt = sum(x["routed_hit"] for x in rs) / n
+        print(f"  {cat:22} {n:>4} {t1:>6.0%} {rt:>6.0%}")
+    print()
+
+    # Confusion: where did top-1 send the misses?
+    print("── Top-1 confusion (ground_truth -> predicted, misses only) ──")
+    conf = Counter()
+    for r in routable:
+        if not r["top1_hit"]:
+            conf[(r["ground_truth"], r["predicted_top1"])] += 1
+    for (gt, pred), c in conf.most_common(15):
+        print(f"  {gt:22} -> {str(pred):22} x{c}")
+
+
+def main():
+    ap = argparse.ArgumentParser()
+    ap.add_argument("--dataset", type=Path, default=DEFAULT_DATASET)
+    ap.add_argument("--per-category", type=int, default=50,
+                    help="max files sampled per folder (0 = all)")
+    ap.add_argument("--seed", type=int, default=42)
+    ap.add_argument("--out", type=Path, default=None,
+                    help="write per-file JSON results here")
+    args = ap.parse_args()
+
+    if not args.dataset.is_dir():
+        print(f"Dataset not found: {args.dataset}", file=sys.stderr)
+        sys.exit(1)
+
+    per_cat = args.per_category or 10**9
+    files = collect_files(args.dataset, per_cat, args.seed)
+    if not files:
+        print("No .sub files found under mapped folders.", file=sys.stderr)
+        sys.exit(1)
+
+    print(f"Loading decoder + protocol DB ...")
+    decoder = get_pattern_decoder()
+
+    t0 = time.time()
+    results = evaluate(files, decoder)
+    dt = time.time() - t0
+    print(f"Evaluated {len(files)} files in {dt:.1f}s "
+          f"({dt/len(files)*1000:.0f} ms/file)\n")
+
+    report(results)
+
+    if args.out:
+        args.out.write_text(json.dumps(results, indent=2))
+        print(f"\nPer-file results -> {args.out}")
+
+
+if __name__ == "__main__":
+    main()

scripts/train_category_classifier.py

@@ -0,0 +1,260 @@
+#!/usr/bin/env python3
+"""
+Device-Category Classifier — Phase 3A (statistical ML)
+======================================================
+
+PLAN_TO_PROD Phase 3A: "Statistical Features (Lowest effort, highest ROI)".
+Trains a supervised classifier that predicts a device *category* from the
+timing/statistical features of a RAW Sub-GHz capture — the gap the heuristic
+category router struggles with, because shared line-encoders (Princeton,
+EV1527, Holtek, ...) reuse identical timing across device types.
+
+Why RAW-only:
+    KEY .sub files already carry a decoded ``Protocol:`` name and are handled
+    well by ``CategoryRouter.route_by_protocol``. ML's job is the RAW signals
+    with no protocol match, so we train and evaluate on RAW captures only.
+
+Honesty guardrails:
+    * GROUP-AWARE split — the same device sub-folder (e.g. one physical remote
+      captured many times) never appears in both train and test, so we don't
+      score inflated accuracy off near-duplicate captures.
+    * The heuristic ``CategoryRouter`` is scored on the *exact same* held-out
+      files, so the ML number is directly comparable, not cherry-picked.
+
+Run:
+    python scripts/train_category_classifier.py
+    python scripts/train_category_classifier.py --per-category 400 --out models/
+"""
+
+import argparse
+import json
+import sys
+import time
+from collections import Counter, defaultdict
+from pathlib import Path
+
+import numpy as np
+
+sys.path.insert(0, str(Path(__file__).parent.parent))
+
+from src.parser.sub_parser import parse_sub_file
+from src.matcher.timing_analyzer import get_timing_analyzer
+from src.matcher.preamble_detector import get_preamble_detector
+from src.matcher.category_router import get_category_router
+from scripts.benchmark_realworld import FOLDER_TO_CATEGORY, DEFAULT_DATASET
+
+# Feature order is frozen — inference must build vectors the same way.
+FEATURE_NAMES = [
+    "freq_mhz", "short_pulse_us", "long_pulse_us", "pulse_ratio",
+    "short_gap_us", "long_gap_us", "gap_ratio", "duty_cycle",
+    "pulse_count", "pulse_mean_abs", "pulse_std_abs", "pulse_min_abs",
+    "pulse_max_abs", "preamble_type",
+]
+
+_PREAMBLE_ID = {"none": 0, "long_burst": 1, "alternating": 2,
+                "sync_word": 3, "custom": 4}
+
+
+def extract_features(pulses, frequency, ta, pd):
+    """Timing + statistical feature vector for one RAW capture (or None)."""
+    if not pulses:
+        return None
+    timing = ta.extract_timing(pulses)
+    short, long = timing.short_pulse_us, timing.long_pulse_us
+    if short == 0 or long == 0:
+        return None
+
+    detected = pd.detect(pulses, short, long)
+    ptype = detected.type if detected else "none"
+
+    abs_p = np.abs(np.asarray(pulses, dtype=float))
+    gap_ratio = (timing.long_gap_us / timing.short_gap_us
+                 if timing.short_gap_us > 0 else 0.0)
+
+    return [
+        frequency / 1_000_000,
+        short,
+        long,
+        timing.pulse_ratio,
+        timing.short_gap_us,
+        timing.long_gap_us,
+        gap_ratio,
+        timing.duty_cycle,
+        len(pulses),
+        float(abs_p.mean()),
+        float(abs_p.std()),
+        float(abs_p.min()),
+        float(abs_p.max()),
+        _PREAMBLE_ID.get(ptype, 0),
+    ]
+
+
+def collect(dataset_root: Path, per_category: int, seed: int):
+    """Return (X, y, groups, freqs) over RAW files in the mapped folders.
+
+    groups = device sub-folder path (keeps repeat captures of one device
+    together across the train/test split).
+    """
+    import random
+    rng = random.Random(seed)
+    ta, pd = get_timing_analyzer(), get_preamble_detector()
+
+    X, y, groups, freqs = [], [], [], []
+    skipped = Counter()
+    for folder, category in FOLDER_TO_CATEGORY.items():
+        folder_path = dataset_root / folder
+        if not folder_path.is_dir():
+            continue
+        subs = sorted(folder_path.rglob("*.sub"))
+        rng.shuffle(subs)
+        taken = 0
+        for p in subs:
+            if taken >= per_category:
+                break
+            try:
+                meta = parse_sub_file(str(p))
+            except Exception:
+                skipped["parse_error"] += 1
+                continue
+            if not getattr(meta, "has_raw_data", False):
+                skipped["no_raw"] += 1
+                continue
+            feats = extract_features(meta.raw_data, meta.frequency, ta, pd)
+            if feats is None:
+                skipped["no_timing"] += 1
+                continue
+            X.append(feats)
+            y.append(category)
+            groups.append(str(p.parent))
+            freqs.append(meta.frequency)
+            taken += 1
+    return np.array(X), np.array(y), np.array(groups), np.array(freqs), skipped
+
+
+def heuristic_top1(X_row, freq, router):
+    """Heuristic router's top-1 category for one feature row (same features)."""
+    # X columns: 0 freq_mhz,1 short,2 long,...,8 pulse_count,...,13 preamble
+    inv = {v: k for k, v in _PREAMBLE_ID.items()}
+    pred = router.predict(
+        frequency=int(freq),
+        short_pulse_us=int(X_row[1]),
+        long_pulse_us=int(X_row[2]),
+        pulse_count=int(X_row[8]),
+        preamble_type=inv.get(int(X_row[13]), "none"),
+    )
+    return pred.primary_category, list(pred.allowed_categories or []), pred.use_full_db
+
+
+def main():
+    ap = argparse.ArgumentParser()
+    ap.add_argument("--dataset", type=Path, default=DEFAULT_DATASET)
+    ap.add_argument("--per-category", type=int, default=400,
+                    help="max RAW files sampled per folder")
+    ap.add_argument("--seed", type=int, default=42)
+    ap.add_argument("--test-size", type=float, default=0.25)
+    ap.add_argument("--out", type=Path, default=Path("models"))
+    args = ap.parse_args()
+
+    if not args.dataset.is_dir():
+        print(f"Dataset not found: {args.dataset}", file=sys.stderr)
+        sys.exit(1)
+
+    from sklearn.ensemble import RandomForestClassifier, GradientBoostingClassifier
+    from sklearn.model_selection import GroupShuffleSplit
+    from sklearn.metrics import (accuracy_score, balanced_accuracy_score,
+                                 f1_score, confusion_matrix, classification_report)
+    import joblib
+
+    print("Collecting + extracting features ...")
+    t0 = time.time()
+    X, y, groups, freqs, skipped = collect(args.dataset, args.per_category, args.seed)
+    print(f"  {len(X)} RAW samples in {time.time()-t0:.1f}s  (skipped: {dict(skipped)})")
+    print(f"  class balance: {dict(Counter(y))}")
+    print(f"  distinct device groups: {len(set(groups))}")
+
+    # Group-aware split (no device leakage between train/test)
+    gss = GroupShuffleSplit(n_splits=1, test_size=args.test_size,
+                            random_state=args.seed)
+    train_idx, test_idx = next(gss.split(X, y, groups))
+    Xtr, Xte = X[train_idx], X[test_idx]
+    ytr, yte = y[train_idx], y[test_idx]
+    fte = freqs[test_idx]
+    print(f"  train={len(Xtr)}  test={len(Xte)} "
+          f"(group-disjoint, {len(set(groups[test_idx]))} test groups)")
+
+    models = {
+        "random_forest": RandomForestClassifier(
+            n_estimators=300, class_weight="balanced",
+            random_state=args.seed, n_jobs=-1),
+        "gradient_boost": GradientBoostingClassifier(random_state=args.seed),
+    }
+
+    results = {}
+    best_name, best_bal = None, -1.0
+    for name, clf in models.items():
+        clf.fit(Xtr, ytr)
+        pred = clf.predict(Xte)
+        acc = accuracy_score(yte, pred)
+        bal = balanced_accuracy_score(yte, pred)
+        f1 = f1_score(yte, pred, average="macro")
+        results[name] = (acc, bal, f1)
+        print(f"\n── {name} ──")
+        print(f"  accuracy          : {acc:.1%}")
+        print(f"  balanced accuracy : {bal:.1%}")
+        print(f"  macro F1          : {f1:.1%}")
+        if bal > best_bal:
+            best_name, best_bal, best_clf, best_pred = name, bal, clf, pred
+
+    # ── Heuristic baseline on the SAME test files ──────────────────────────
+    h_top1, h_routed = 0, 0
+    labels = sorted(set(y))
+    for row, freq, truth in zip(Xte, fte, yte):
+        cat, allowed, full = heuristic_top1(row, freq, router=get_category_router())
+        if cat == truth:
+            h_top1 += 1
+        if truth in allowed or full:
+            h_routed += 1
+    n = len(Xte)
+
+    print("\n" + "=" * 66)
+    print(f"BEST MODEL: {best_name}  (balanced acc {best_bal:.1%})")
+    print("=" * 66)
+    print(f"{'':26}{'top-1':>8}{'routed':>9}")
+    print(f"{'heuristic router':26}{h_top1/n:>8.1%}{h_routed/n:>9.1%}")
+    print(f"{'ML classifier (top-1)':26}"
+          f"{accuracy_score(yte, best_pred):>8.1%}{'—':>9}")
+    print("\nPer-category (best model):")
+    print(classification_report(yte, best_pred, zero_division=0))
+    print("Confusion (rows=truth, cols=pred):  labels=", labels)
+    print(confusion_matrix(yte, best_pred, labels=labels))
+    print("\nFeature importances (best model, if available):")
+    if hasattr(best_clf, "feature_importances_"):
+        for nm, imp in sorted(zip(FEATURE_NAMES, best_clf.feature_importances_),
+                              key=lambda t: -t[1]):
+            print(f"  {nm:16} {imp:.3f}")
+
+    # ── Persist ────────────────────────────────────────────────────────────
+    args.out.mkdir(parents=True, exist_ok=True)
+    model_path = args.out / "category_classifier.joblib"
+    joblib.dump({"model": best_clf, "features": FEATURE_NAMES,
+                 "classes": list(best_clf.classes_)}, model_path)
+    meta = {
+        "best_model": best_name,
+        "n_samples": int(len(X)),
+        "n_train": int(len(Xtr)),
+        "n_test": int(len(Xte)),
+        "features": FEATURE_NAMES,
+        "class_balance": {k: int(v) for k, v in Counter(y).items()},
+        "metrics": {k: {"accuracy": v[0], "balanced_accuracy": v[1],
+                        "macro_f1": v[2]} for k, v in results.items()},
+        "heuristic_baseline_same_test": {
+            "top1": h_top1 / n, "routed": h_routed / n, "n_test": n},
+    }
+    (args.out / "category_classifier_metrics.json").write_text(
+        json.dumps(meta, indent=2))
+    print(f"\nSaved model  -> {model_path}")
+    print(f"Saved metrics-> {args.out / 'category_classifier_metrics.json'}")
+
+
+if __name__ == "__main__":
+    main()

src/matcher/category_router.py

@@ -36,6 +36,36 @@ class DeviceCategory:
     UNKNOWN          = "Unknown"
 
 
+# ── Decoded-protocol-name routing (KEY .sub files) ─────────────────────────
+# A KEY file already carries a Flipper `Protocol:` name. For device-specific
+# brands that name alone pins the category. Checked in order; first hit wins.
+# (keywords, category, confidence)
+_PROTOCOL_BRAND_RULES = [
+    (("honeywell", "2gig", "magellan", "scher", "khan", "mastercode"),
+     DeviceCategory.SECURITY_SENSOR, 0.80),
+    (("came", "nice", "faac", "hormann", "somfy", "keeloq", "gate", "megacode",
+      "linear", "chamberlain", "liftmaster", "marantec", "doorhan", "an-motors",
+      "an_motors", "aprimatic", "beninca", "bft", "security+", "secplus", "genie",
+      "clemsa", "ansonic", "sommer", "novoferm", "dooya", "alutech", "elmes",
+      "nero", "hcs101", "starline", "star_line"),
+     DeviceCategory.GARAGE_DOOR, 0.80),
+    (("nexus", "lacrosse", "acurite", "oregon", "ambient", "infactory", "auriol",
+      "bresser", "thermo", "gt-wt", "gt_wt", "wt450", "tx8300", "tx_8300",
+      "wendox", "kedsum"),
+     DeviceCategory.WEATHER_SENSOR, 0.75),
+    (("tpms", "schrader", "pmv"),
+     DeviceCategory.TPMS, 0.80),
+]
+
+# Shared line-encoder silicon reused across remotes/doorbells/fans/gates. The
+# name does NOT determine device type, so we route to the whole family rather
+# than dishonestly narrowing to one category.
+_GENERIC_ENCODERS = (
+    "princeton", "ev1527", "pt2262", "pt2264", "holtek", "ht12", "ht6",
+    "smc5326", "intertechno", "rcswitch", "rc-switch", "x10", "power_smart",
+)
+
+
 @dataclass
 class CategoryPrediction:
     """Result of category routing"""
@@ -136,6 +166,72 @@ class CategoryRouter:
             use_full_db=True,
         )
 
+    def route_by_protocol(
+        self, protocol: Optional[str], frequency: int
+    ) -> CategoryPrediction:
+        """
+        Route a *decoded* KEY-file to a category from its Protocol name.
+
+        KEY .sub files have no RAW pulses to time, but they DO carry a Flipper
+        `Protocol:` name. Device-specific brands (CAME, Nice, KeeLoq, Security+,
+        Honeywell, ...) pin one category. Generic shared encoders (Princeton,
+        EV1527, Holtek, Intertechno, ...) are the same silicon in remotes,
+        doorbells, fans and gates — so we route to the whole family (a broad
+        allowed set) rather than guessing a single type.
+
+        Args:
+            protocol:  Decoded Flipper protocol name (e.g. "CAME", "Princeton")
+            frequency: Carrier frequency in Hz
+
+        Returns:
+            CategoryPrediction
+        """
+        name = (protocol or "").strip().lower()
+        freq_mhz = (frequency or 0) / 1_000_000
+
+        if not name or name in ("raw", "binraw", "unknown"):
+            return CategoryPrediction(
+                primary_category=DeviceCategory.UNKNOWN,
+                confidence=0.0,
+                allowed_categories=[],
+                reasoning="no decoded protocol name",
+                use_full_db=True,
+            )
+
+        # Device-specific brand → single confident category
+        for keywords, category, conf in _PROTOCOL_BRAND_RULES:
+            if any(k in name for k in keywords):
+                return CategoryPrediction(
+                    primary_category=category,
+                    confidence=conf,
+                    allowed_categories=[category],
+                    reasoning=f"decoded protocol '{protocol}' → {category}",
+                )
+
+        # Generic shared encoder → broad family (honest 'routed' signal)
+        if any(k in name for k in _GENERIC_ENCODERS):
+            allowed = [DeviceCategory.REMOTE_CONTROL, DeviceCategory.DOORBELL,
+                       DeviceCategory.FAN_CONTROLLER, DeviceCategory.GARAGE_DOOR]
+            # 315/390 bands skew toward garage/gate; 433 is the free-for-all.
+            primary = (DeviceCategory.GARAGE_DOOR if 300 <= freq_mhz <= 392
+                       else DeviceCategory.REMOTE_CONTROL)
+            return CategoryPrediction(
+                primary_category=primary,
+                confidence=0.45,
+                allowed_categories=allowed,
+                reasoning=(f"generic encoder '{protocol}' @ {freq_mhz:.1f} MHz "
+                           "→ shared-silicon family (remote/doorbell/fan/gate)"),
+            )
+
+        # Unrecognised name — don't guess, search everything.
+        return CategoryPrediction(
+            primary_category=DeviceCategory.UNKNOWN,
+            confidence=0.0,
+            allowed_categories=[],
+            reasoning=f"unrecognised protocol '{protocol}'",
+            use_full_db=True,
+        )
+
     # ── Band-specific helpers ──────────────────────────────────────────────
 
     def _route_300mhz_band(

src/matcher/device_identifier.py

@@ -117,16 +117,17 @@ class DeviceIdentifier:
         t0 = time.time()
         result = IdentificationResult(signal_metadata=signal_data)
 
-        if not signal_data.has_raw_data:
-            # No raw data - cannot identify
+        if not signal_data.has_raw_data and not signal_data.protocol:
+            # No raw data AND no decoded protocol name - cannot identify
             result.method = "none"
             return result
 
-        # Step 1: Pattern-based decoding (heuristic)
+        # Step 1: Pattern-based decoding (heuristic). For KEY files this routes
+        # by the decoded protocol name; for RAW files it times the pulses.
         heuristic_matches = self.pattern_decoder.decode(signal_data)
 
-        # Step 2: Statistical classification (if enabled and sufficient data)
-        if use_statistical and heuristic_matches:
+        # Step 2: Statistical classification (needs RAW pulses; skip on KEY)
+        if use_statistical and heuristic_matches and signal_data.has_raw_data:
             statistical_matches = self._apply_statistical_scoring(
                 signal_data,
                 heuristic_matches

src/matcher/pattern_decoder.py

@@ -103,6 +103,10 @@ class PatternDecoder:
             List of DeviceMatch sorted by confidence (highest first)
         """
         if not metadata.has_raw_data:
+            # KEY files carry no pulses to time, but a decoded Protocol name is
+            # itself identifying — route it instead of returning nothing.
+            if metadata.protocol:
+                return self._decode_from_key(metadata)
             return []
 
         pulses = metadata.raw_data
@@ -125,6 +129,37 @@ class PatternDecoder:
         # Deduplicate and rank by confidence
         return self._rank_matches(matches)
 
+    def _decode_from_key(self, metadata: SignalMetadata) -> List[DeviceMatch]:
+        """
+        Identify a decoded (KEY) .sub file from its Protocol name.
+
+        KEY files have no RAW pulses but carry a decoded `Protocol:` name,
+        frequency and key. We route the protocol name to a device category and
+        emit a match surfacing the real Flipper protocol + routed category, so
+        decoded uploads are no longer silently unidentifiable.
+        """
+        pred = self.category_router.route_by_protocol(
+            metadata.protocol, metadata.frequency
+        )
+        signature = ProtocolSignature(
+            name=metadata.protocol,
+            category=pred.primary_category,
+            frequency=metadata.frequency or 433920000,
+        )
+        details = {
+            "predicted_category": pred.primary_category,
+            "allowed_categories": pred.allowed_categories,
+            "routing_reason": pred.reasoning,
+            "source": "decoded_key",
+            "bit_length": metadata.bit_length,
+        }
+        return [DeviceMatch(
+            protocol=signature,
+            confidence=pred.confidence,
+            match_method="decoded_key",
+            details=details,
+        )]
+
     def _identify_pulse_widths(self, pulses: List[int]) -> Tuple[int, int, int, int]:
         """
         Identify SHORT/LONG pulse and gap durations using robust timing analyzer